Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Enable mypy part 1 (addons and api) #5759

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 29 additions & 19 deletions supervisor/addons/addon.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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

Expand All @@ -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]

Expand Down Expand Up @@ -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]

Expand Down Expand Up @@ -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")
Expand Down Expand Up @@ -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"))
Expand All @@ -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

Expand Down Expand Up @@ -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()

Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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)

Expand Down Expand Up @@ -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
)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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]]:
Expand Down Expand Up @@ -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()
8 changes: 4 additions & 4 deletions supervisor/addons/build.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -23,7 +23,7 @@
from .validate import SCHEMA_BUILD_CONFIG

if TYPE_CHECKING:
from . import AnyAddon
from .manager import AnyAddon


class AddonBuild(FileConfiguration, CoreSysAttributes):
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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()),
Expand Down
3 changes: 2 additions & 1 deletion supervisor/addons/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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]
Expand Down
4 changes: 2 additions & 2 deletions supervisor/addons/model.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)

Expand Down Expand Up @@ -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]

Expand Down
8 changes: 4 additions & 4 deletions supervisor/addons/options.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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):
Expand All @@ -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)
Expand Down
20 changes: 10 additions & 10 deletions supervisor/api/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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
),
]
)

Expand All @@ -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),
Expand All @@ -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),
Expand All @@ -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),
Expand Down
Loading