jeans/supervisor
jeans/supervisor
1
0
Fork 0
mirror of https://github.com/home-assistant/supervisor.git synced 2025-07-24 09:36:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve type hinting for add-ons (#1769)

Browse Source 
* Improve type hinting for add-ons

* Split installed vs AnyAddons extraction
This commit is contained in:
Franck Nijhof 2020-06-03 09:28:02 +02:00 committed by GitHub
parent f13d08d37a
commit 9fe35b4fb5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 49 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
supervisor/addons/addon.py
View File

@ -143,7 +143,7 @@ class Addon(AddonModel):
return self.persist.get(ATTR_BOOT, super().boot)
@boot.setter
def boot(self, value: bool):
def boot(self, value: bool) -> None:
"""Store user boot options."""
self.persist[ATTR_BOOT] = value
@ -153,7 +153,7 @@ class Addon(AddonModel):
return self.persist.get(ATTR_AUTO_UPDATE, super().auto_update)
@auto_update.setter
def auto_update(self, value: bool):
def auto_update(self, value: bool) -> None:
"""Set auto update."""
self.persist[ATTR_AUTO_UPDATE] = value
@ -190,7 +190,7 @@ class Addon(AddonModel):
return self.persist[ATTR_PROTECTED]
@protected.setter
def protected(self, value: bool):
def protected(self, value: bool) -> None:
"""Set add-on in protected mode."""
self.persist[ATTR_PROTECTED] = value
@ -200,7 +200,7 @@ class Addon(AddonModel):
return self.persist.get(ATTR_NETWORK, super().ports)
@ports.setter
def ports(self, value: Optional[Dict[str, Optional[int]]]):
def ports(self, value: Optional[Dict[str, Optional[int]]]) -> None:
"""Set custom ports of add-on."""
if value is None:
self.persist.pop(ATTR_NETWORK, None)
@ -232,6 +232,8 @@ class Addon(AddonModel):
if not url:
return None
webui = RE_WEBUI.match(url)
if not webui:
return None
# extract arguments
t_port = webui.group("t_port")
@ -274,7 +276,7 @@ class Addon(AddonModel):
return self.persist[ATTR_INGRESS_PANEL]
@ingress_panel.setter
def ingress_panel(self, value: bool):
def ingress_panel(self, value: bool) -> None:
"""Return True if the add-on access support ingress."""
self.persist[ATTR_INGRESS_PANEL] = value
@ -310,50 +312,50 @@ class Addon(AddonModel):
return input_data
@audio_input.setter
def audio_input(self, value: Optional[str]):
def audio_input(self, value: Optional[str]) -> None:
"""Set audio input settings."""
self.persist[ATTR_AUDIO_INPUT] = value
@property
def image(self):
def image(self) -> Optional[str]:
"""Return image name of add-on."""
return self.persist.get(ATTR_IMAGE)
@property
def need_build(self):
def need_build(self) -> bool:
"""Return True if this add-on need a local build."""
return ATTR_IMAGE not in self.data
@property
def path_data(self):
def path_data(self) -> Path:
"""Return add-on data path inside Supervisor."""
return Path(self.sys_config.path_addons_data, self.slug)
@property
def path_extern_data(self):
def path_extern_data(self) -> PurePath:
"""Return add-on data path external for Docker."""
return PurePath(self.sys_config.path_extern_addons_data, self.slug)
@property
def path_options(self):
def path_options(self) -> Path:
"""Return path to add-on options."""
return Path(self.path_data, "options.json")
@property
def path_pulse(self):
def path_pulse(self) -> Path:
"""Return path to asound config."""
return Path(self.sys_config.path_tmp, f"{self.slug}_pulse")
@property
def path_extern_pulse(self):
def path_extern_pulse(self) -> Path:
"""Return path to asound config for Docker."""
return Path(self.sys_config.path_extern_tmp, f"{self.slug}_pulse")
def save_persist(self):
def save_persist(self) -> None:
"""Save data of add-on."""
self.sys_addons.data.save_data()
async def write_options(self):
async def write_options(self) -> None:
"""Return True if add-on options is written to data."""
schema = self.schema
options = self.options
@ -378,7 +380,7 @@ class Addon(AddonModel):
raise AddonsError()
async def remove_data(self):
async def remove_data(self) -> None:
"""Remove add-on data."""
if not self.path_data.is_dir():
return
@ -386,7 +388,7 @@ class Addon(AddonModel):
_LOGGER.info("Remove add-on data folder %s", self.path_data)
await remove_data(self.path_data)
def write_pulse(self):
def write_pulse(self) -> None:
"""Write asound config to file and return True on success."""
pulse_config = self.sys_plugins.audio.pulse_client(
input_profile=self.audio_input, output_profile=self.audio_output
@ -518,7 +520,7 @@ class Addon(AddonModel):
except DockerAPIError:
raise AddonsError() from None
async def write_stdin(self, data):
async def write_stdin(self, data) -> None:
"""Write data to add-on stdin.
Return a coroutine.

6
supervisor/addons/model.py
View File

@ -185,7 +185,7 @@ class AddonModel(CoreSysAttributes, ABC):
return self.data[ATTR_VERSION]
@property
def version(self) -> str:
def version(self) -> Optional[str]:
"""Return version of add-on."""
return self.data[ATTR_VERSION]
@ -288,7 +288,7 @@ class AddonModel(CoreSysAttributes, ABC):
return self.data[ATTR_HOST_DBUS]
@property
def devices(self) -> Optional[List[str]]:
def devices(self) -> List[str]:
"""Return devices of add-on."""
return self.data.get(ATTR_DEVICES, [])
@ -452,7 +452,7 @@ class AddonModel(CoreSysAttributes, ABC):
return self.data.get(ATTR_MACHINE, [])
@property
def image(self) -> str:
def image(self) -> Optional[str]:
"""Generate image name from data."""
return self._image(self.data)

57
supervisor/api/addons.py
View File

@ -122,9 +122,7 @@ SCHEMA_SECURITY = vol.Schema({vol.Optional(ATTR_PROTECTED): vol.Boolean()})
class APIAddons(CoreSysAttributes):
"""Handle RESTful API for add-on functions."""
def _extract_addon(
self, request: web.Request, check_installed: bool = True
) -> AnyAddon:
def _extract_addon(self, request: web.Request) -> AnyAddon:
"""Return addon, throw an exception it it doesn't exist."""
addon_slug: str = request.match_info.get("addon")
@ -139,9 +137,12 @@ class APIAddons(CoreSysAttributes):
if not addon:
raise APIError(f"Addon {addon_slug} does not exist")
if check_installed and not addon.is_installed:
raise APIError(f"Addon {addon.slug} is not installed")
return addon
def _extract_addon_installed(self, request: web.Request) -> Addon:
addon = self._extract_addon(request)
if not isinstance(addon, Addon) or not addon.is_installed:
raise APIError("Addon is not installed")
return addon
@api_process
@ -190,7 +191,7 @@ class APIAddons(CoreSysAttributes):
@api_process
async def info(self, request: web.Request) -> Dict[str, Any]:
"""Return add-on information."""
addon: AnyAddon = self._extract_addon(request, check_installed=False)
addon: AnyAddon = self._extract_addon(request)
data = {
ATTR_NAME: addon.name,
@ -257,7 +258,7 @@ class APIAddons(CoreSysAttributes):
ATTR_INGRESS_PANEL: None,
}
if addon.is_installed:
if isinstance(addon, Addon) and addon.is_installed:
data.update(
{
ATTR_STATE: await addon.state(),
@ -279,7 +280,7 @@ class APIAddons(CoreSysAttributes):
@api_process
async def options(self, request: web.Request) -> None:
"""Store user options for add-on."""
addon: AnyAddon = self._extract_addon(request)
addon = self._extract_addon_installed(request)
# Update secrets for validation
await self.sys_secrets.reload()
@ -312,7 +313,7 @@ class APIAddons(CoreSysAttributes):
@api_process
async def security(self, request: web.Request) -> None:
"""Store security options for add-on."""
addon: AnyAddon = self._extract_addon(request)
addon = self._extract_addon_installed(request)
body: Dict[str, Any] = await api_validate(SCHEMA_SECURITY, request)
if ATTR_PROTECTED in body:
@ -324,7 +325,8 @@ class APIAddons(CoreSysAttributes):
@api_process
async def stats(self, request: web.Request) -> Dict[str, Any]:
"""Return resource information."""
addon: AnyAddon = self._extract_addon(request)
addon = self._extract_addon_installed(request)
stats: DockerStats = await addon.stats()
return {
@ -341,62 +343,55 @@ class APIAddons(CoreSysAttributes):
@api_process
def install(self, request: web.Request) -> Awaitable[None]:
"""Install add-on."""
addon: AnyAddon = self._extract_addon(request, check_installed=False)
addon = self._extract_addon(request)
return asyncio.shield(addon.install())
@api_process
def uninstall(self, request: web.Request) -> Awaitable[None]:
"""Uninstall add-on."""
addon: AnyAddon = self._extract_addon(request)
addon = self._extract_addon_installed(request)
return asyncio.shield(addon.uninstall())
@api_process
def start(self, request: web.Request) -> Awaitable[None]:
"""Start add-on."""
addon: AnyAddon = self._extract_addon(request)
addon = self._extract_addon_installed(request)
return asyncio.shield(addon.start())
@api_process
def stop(self, request: web.Request) -> Awaitable[None]:
"""Stop add-on."""
addon: AnyAddon = self._extract_addon(request)
addon = self._extract_addon_installed(request)
return asyncio.shield(addon.stop())
@api_process
def update(self, request: web.Request) -> Awaitable[None]:
"""Update add-on."""
addon: AnyAddon = self._extract_addon(request)
if addon.latest_version == addon.version:
raise APIError("No update available!")
addon: Addon = self._extract_addon_installed(request)
return asyncio.shield(addon.update())
@api_process
def restart(self, request: web.Request) -> Awaitable[None]:
"""Restart add-on."""
addon: AnyAddon = self._extract_addon(request)
addon: Addon = self._extract_addon_installed(request)
return asyncio.shield(addon.restart())
@api_process
def rebuild(self, request: web.Request) -> Awaitable[None]:
"""Rebuild local build add-on."""
addon: AnyAddon = self._extract_addon(request)
if not addon.need_build:
raise APIError("Only local build addons are supported")
addon = self._extract_addon_installed(request)
return asyncio.shield(addon.rebuild())
@api_process_raw(CONTENT_TYPE_BINARY)
def logs(self, request: web.Request) -> Awaitable[bytes]:
"""Return logs from add-on."""
addon: AnyAddon = self._extract_addon(request)
addon = self._extract_addon_installed(request)
return addon.logs()
@api_process_raw(CONTENT_TYPE_PNG)
async def icon(self, request: web.Request) -> bytes:
"""Return icon from add-on."""
addon: AnyAddon = self._extract_addon(request, check_installed=False)
addon = self._extract_addon(request)
if not addon.with_icon:
raise APIError(f"No icon found for add-on {addon.slug}!")
@ -406,7 +401,7 @@ class APIAddons(CoreSysAttributes):
@api_process_raw(CONTENT_TYPE_PNG)
async def logo(self, request: web.Request) -> bytes:
"""Return logo from add-on."""
addon: AnyAddon = self._extract_addon(request, check_installed=False)
addon = self._extract_addon(request)
if not addon.with_logo:
raise APIError(f"No logo found for add-on {addon.slug}!")
@ -416,7 +411,7 @@ class APIAddons(CoreSysAttributes):
@api_process_raw(CONTENT_TYPE_TEXT)
async def changelog(self, request: web.Request) -> str:
"""Return changelog from add-on."""
addon: AnyAddon = self._extract_addon(request, check_installed=False)
addon = self._extract_addon(request)
if not addon.with_changelog:
raise APIError(f"No changelog found for add-on {addon.slug}!")
@ -426,7 +421,7 @@ class APIAddons(CoreSysAttributes):
@api_process_raw(CONTENT_TYPE_TEXT)
async def documentation(self, request: web.Request) -> str:
"""Return documentation from add-on."""
addon: AnyAddon = self._extract_addon(request, check_installed=False)
addon = self._extract_addon(request)
if not addon.with_documentation:
raise APIError(f"No documentation found for add-on {addon.slug}!")
@ -436,7 +431,7 @@ class APIAddons(CoreSysAttributes):
@api_process
async def stdin(self, request: web.Request) -> None:
"""Write to stdin of add-on."""
addon: AnyAddon = self._extract_addon(request)
addon = self._extract_addon_installed(request)
if not addon.with_stdin:
raise APIError(f"STDIN not supported the {addon.slug} add-on")
@ -448,7 +443,7 @@ def _pretty_devices(addon: AnyAddon) -> List[str]:
"""Return a simplified device list."""
dev_list = addon.devices
if not dev_list:
return None
return []
return [row.split(":")[0] for row in dev_list]