1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
"""Main client for supervisor."""
from typing import Self
from aiohttp import ClientSession, ClientTimeout
from .addons import AddonsClient
from .backups import BackupsClient
from .client import _SupervisorClient
from .discovery import DiscoveryClient
from .homeassistant import HomeAssistantClient
from .host import HostClient
from .jobs import JobsClient
from .models.root import AvailableUpdate, AvailableUpdates, RootInfo
from .mounts import MountsClient
from .network import NetworkClient
from .os import OSClient
from .resolution import ResolutionClient
from .store import StoreClient
from .supervisor import SupervisorManagementClient
class SupervisorClient:
"""Main supervisor client for all Supervisor access."""
def __init__(
self,
api_host: str,
token: str,
session: ClientSession | None = None,
) -> None:
"""Initialize client."""
self._client = _SupervisorClient(api_host, token, session)
self._addons = AddonsClient(self._client)
self._os = OSClient(self._client)
self._backups = BackupsClient(self._client)
self._discovery = DiscoveryClient(self._client)
self._jobs = JobsClient(self._client)
self._mounts = MountsClient(self._client)
self._network = NetworkClient(self._client)
self._host = HostClient(self._client)
self._resolution = ResolutionClient(self._client)
self._store = StoreClient(self._client)
self._supervisor = SupervisorManagementClient(self._client)
self._homeassistant = HomeAssistantClient(self._client)
@property
def addons(self) -> AddonsClient:
"""Get addons component client."""
return self._addons
@property
def homeassistant(self) -> HomeAssistantClient:
"""Get Home Assistant component client."""
return self._homeassistant
@property
def os(self) -> OSClient:
"""Get OS component client."""
return self._os
@property
def backups(self) -> BackupsClient:
"""Get backups component client."""
return self._backups
@property
def discovery(self) -> DiscoveryClient:
"""Get discovery component client."""
return self._discovery
@property
def jobs(self) -> JobsClient:
"""Get jobs component client."""
return self._jobs
@property
def mounts(self) -> MountsClient:
"""Get mounts component client."""
return self._mounts
@property
def network(self) -> NetworkClient:
"""Get network component client."""
return self._network
@property
def host(self) -> HostClient:
"""Get host component client."""
return self._host
@property
def resolution(self) -> ResolutionClient:
"""Get resolution center component client."""
return self._resolution
@property
def store(self) -> StoreClient:
"""Get store component client."""
return self._store
@property
def supervisor(self) -> SupervisorManagementClient:
"""Get supervisor component client."""
return self._supervisor
async def info(self) -> RootInfo:
"""Get root info."""
result = await self._client.get("info")
return RootInfo.from_dict(result.data)
async def reload_updates(self) -> None:
"""Reload updates.
Reload main components update information (OS, Supervisor, Core and plug-ins).
"""
await self._client.post("reload_updates", timeout=ClientTimeout(total=300))
async def refresh_updates(self) -> None:
"""Refresh updates.
Discouraged as this endpoint does two things at once. Use the `reload_updates()`
and `store.reload()` instead.
"""
await self._client.post("refresh_updates", timeout=ClientTimeout(total=300))
async def available_updates(self) -> list[AvailableUpdate]:
"""Get available updates."""
result = await self._client.get("available_updates")
return AvailableUpdates.from_dict(result.data).available_updates
async def close(self) -> None:
"""Close open client session."""
await self._client.close()
async def __aenter__(self) -> Self:
"""Async enter, closes session on exit."""
return self
async def __aexit__(self, *_exc_info: object) -> None:
"""Close session."""
await self.close()
|
