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
|
"""Addons client for Supervisor."""
from typing import Any
from .client import _SupervisorComponentClient
from .const import TIMEOUT_60_SECONDS, ResponseType
from .models.addons import (
AddonsConfigValidate,
AddonsList,
AddonsOptions,
AddonsRebuild,
AddonsSecurityOptions,
AddonsStats,
AddonsUninstall,
InstalledAddon,
InstalledAddonComplete,
)
class AddonsClient(_SupervisorComponentClient):
"""Handles installed addon access in Supervisor."""
async def list(self) -> list[InstalledAddon]:
"""Get installed addons."""
result = await self._client.get("addons")
return AddonsList.from_dict(result.data).addons
async def addon_info(self, addon: str) -> InstalledAddonComplete:
"""Get all info for addon."""
result = await self._client.get(f"addons/{addon}/info")
return InstalledAddonComplete.from_dict(result.data)
async def uninstall_addon(
self,
addon: str,
options: AddonsUninstall | None = None,
) -> None:
"""Uninstall an addon."""
await self._client.post(
f"addons/{addon}/uninstall",
json=options.to_dict() if options else None,
timeout=TIMEOUT_60_SECONDS,
)
async def start_addon(self, addon: str) -> None:
"""Start an addon."""
await self._client.post(f"addons/{addon}/start", timeout=TIMEOUT_60_SECONDS)
async def stop_addon(self, addon: str) -> None:
"""Stop an addon."""
await self._client.post(f"addons/{addon}/stop", timeout=TIMEOUT_60_SECONDS)
async def restart_addon(self, addon: str) -> None:
"""Restart an addon."""
await self._client.post(f"addons/{addon}/restart", timeout=None)
async def set_addon_options(self, addon: str, options: AddonsOptions) -> None:
"""Set options for addon."""
await self._client.post(f"addons/{addon}/options", json=options.to_dict())
async def addon_config_validate(
self,
addon: str,
config: dict[str, Any],
) -> AddonsConfigValidate:
"""Validate config for an addon."""
result = await self._client.post(
f"addons/{addon}/options/validate",
response_type=ResponseType.JSON,
json=config,
)
return AddonsConfigValidate.from_dict(result.data)
async def addon_config(self, addon: str) -> dict[str, Any]:
"""Get config for addon."""
result = await self._client.get(f"addons/{addon}/options/config")
return result.data
async def rebuild_addon(
self,
addon: str,
options: AddonsRebuild | None = None,
) -> None:
"""Rebuild an addon (only available for local addons built from source)."""
await self._client.post(
f"addons/{addon}/rebuild",
json=options.to_dict() if options else None,
)
async def write_addon_stdin(self, addon: str, stdin: bytes) -> None:
"""Write to stdin of an addon (if supported by addon)."""
await self._client.post(f"addons/{addon}/stdin", data=stdin)
async def set_addon_security(
self, addon: str, options: AddonsSecurityOptions
) -> None:
"""Set security options for addon."""
await self._client.post(f"addons/{addon}/security", json=options.to_dict())
async def addon_stats(self, addon: str) -> AddonsStats:
"""Get stats for addon."""
result = await self._client.get(f"addons/{addon}/stats")
return AddonsStats.from_dict(result.data)
# Omitted for now - Log endpoints
|
