Enable mypy part 1 (addons and api) (#5759)

* Fix mypy issues in addons

* Fix mypy issues in api

* fix docstring

* Brackets instead of get with default

Author: mdegat01

supervisor/addons/addon.py

@@ -18,7 +18,7 @@
 from typing import Any, Final
 
 import aiohttp
-from awesomeversion import AwesomeVersionCompareException
+from awesomeversion import AwesomeVersion, AwesomeVersionCompareException
 from deepmerge import Merger
 from securetar import AddFileError, atomic_contents_add, secure_path
 import voluptuous as vol
@@ -285,28 +285,28 @@ def is_detached(self) -> bool:
     @property
     def with_icon(self) -> bool:
         """Return True if an icon exists."""
-        if self.is_detached:
+        if self.is_detached or not self.addon_store:
             return super().with_icon
         return self.addon_store.with_icon
 
     @property
     def with_logo(self) -> bool:
         """Return True if a logo exists."""
-        if self.is_detached:
+        if self.is_detached or not self.addon_store:
             return super().with_logo
         return self.addon_store.with_logo
 
     @property
     def with_changelog(self) -> bool:
         """Return True if a changelog exists."""
-        if self.is_detached:
+        if self.is_detached or not self.addon_store:
             return super().with_changelog
         return self.addon_store.with_changelog
 
     @property
     def with_documentation(self) -> bool:
         """Return True if a documentation exists."""
-        if self.is_detached:
+        if self.is_detached or not self.addon_store:
             return super().with_documentation
         return self.addon_store.with_documentation
 
@@ -316,7 +316,7 @@ def available(self) -> bool:
         return self._available(self.data_store)
 
     @property
-    def version(self) -> str | None:
+    def version(self) -> AwesomeVersion:
         """Return installed version."""
         return self.persist[ATTR_VERSION]
 
@@ -464,7 +464,7 @@ def ingress_entry(self) -> str | None:
         return None
 
     @property
-    def latest_version(self) -> str:
+    def latest_version(self) -> AwesomeVersion:
         """Return version of add-on."""
         return self.data_store[ATTR_VERSION]
 
@@ -518,9 +518,8 @@ def ingress_url(self) -> str | None:
     def webui(self) -> str | None:
         """Return URL to webui or None."""
         url = super().webui
-        if not url:
+        if not url or not (webui := RE_WEBUI.match(url)):
             return None
-        webui = RE_WEBUI.match(url)
 
         # extract arguments
         t_port = webui.group("t_port")
@@ -675,10 +674,9 @@ async def save_persist(self) -> None:
 
     async def watchdog_application(self) -> bool:
         """Return True if application is running."""
-        url = super().watchdog
-        if not url:
+        url = self.watchdog_url
+        if not url or not (application := RE_WATCHDOG.match(url)):
             return True
-        application = RE_WATCHDOG.match(url)
 
         # extract arguments
         t_port = int(application.group("t_port"))
@@ -687,8 +685,10 @@ async def watchdog_application(self) -> bool:
         s_suffix = application.group("s_suffix") or ""
 
         # search host port for this docker port
-        if self.host_network:
-            port = self.ports.get(f"{t_port}/tcp", t_port)
+        if self.host_network and self.ports:
+            port = self.ports.get(f"{t_port}/tcp")
+            if port is None:
+                port = t_port
         else:
             port = t_port
 
@@ -777,6 +777,9 @@ async def _check_ingress_port(self):
     )
     async def install(self) -> None:
         """Install and setup this addon."""
+        if not self.addon_store:
+            raise AddonsError("Missing from store, cannot install!")
+
         await self.sys_addons.data.install(self.addon_store)
         await self.load()
 
@@ -880,6 +883,9 @@ async def update(self) -> asyncio.Task | None:
         Returns a Task that completes when addon has state 'started' (see start)
         if it was running. Else nothing is returned.
         """
+        if not self.addon_store:
+            raise AddonsError("Missing from store, cannot update!")
+
         old_image = self.image
         # Cache data to prevent races with other updates to global
         store = self.addon_store.clone()
@@ -936,7 +942,9 @@ async def rebuild(self) -> asyncio.Task | None:
             except DockerError as err:
                 raise AddonsError() from err
 
-            await self.sys_addons.data.update(self.addon_store)
+            if self.addon_store:
+                await self.sys_addons.data.update(self.addon_store)
+
             await self._check_ingress_port()
             _LOGGER.info("Add-on '%s' successfully rebuilt", self.slug)
 
@@ -965,7 +973,9 @@ def write_pulse_config():
             await self.sys_run_in_executor(write_pulse_config)
         except OSError as err:
             if err.errno == errno.EBADMSG:
-                self.sys_resolution.unhealthy = UnhealthyReason.OSERROR_BAD_MESSAGE
+                self.sys_resolution.add_unhealthy_reason(
+                    UnhealthyReason.OSERROR_BAD_MESSAGE
+                )
             _LOGGER.error(
                 "Add-on %s can't write pulse/client.config: %s", self.slug, err
             )
@@ -1324,7 +1334,7 @@ def _addon_backup(
                             arcname="config",
                         )
 
-        wait_for_start: Awaitable[None] | None = None
+        wait_for_start: asyncio.Task | None = None
 
         data = {
             ATTR_USER: self.persist,
@@ -1370,7 +1380,7 @@ async def restore(self, tar_file: tarfile.TarFile) -> asyncio.Task | None:
         Returns a Task that completes when addon has state 'started' (see start)
         if addon is started after restore. Else nothing is returned.
         """
-        wait_for_start: Awaitable[None] | None = None
+        wait_for_start: asyncio.Task | None = None
 
         # Extract backup
         def _extract_tarfile() -> tuple[TemporaryDirectory, dict[str, Any]]:
@@ -1594,6 +1604,6 @@ async def watchdog_container(self, event: DockerContainerStateEvent) -> None:
 
     def refresh_path_cache(self) -> Awaitable[None]:
         """Refresh cache of existing paths."""
-        if self.is_detached:
+        if self.is_detached or not self.addon_store:
             return super().refresh_path_cache()
         return self.addon_store.refresh_path_cache()

supervisor/addons/build.py

@@ -4,7 +4,7 @@
 
 from functools import cached_property
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from awesomeversion import AwesomeVersion
 
@@ -23,7 +23,7 @@
 from .validate import SCHEMA_BUILD_CONFIG
 
 if TYPE_CHECKING:
-    from . import AnyAddon
+    from .manager import AnyAddon
 
 
 class AddonBuild(FileConfiguration, CoreSysAttributes):
@@ -63,7 +63,7 @@ async def save_data(self):
     @cached_property
     def arch(self) -> str:
         """Return arch of the add-on."""
-        return self.sys_arch.match(self.addon.arch)
+        return self.sys_arch.match([self.addon.arch])
 
     @property
     def base_image(self) -> str:
@@ -126,7 +126,7 @@ def get_docker_args(self, version: AwesomeVersion, image: str | None = None):
 
         Must be run in executor.
         """
-        args = {
+        args: dict[str, Any] = {
             "path": str(self.addon.path_location),
             "tag": f"{image or self.addon.image}:{version!s}",
             "dockerfile": str(self.get_dockerfile()),

supervisor/addons/manager.py

@@ -194,6 +194,7 @@ async def install(self, slug: str) -> None:
 
         _LOGGER.info("Add-on '%s' successfully installed", slug)
 
+    @Job(name="addon_manager_uninstall")
     async def uninstall(self, slug: str, *, remove_config: bool = False) -> None:
         """Remove an add-on."""
         if slug not in self.local:
@@ -313,7 +314,7 @@ async def restore(
         if slug not in self.local:
             _LOGGER.debug("Add-on %s is not local available for restore", slug)
             addon = Addon(self.coresys, slug)
-            had_ingress = False
+            had_ingress: bool | None = False
         else:
             _LOGGER.debug("Add-on %s is local available for restore", slug)
             addon = self.local[slug]

supervisor/addons/model.py

@@ -294,7 +294,7 @@ def webui(self) -> str | None:
         return self.data.get(ATTR_WEBUI)
 
     @property
-    def watchdog(self) -> str | None:
+    def watchdog_url(self) -> str | None:
         """Return URL to for watchdog or None."""
         return self.data.get(ATTR_WATCHDOG)
 
@@ -606,7 +606,7 @@ def schema(self) -> AddonOptions:
         return AddonOptions(self.coresys, raw_schema, self.name, self.slug)
 
     @property
-    def schema_ui(self) -> list[dict[any, any]] | None:
+    def schema_ui(self) -> list[dict[Any, Any]] | None:
         """Create a UI schema for add-on options."""
         raw_schema = self.data[ATTR_SCHEMA]
 

supervisor/addons/options.py

@@ -137,7 +137,7 @@ def _single_validate(self, typ: str, value: Any, key: str):
             ) from None
 
         # prepare range
-        range_args = {}
+        range_args: dict[str, Any] = {}
         for group_name in _SCHEMA_LENGTH_PARTS:
             group_value = match.group(group_name)
             if group_value:
@@ -390,14 +390,14 @@ def _nested_ui_dict(
         multiple: bool = False,
     ) -> None:
         """UI nested dict items."""
-        ui_node = {
+        ui_node: dict[str, Any] = {
             "name": key,
             "type": "schema",
             "optional": True,
             "multiple": multiple,
         }
 
-        nested_schema = []
+        nested_schema: list[dict[str, Any]] = []
         for c_key, c_value in option_dict.items():
             # Nested?
             if isinstance(c_value, list):
@@ -413,7 +413,7 @@ def _create_device_filter(str_filter: str) -> dict[str, Any]:
     """Generate device Filter."""
     raw_filter = dict(value.split("=") for value in str_filter.split(";"))
 
-    clean_filter = {}
+    clean_filter: dict[str, Any] = {}
     for key, value in raw_filter.items():
         if key == "subsystem":
             clean_filter[key] = UdevSubsystem(value)

supervisor/api/__init__.py

@@ -6,7 +6,7 @@
 from pathlib import Path
 from typing import Any
 
-from aiohttp import web
+from aiohttp import hdrs, web
 
 from ..const import AddonState
 from ..coresys import CoreSys, CoreSysAttributes
@@ -82,15 +82,13 @@ def __init__(self, coresys: CoreSys):
         self._site: web.TCPSite | None = None
 
         # share single host API handler for reuse in logging endpoints
-        self._api_host: APIHost | None = None
+        self._api_host: APIHost = APIHost()
+        self._api_host.coresys = coresys
 
     async def load(self) -> None:
         """Register REST API Calls."""
         static_resource_configs: list[StaticResourceConfig] = []
 
-        self._api_host = APIHost()
-        self._api_host.coresys = self.coresys
-
         self._register_addons()
         self._register_audio()
         self._register_auth()
@@ -526,7 +524,7 @@ def _register_addons(self) -> None:
 
         self.webapp.add_routes(
             [
-                web.get("/addons", api_addons.list),
+                web.get("/addons", api_addons.list_addons),
                 web.post("/addons/{addon}/uninstall", api_addons.uninstall),
                 web.post("/addons/{addon}/start", api_addons.start),
                 web.post("/addons/{addon}/stop", api_addons.stop),
@@ -594,7 +592,9 @@ def _register_ingress(self) -> None:
                 web.post("/ingress/session", api_ingress.create_session),
                 web.post("/ingress/validate_session", api_ingress.validate_session),
                 web.get("/ingress/panels", api_ingress.panels),
-                web.view("/ingress/{token}/{path:.*}", api_ingress.handler),
+                web.route(
+                    hdrs.METH_ANY, "/ingress/{token}/{path:.*}", api_ingress.handler
+                ),
             ]
         )
 
@@ -605,7 +605,7 @@ def _register_backups(self) -> None:
 
         self.webapp.add_routes(
             [
-                web.get("/backups", api_backups.list),
+                web.get("/backups", api_backups.list_backups),
                 web.get("/backups/info", api_backups.info),
                 web.post("/backups/options", api_backups.options),
                 web.post("/backups/reload", api_backups.reload),
@@ -632,7 +632,7 @@ def _register_services(self) -> None:
 
         self.webapp.add_routes(
             [
-                web.get("/services", api_services.list),
+                web.get("/services", api_services.list_services),
                 web.get("/services/{service}", api_services.get_service),
                 web.post("/services/{service}", api_services.set_service),
                 web.delete("/services/{service}", api_services.del_service),
@@ -646,7 +646,7 @@ def _register_discovery(self) -> None:
 
         self.webapp.add_routes(
             [
-                web.get("/discovery", api_discovery.list),
+                web.get("/discovery", api_discovery.list_discovery),
                 web.get("/discovery/{uuid}", api_discovery.get_discovery),
                 web.delete("/discovery/{uuid}", api_discovery.del_discovery),
                 web.post("/discovery", api_discovery.set_discovery),