From dafc2cfec27777c3f4c68a2eb4c5cd5d58fda4c2 Mon Sep 17 00:00:00 2001 From: "dependabot[bot]" <49699333+dependabot[bot]@users.noreply.github.com> Date: Tue, 21 Sep 2021 18:22:56 +0200 Subject: [PATCH] Bump pyupgrade from 2.26.0 to 2.26.0.post1 (#3131) * Bump pyupgrade from 2.26.0 to 2.26.0.post1 Bumps [pyupgrade](https://github.com/asottile/pyupgrade) from 2.26.0 to 2.26.0.post1. - [Release notes](https://github.com/asottile/pyupgrade/releases) - [Commits](https://github.com/asottile/pyupgrade/commits) --- updated-dependencies: - dependency-name: pyupgrade dependency-type: direct:production update-type: version-update:semver-patch ... Signed-off-by: dependabot[bot] * Update .pre-commit-config.yaml * Fixes Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Pascal Vizeli --- .pre-commit-config.yaml | 4 +- requirements_tests.txt | 2 +- supervisor/addons/__init__.py | 18 +++--- supervisor/addons/addon.py | 16 ++--- supervisor/addons/build.py | 6 +- supervisor/addons/data.py | 4 +- supervisor/addons/model.py | 32 +++++----- supervisor/addons/options.py | 34 +++++------ supervisor/addons/validate.py | 6 +- supervisor/api/addons.py | 12 ++-- supervisor/api/audio.py | 6 +- supervisor/api/auth.py | 7 +-- supervisor/api/cli.py | 6 +- supervisor/api/dns.py | 6 +- supervisor/api/docker.py | 4 +- supervisor/api/hardware.py | 8 +-- supervisor/api/homeassistant.py | 6 +- supervisor/api/info.py | 4 +- supervisor/api/ingress.py | 12 ++-- supervisor/api/jobs.py | 4 +- supervisor/api/multicast.py | 6 +- supervisor/api/network.py | 18 +++--- supervisor/api/observer.py | 6 +- supervisor/api/os.py | 6 +- supervisor/api/resolution.py | 4 +- supervisor/api/security.py | 4 +- supervisor/api/store.py | 16 ++--- supervisor/api/supervisor.py | 6 +- supervisor/api/utils.py | 12 ++-- supervisor/arch.py | 9 ++- supervisor/auth.py | 4 +- supervisor/backups/backup.py | 20 +++---- supervisor/backups/manager.py | 4 +- supervisor/bus.py | 4 +- supervisor/config.py | 4 +- supervisor/core.py | 4 +- supervisor/coresys.py | 60 +++++++++---------- supervisor/dbus/agent/__init__.py | 4 +- supervisor/dbus/agent/apparmor.py | 4 +- supervisor/dbus/agent/datadisk.py | 4 +- supervisor/dbus/hostname.py | 4 +- supervisor/dbus/interface.py | 4 +- supervisor/dbus/manager.py | 3 +- supervisor/dbus/network/__init__.py | 8 +-- supervisor/dbus/network/configuration.py | 10 ++-- supervisor/dbus/network/dns.py | 6 +- supervisor/dbus/payloads/generate.py | 4 +- supervisor/dbus/timedate.py | 4 +- supervisor/discovery/__init__.py | 14 ++--- supervisor/docker/__init__.py | 8 +-- supervisor/docker/addon.py | 34 +++++------ supervisor/docker/audio.py | 10 ++-- supervisor/docker/homeassistant.py | 6 +- supervisor/docker/interface.py | 14 ++--- supervisor/docker/multicast.py | 3 +- supervisor/docker/network.py | 8 +-- supervisor/hardware/data.py | 11 ++-- supervisor/hardware/manager.py | 8 +-- supervisor/hardware/policy.py | 11 ++-- supervisor/homeassistant/api.py | 10 ++-- supervisor/homeassistant/secrets.py | 4 +- supervisor/homeassistant/websocket.py | 12 ++-- supervisor/host/manager.py | 5 +- supervisor/host/network.py | 41 +++++++------ supervisor/host/sound.py | 24 ++++---- supervisor/ingress.py | 10 ++-- supervisor/jobs/__init__.py | 10 ++-- supervisor/jobs/decorator.py | 6 +- supervisor/misc/scheduler.py | 6 +- supervisor/os/data_disk.py | 6 +- supervisor/plugins/dns.py | 24 ++++---- supervisor/resolution/check.py | 8 +-- supervisor/resolution/checks/addon_pwned.py | 4 +- supervisor/resolution/checks/base.py | 4 +- supervisor/resolution/checks/core_security.py | 4 +- supervisor/resolution/checks/core_trust.py | 4 +- supervisor/resolution/checks/free_space.py | 6 +- supervisor/resolution/checks/plugin_trust.py | 4 +- .../resolution/checks/supervisor_trust.py | 4 +- supervisor/resolution/evaluate.py | 7 +-- supervisor/resolution/evaluations/apparmor.py | 3 +- supervisor/resolution/evaluations/base.py | 3 +- .../resolution/evaluations/container.py | 6 +- .../resolution/evaluations/content_trust.py | 3 +- supervisor/resolution/evaluations/dbus.py | 3 +- .../evaluations/docker_configuration.py | 3 +- .../resolution/evaluations/docker_version.py | 3 +- .../resolution/evaluations/job_conditions.py | 3 +- supervisor/resolution/evaluations/lxc.py | 3 +- .../resolution/evaluations/network_manager.py | 3 +- .../evaluations/operating_system.py | 3 +- .../resolution/evaluations/privileged.py | 3 +- .../resolution/evaluations/source_mods.py | 3 +- supervisor/resolution/evaluations/systemd.py | 3 +- supervisor/resolution/fixup.py | 3 +- supervisor/resolution/fixups/base.py | 4 +- .../resolution/fixups/clear_full_backup.py | 4 +- .../resolution/fixups/store_execute_reload.py | 4 +- .../resolution/fixups/store_execute_remove.py | 4 +- .../resolution/fixups/store_execute_reset.py | 4 +- supervisor/resolution/module.py | 22 +++---- supervisor/resolution/validate.py | 3 +- supervisor/services/__init__.py | 6 +- supervisor/services/data.py | 6 +- supervisor/services/interface.py | 12 ++-- supervisor/services/modules/mqtt.py | 8 +-- supervisor/services/modules/mysql.py | 8 +-- supervisor/store/__init__.py | 5 +- supervisor/store/data.py | 10 ++-- supervisor/store/git.py | 4 +- supervisor/store/repository.py | 4 +- supervisor/utils/__init__.py | 4 +- supervisor/utils/codenotary.py | 4 +- supervisor/utils/common.py | 8 +-- supervisor/utils/dt.py | 4 +- supervisor/utils/gdbus.py | 26 ++++---- supervisor/utils/pwned.py | 3 +- supervisor/utils/tar.py | 6 +- tests/docker/test_addon.py | 15 +++-- tests/hardware/test_module.py | 20 +++---- 120 files changed, 491 insertions(+), 524 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 3785ca79b..179761d64 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -28,7 +28,7 @@ repos: hooks: - id: isort - repo: https://github.com/asottile/pyupgrade - rev: v2.6.2 + rev: v2.26.0 hooks: - id: pyupgrade - args: [--py37-plus] + args: [--py39-plus] diff --git a/requirements_tests.txt b/requirements_tests.txt index 5dd2208d9..948ffe898 100644 --- a/requirements_tests.txt +++ b/requirements_tests.txt @@ -11,4 +11,4 @@ pytest-asyncio==0.12.0 # NB!: Versions over 0.12.0 breaks pytest-aiohttp (https: pytest-cov==2.12.1 pytest-timeout==1.4.2 pytest==6.2.5 -pyupgrade==2.26.0 +pyupgrade==2.26.0.post1 diff --git a/supervisor/addons/__init__.py b/supervisor/addons/__init__.py index c7eadb588..e16b8e52f 100644 --- a/supervisor/addons/__init__.py +++ b/supervisor/addons/__init__.py @@ -3,7 +3,7 @@ import asyncio from contextlib import suppress import logging import tarfile -from typing import Dict, List, Optional, Union +from typing import Optional, Union from ..const import AddonBoot, AddonStartup, AddonState from ..coresys import CoreSys, CoreSysAttributes @@ -38,17 +38,17 @@ class AddonManager(CoreSysAttributes): """Initialize Docker base wrapper.""" self.coresys: CoreSys = coresys self.data: AddonsData = AddonsData(coresys) - self.local: Dict[str, Addon] = {} - self.store: Dict[str, AddonStore] = {} + self.local: dict[str, Addon] = {} + self.store: dict[str, AddonStore] = {} @property - def all(self) -> List[AnyAddon]: + def all(self) -> list[AnyAddon]: """Return a list of all add-ons.""" - addons: Dict[str, AnyAddon] = {**self.store, **self.local} + addons: dict[str, AnyAddon] = {**self.store, **self.local} return list(addons.values()) @property - def installed(self) -> List[Addon]: + def installed(self) -> list[Addon]: """Return a list of all installed add-ons.""" return list(self.local.values()) @@ -89,7 +89,7 @@ class AddonManager(CoreSysAttributes): async def boot(self, stage: AddonStartup) -> None: """Boot add-ons with mode auto.""" - tasks: List[Addon] = [] + tasks: list[Addon] = [] for addon in self.installed: if addon.boot != AddonBoot.AUTO or addon.startup != stage: continue @@ -123,7 +123,7 @@ class AddonManager(CoreSysAttributes): async def shutdown(self, stage: AddonStartup) -> None: """Shutdown addons.""" - tasks: List[Addon] = [] + tasks: list[Addon] = [] for addon in self.installed: if addon.state != AddonState.STARTED or addon.startup != stage: continue @@ -371,7 +371,7 @@ class AddonManager(CoreSysAttributes): @Job(conditions=[JobCondition.FREE_SPACE, JobCondition.INTERNET_HOST]) async def repair(self) -> None: """Repair local add-ons.""" - needs_repair: List[Addon] = [] + needs_repair: list[Addon] = [] # Evaluate Add-ons to repair for addon in self.installed: diff --git a/supervisor/addons/addon.py b/supervisor/addons/addon.py index 918dc9c09..d046e8fda 100644 --- a/supervisor/addons/addon.py +++ b/supervisor/addons/addon.py @@ -10,7 +10,7 @@ import secrets import shutil import tarfile from tempfile import TemporaryDirectory -from typing import Any, Awaitable, Dict, List, Optional, Set +from typing import Any, Awaitable, Optional import aiohttp import voluptuous as vol @@ -187,17 +187,17 @@ class Addon(AddonModel): return self.version != self.latest_version @property - def dns(self) -> List[str]: + def dns(self) -> list[str]: """Return list of DNS name for that add-on.""" return [f"{self.hostname}.{DNS_SUFFIX}"] @property - def options(self) -> Dict[str, Any]: + def options(self) -> dict[str, Any]: """Return options with local changes.""" return {**self.data[ATTR_OPTIONS], **self.persist[ATTR_OPTIONS]} @options.setter - def options(self, value: Optional[Dict[str, Any]]) -> None: + def options(self, value: Optional[dict[str, Any]]) -> None: """Store user add-on options.""" self.persist[ATTR_OPTIONS] = {} if value is None else deepcopy(value) @@ -274,12 +274,12 @@ class Addon(AddonModel): self.persist[ATTR_PROTECTED] = value @property - def ports(self) -> Optional[Dict[str, Optional[int]]]: + def ports(self) -> Optional[dict[str, Optional[int]]]: """Return ports of add-on.""" return self.persist.get(ATTR_NETWORK, super().ports) @ports.setter - def ports(self, value: Optional[Dict[str, Optional[int]]]) -> None: + 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) @@ -425,7 +425,7 @@ class Addon(AddonModel): return Path(self.sys_config.path_extern_tmp, f"{self.slug}_pulse") @property - def devices(self) -> Set[Device]: + def devices(self) -> set[Device]: """Extract devices from add-on options.""" options_schema = self.schema with suppress(vol.Invalid): @@ -434,7 +434,7 @@ class Addon(AddonModel): return options_schema.devices @property - def pwned(self) -> Set[str]: + def pwned(self) -> set[str]: """Extract pwned data for add-on options.""" options_schema = self.schema with suppress(vol.Invalid): diff --git a/supervisor/addons/build.py b/supervisor/addons/build.py index 58a66885e..a011fc092 100644 --- a/supervisor/addons/build.py +++ b/supervisor/addons/build.py @@ -2,7 +2,7 @@ from __future__ import annotations from pathlib import Path -from typing import TYPE_CHECKING, Dict +from typing import TYPE_CHECKING from awesomeversion import AwesomeVersion @@ -60,12 +60,12 @@ class AddonBuild(FileConfiguration, CoreSysAttributes): return self._data[ATTR_SQUASH] @property - def additional_args(self) -> Dict[str, str]: + def additional_args(self) -> dict[str, str]: """Return additional Docker build arguments.""" return self._data[ATTR_ARGS] @property - def additional_labels(self) -> Dict[str, str]: + def additional_labels(self) -> dict[str, str]: """Return additional Docker labels.""" return self._data[ATTR_LABELS] diff --git a/supervisor/addons/data.py b/supervisor/addons/data.py index 077e77cb9..47c7002bc 100644 --- a/supervisor/addons/data.py +++ b/supervisor/addons/data.py @@ -1,6 +1,6 @@ """Init file for Supervisor add-on data.""" from copy import deepcopy -from typing import Any, Dict +from typing import Any from ..const import ( ATTR_IMAGE, @@ -16,7 +16,7 @@ from ..utils.common import FileConfiguration from .addon import Addon from .validate import SCHEMA_ADDONS_FILE -Config = Dict[str, Any] +Config = dict[str, Any] class AddonsData(FileConfiguration, CoreSysAttributes): diff --git a/supervisor/addons/model.py b/supervisor/addons/model.py index 8ce54b7dd..945b0261c 100644 --- a/supervisor/addons/model.py +++ b/supervisor/addons/model.py @@ -1,7 +1,7 @@ """Init file for Supervisor add-ons.""" from abc import ABC, abstractmethod from pathlib import Path -from typing import Any, Awaitable, Dict, List, Optional +from typing import Any, Awaitable, Optional from awesomeversion import AwesomeVersion, AwesomeVersionException @@ -83,7 +83,7 @@ from .const import ATTR_BACKUP from .options import AddonOptions, UiOptions from .validate import RE_SERVICE, RE_VOLUME -Data = Dict[str, Any] +Data = dict[str, Any] class AddonModel(CoreSysAttributes, ABC): @@ -115,7 +115,7 @@ class AddonModel(CoreSysAttributes, ABC): return self._available(self.data) @property - def options(self) -> Dict[str, Any]: + def options(self) -> dict[str, Any]: """Return options with local changes.""" return self.data[ATTR_OPTIONS] @@ -140,7 +140,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.slug.replace("_", "-") @property - def dns(self) -> List[str]: + def dns(self) -> list[str]: """Return list of DNS name for that add-on.""" return [] @@ -228,7 +228,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_STAGE] @property - def services_role(self) -> Dict[str, str]: + def services_role(self) -> dict[str, str]: """Return dict of services with rights.""" services_list = self.data.get(ATTR_SERVICES, []) @@ -241,17 +241,17 @@ class AddonModel(CoreSysAttributes, ABC): return services @property - def discovery(self) -> List[str]: + def discovery(self) -> list[str]: """Return list of discoverable components/platforms.""" return self.data.get(ATTR_DISCOVERY, []) @property - def ports_description(self) -> Optional[Dict[str, str]]: + def ports_description(self) -> Optional[dict[str, str]]: """Return descriptions of ports.""" return self.data.get(ATTR_PORTS_DESCRIPTION) @property - def ports(self) -> Optional[Dict[str, Optional[int]]]: + def ports(self) -> Optional[dict[str, Optional[int]]]: """Return ports of add-on.""" return self.data.get(ATTR_PORTS) @@ -311,17 +311,17 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_HOST_DBUS] @property - def static_devices(self) -> List[Path]: + def static_devices(self) -> list[Path]: """Return static devices of add-on.""" return [Path(node) for node in self.data.get(ATTR_DEVICES, [])] @property - def environment(self) -> Optional[Dict[str, str]]: + def environment(self) -> Optional[dict[str, str]]: """Return environment of add-on.""" return self.data.get(ATTR_ENVIRONMENT) @property - def privileged(self) -> List[Capabilities]: + def privileged(self) -> list[Capabilities]: """Return list of privilege.""" return self.data.get(ATTR_PRIVILEGED, []) @@ -360,7 +360,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_HASSIO_ROLE] @property - def backup_exclude(self) -> List[str]: + def backup_exclude(self) -> list[str]: """Return Exclude list for backup.""" return self.data.get(ATTR_BACKUP_EXCLUDE, []) @@ -495,12 +495,12 @@ class AddonModel(CoreSysAttributes, ABC): return self.path_documentation.exists() @property - def supported_arch(self) -> List[str]: + def supported_arch(self) -> list[str]: """Return list of supported arch.""" return self.data[ATTR_ARCH] @property - def supported_machine(self) -> List[str]: + def supported_machine(self) -> list[str]: """Return list of supported machine.""" return self.data.get(ATTR_MACHINE, []) @@ -515,7 +515,7 @@ class AddonModel(CoreSysAttributes, ABC): return ATTR_IMAGE not in self.data @property - def map_volumes(self) -> Dict[str, str]: + def map_volumes(self) -> dict[str, str]: """Return a dict of {volume: policy} from add-on.""" volumes = {} for volume in self.data[ATTR_MAP]: @@ -566,7 +566,7 @@ class AddonModel(CoreSysAttributes, ABC): return AddonOptions(self.coresys, raw_schema, self.name, self.slug) @property - def schema_ui(self) -> Optional[List[Dict[any, any]]]: + def schema_ui(self) -> Optional[list[dict[any, any]]]: """Create a UI schema for add-on options.""" raw_schema = self.data[ATTR_SCHEMA] diff --git a/supervisor/addons/options.py b/supervisor/addons/options.py index 6ffedae3f..aa6b7bda1 100644 --- a/supervisor/addons/options.py +++ b/supervisor/addons/options.py @@ -3,7 +3,7 @@ import hashlib import logging from pathlib import Path import re -from typing import Any, Dict, List, Set, Union +from typing import Any, Union import voluptuous as vol @@ -59,13 +59,13 @@ class AddonOptions(CoreSysAttributes): """Validate Add-ons Options.""" def __init__( - self, coresys: CoreSys, raw_schema: Dict[str, Any], name: str, slug: str + self, coresys: CoreSys, raw_schema: dict[str, Any], name: str, slug: str ): """Validate schema.""" self.coresys: CoreSys = coresys - self.raw_schema: Dict[str, Any] = raw_schema - self.devices: Set[Device] = set() - self.pwned: Set[str] = set() + self.raw_schema: dict[str, Any] = raw_schema + self.devices: set[Device] = set() + self.pwned: set[str] = set() self._name = name self._slug = slug @@ -187,7 +187,7 @@ class AddonOptions(CoreSysAttributes): f"Fatal error for option '{key}' with type '{typ}' in {self._name} ({self._slug})" ) from None - def _nested_validate_list(self, typ: Any, data_list: List[Any], key: str): + def _nested_validate_list(self, typ: Any, data_list: list[Any], key: str): """Validate nested items.""" options = [] @@ -209,7 +209,7 @@ class AddonOptions(CoreSysAttributes): return options def _nested_validate_dict( - self, typ: Dict[Any, Any], data_dict: Dict[Any, Any], key: str + self, typ: dict[Any, Any], data_dict: dict[Any, Any], key: str ): """Validate nested items.""" options = {} @@ -241,7 +241,7 @@ class AddonOptions(CoreSysAttributes): return options def _check_missing_options( - self, origin: Dict[Any, Any], exists: Dict[Any, Any], root: str + self, origin: dict[Any, Any], exists: dict[Any, Any], root: str ) -> None: """Check if all options are exists.""" missing = set(origin) - set(exists) @@ -267,9 +267,9 @@ class UiOptions(CoreSysAttributes): """Initialize UI option render.""" self.coresys = coresys - def __call__(self, raw_schema: Dict[str, Any]) -> List[Dict[str, Any]]: + def __call__(self, raw_schema: dict[str, Any]) -> list[dict[str, Any]]: """Generate UI schema.""" - ui_schema: List[Dict[str, Any]] = [] + ui_schema: list[dict[str, Any]] = [] # read options for key, value in raw_schema.items(): @@ -287,13 +287,13 @@ class UiOptions(CoreSysAttributes): def _single_ui_option( self, - ui_schema: List[Dict[str, Any]], + ui_schema: list[dict[str, Any]], value: str, key: str, multiple: bool = False, ) -> None: """Validate a single element.""" - ui_node: Dict[str, Union[str, bool, float, List[str]]] = {"name": key} + ui_node: dict[str, Union[str, bool, float, list[str]]] = {"name": key} # If multiple if multiple: @@ -365,8 +365,8 @@ class UiOptions(CoreSysAttributes): def _nested_ui_list( self, - ui_schema: List[Dict[str, Any]], - option_list: List[Any], + ui_schema: list[dict[str, Any]], + option_list: list[Any], key: str, ) -> None: """UI nested list items.""" @@ -383,8 +383,8 @@ class UiOptions(CoreSysAttributes): def _nested_ui_dict( self, - ui_schema: List[Dict[str, Any]], - option_dict: Dict[str, Any], + ui_schema: list[dict[str, Any]], + option_dict: dict[str, Any], key: str, multiple: bool = False, ) -> None: @@ -408,7 +408,7 @@ class UiOptions(CoreSysAttributes): ui_schema.append(ui_node) -def _create_device_filter(str_filter: str) -> Dict[str, Any]: +def _create_device_filter(str_filter: str) -> dict[str, Any]: """Generate device Filter.""" raw_filter = dict(value.split("=") for value in str_filter.split(";")) diff --git a/supervisor/addons/validate.py b/supervisor/addons/validate.py index cbf7ec2c4..ee303a48b 100644 --- a/supervisor/addons/validate.py +++ b/supervisor/addons/validate.py @@ -2,7 +2,7 @@ import logging import re import secrets -from typing import Any, Dict +from typing import Any import uuid import voluptuous as vol @@ -148,7 +148,7 @@ RE_MACHINE = re.compile( ) -def _warn_addon_config(config: Dict[str, Any]): +def _warn_addon_config(config: dict[str, Any]): """Warn about miss configs.""" name = config.get(ATTR_NAME) if not name: @@ -179,7 +179,7 @@ def _warn_addon_config(config: Dict[str, Any]): def _migrate_addon_config(protocol=False): """Migrate addon config.""" - def _migrate(config: Dict[str, Any]): + def _migrate(config: dict[str, Any]): name = config.get(ATTR_NAME) if not name: raise vol.Invalid("Invalid Add-on config!") diff --git a/supervisor/api/addons.py b/supervisor/api/addons.py index aa316828b..8ca5b5901 100644 --- a/supervisor/api/addons.py +++ b/supervisor/api/addons.py @@ -1,7 +1,7 @@ """Init file for Supervisor Home Assistant RESTful API.""" import asyncio import logging -from typing import Any, Awaitable, Dict, List +from typing import Any, Awaitable from aiohttp import web import voluptuous as vol @@ -156,7 +156,7 @@ class APIAddons(CoreSysAttributes): return addon @api_process - async def list(self, request: web.Request) -> Dict[str, Any]: + async def list(self, request: web.Request) -> dict[str, Any]: """Return all add-ons or repositories.""" data_addons = [ { @@ -201,7 +201,7 @@ class APIAddons(CoreSysAttributes): await asyncio.shield(self.sys_store.reload()) @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return add-on information.""" addon: AnyAddon = self._extract_addon(request) @@ -390,7 +390,7 @@ class APIAddons(CoreSysAttributes): async def security(self, request: web.Request) -> None: """Store security options for add-on.""" addon = self._extract_addon_installed(request) - body: Dict[str, Any] = await api_validate(SCHEMA_SECURITY, request) + body: dict[str, Any] = await api_validate(SCHEMA_SECURITY, request) if ATTR_PROTECTED in body: _LOGGER.warning("Changing protected flag for %s!", addon.slug) @@ -399,7 +399,7 @@ class APIAddons(CoreSysAttributes): addon.save_persist() @api_process - async def stats(self, request: web.Request) -> Dict[str, Any]: + async def stats(self, request: web.Request) -> dict[str, Any]: """Return resource information.""" addon = self._extract_addon_installed(request) @@ -503,6 +503,6 @@ class APIAddons(CoreSysAttributes): await asyncio.shield(addon.write_stdin(data)) -def _pretty_services(addon: AnyAddon) -> List[str]: +def _pretty_services(addon: AnyAddon) -> list[str]: """Return a simplified services role list.""" return [f"{name}:{access}" for name, access in addon.services_role.items()] diff --git a/supervisor/api/audio.py b/supervisor/api/audio.py index 7dae7ab3c..6dad3fa00 100644 --- a/supervisor/api/audio.py +++ b/supervisor/api/audio.py @@ -1,7 +1,7 @@ """Init file for Supervisor Audio RESTful API.""" import asyncio import logging -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from aiohttp import web import attr @@ -67,7 +67,7 @@ class APIAudio(CoreSysAttributes): """Handle RESTful API for Audio functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return Audio information.""" return { ATTR_VERSION: self.sys_plugins.audio.version, @@ -89,7 +89,7 @@ class APIAudio(CoreSysAttributes): } @api_process - async def stats(self, request: web.Request) -> Dict[str, Any]: + async def stats(self, request: web.Request) -> dict[str, Any]: """Return resource information.""" stats = await self.sys_plugins.audio.stats() diff --git a/supervisor/api/auth.py b/supervisor/api/auth.py index a832a0b06..143f1a7ec 100644 --- a/supervisor/api/auth.py +++ b/supervisor/api/auth.py @@ -1,7 +1,6 @@ """Init file for Supervisor auth/SSO RESTful API.""" import asyncio import logging -from typing import Dict from aiohttp import BasicAuth, web from aiohttp.hdrs import AUTHORIZATION, CONTENT_TYPE, WWW_AUTHENTICATE @@ -29,7 +28,7 @@ SCHEMA_PASSWORD_RESET = vol.Schema( } ) -REALM_HEADER: Dict[str, str] = { +REALM_HEADER: dict[str, str] = { WWW_AUTHENTICATE: 'Basic realm="Home Assistant Authentication"' } @@ -46,7 +45,7 @@ class APIAuth(CoreSysAttributes): return self.sys_auth.check_login(addon, auth.login, auth.password) def _process_dict( - self, request: web.Request, addon: Addon, data: Dict[str, str] + self, request: web.Request, addon: Addon, data: dict[str, str] ) -> bool: """Process login with dict data. @@ -86,7 +85,7 @@ class APIAuth(CoreSysAttributes): @api_process async def reset(self, request: web.Request) -> None: """Process reset password request.""" - body: Dict[str, str] = await api_validate(SCHEMA_PASSWORD_RESET, request) + body: dict[str, str] = await api_validate(SCHEMA_PASSWORD_RESET, request) await asyncio.shield( self.sys_auth.change_password(body[ATTR_USERNAME], body[ATTR_PASSWORD]) ) diff --git a/supervisor/api/cli.py b/supervisor/api/cli.py index f82f02ce6..ebaf287d7 100644 --- a/supervisor/api/cli.py +++ b/supervisor/api/cli.py @@ -1,7 +1,7 @@ """Init file for Supervisor HA cli RESTful API.""" import asyncio import logging -from typing import Any, Dict +from typing import Any from aiohttp import web import voluptuous as vol @@ -32,7 +32,7 @@ class APICli(CoreSysAttributes): """Handle RESTful API for HA Cli functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return HA cli information.""" return { ATTR_VERSION: self.sys_plugins.cli.version, @@ -41,7 +41,7 @@ class APICli(CoreSysAttributes): } @api_process - async def stats(self, request: web.Request) -> Dict[str, Any]: + async def stats(self, request: web.Request) -> dict[str, Any]: """Return resource information.""" stats = await self.sys_plugins.cli.stats() diff --git a/supervisor/api/dns.py b/supervisor/api/dns.py index a31421eb1..2719edd01 100644 --- a/supervisor/api/dns.py +++ b/supervisor/api/dns.py @@ -1,7 +1,7 @@ """Init file for Supervisor DNS RESTful API.""" import asyncio import logging -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from aiohttp import web import voluptuous as vol @@ -40,7 +40,7 @@ class APICoreDNS(CoreSysAttributes): """Handle RESTful API for DNS functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return DNS information.""" return { ATTR_VERSION: self.sys_plugins.dns.version, @@ -63,7 +63,7 @@ class APICoreDNS(CoreSysAttributes): self.sys_plugins.dns.save_data() @api_process - async def stats(self, request: web.Request) -> Dict[str, Any]: + async def stats(self, request: web.Request) -> dict[str, Any]: """Return resource information.""" stats = await self.sys_plugins.dns.stats() diff --git a/supervisor/api/docker.py b/supervisor/api/docker.py index f2895712c..66b305189 100644 --- a/supervisor/api/docker.py +++ b/supervisor/api/docker.py @@ -1,6 +1,6 @@ """Init file for Supervisor Home Assistant RESTful API.""" import logging -from typing import Any, Dict +from typing import Any from aiohttp import web import voluptuous as vol @@ -33,7 +33,7 @@ class APIDocker(CoreSysAttributes): """Handle RESTful API for Docker configuration.""" @api_process - async def registries(self, request) -> Dict[str, Any]: + async def registries(self, request) -> dict[str, Any]: """Return the list of registries.""" data_registries = {} for hostname, registry in self.sys_docker.config.registries.items(): diff --git a/supervisor/api/hardware.py b/supervisor/api/hardware.py index fb24d1580..376bdfd36 100644 --- a/supervisor/api/hardware.py +++ b/supervisor/api/hardware.py @@ -1,6 +1,6 @@ """Init file for Supervisor hardware RESTful API.""" import logging -from typing import Any, Dict +from typing import Any from aiohttp import web @@ -19,7 +19,7 @@ from .utils import api_process _LOGGER: logging.Logger = logging.getLogger(__name__) -def device_struct(device: Device) -> Dict[str, Any]: +def device_struct(device: Device) -> dict[str, Any]: """Return a dict with information of a interface to be used in th API.""" return { ATTR_NAME: device.name, @@ -35,7 +35,7 @@ class APIHardware(CoreSysAttributes): """Handle RESTful API for hardware functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Show hardware info.""" return { ATTR_DEVICES: [ @@ -44,7 +44,7 @@ class APIHardware(CoreSysAttributes): } @api_process - async def audio(self, request: web.Request) -> Dict[str, Any]: + async def audio(self, request: web.Request) -> dict[str, Any]: """Show pulse audio profiles.""" return { ATTR_AUDIO: { diff --git a/supervisor/api/homeassistant.py b/supervisor/api/homeassistant.py index 86bd404ad..579251bda 100644 --- a/supervisor/api/homeassistant.py +++ b/supervisor/api/homeassistant.py @@ -1,7 +1,7 @@ """Init file for Supervisor Home Assistant RESTful API.""" import asyncio import logging -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from aiohttp import web import voluptuous as vol @@ -61,7 +61,7 @@ class APIHomeAssistant(CoreSysAttributes): """Handle RESTful API for Home Assistant functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return host information.""" return { ATTR_VERSION: self.sys_homeassistant.version, @@ -117,7 +117,7 @@ class APIHomeAssistant(CoreSysAttributes): self.sys_homeassistant.save_data() @api_process - async def stats(self, request: web.Request) -> Dict[Any, str]: + async def stats(self, request: web.Request) -> dict[Any, str]: """Return resource information.""" stats = await self.sys_homeassistant.core.stats() if not stats: diff --git a/supervisor/api/info.py b/supervisor/api/info.py index 77ded5aa9..efee6e29b 100644 --- a/supervisor/api/info.py +++ b/supervisor/api/info.py @@ -1,6 +1,6 @@ """Init file for Supervisor info RESTful API.""" import logging -from typing import Any, Dict +from typing import Any from aiohttp import web @@ -31,7 +31,7 @@ class APIInfo(CoreSysAttributes): """Handle RESTful API for info functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Show system info.""" return { ATTR_SUPERVISOR: self.sys_supervisor.version, diff --git a/supervisor/api/ingress.py b/supervisor/api/ingress.py index 44c8d769d..40be8c739 100644 --- a/supervisor/api/ingress.py +++ b/supervisor/api/ingress.py @@ -2,7 +2,7 @@ import asyncio from ipaddress import ip_address import logging -from typing import Any, Dict, Union +from typing import Any, Union import aiohttp from aiohttp import ClientTimeout, hdrs, web @@ -54,7 +54,7 @@ class APIIngress(CoreSysAttributes): return f"http://{addon.ip_address}:{addon.ingress_port}/{path}" @api_process - async def panels(self, request: web.Request) -> Dict[str, Any]: + async def panels(self, request: web.Request) -> dict[str, Any]: """Create a list of panel data.""" addons = {} for addon in self.sys_ingress.addons: @@ -69,14 +69,14 @@ class APIIngress(CoreSysAttributes): @api_process @require_home_assistant - async def create_session(self, request: web.Request) -> Dict[str, Any]: + async def create_session(self, request: web.Request) -> dict[str, Any]: """Create a new session.""" session = self.sys_ingress.create_session() return {ATTR_SESSION: session} @api_process @require_home_assistant - async def validate_session(self, request: web.Request) -> Dict[str, Any]: + async def validate_session(self, request: web.Request) -> dict[str, Any]: """Validate session and extending how long it's valid for.""" data = await api_validate(VALIDATE_SESSION_DATA, request) @@ -220,7 +220,7 @@ class APIIngress(CoreSysAttributes): def _init_header( request: web.Request, addon: str -) -> Union[CIMultiDict, Dict[str, str]]: +) -> Union[CIMultiDict, dict[str, str]]: """Create initial header.""" headers = {} @@ -248,7 +248,7 @@ def _init_header( return headers -def _response_header(response: aiohttp.ClientResponse) -> Dict[str, str]: +def _response_header(response: aiohttp.ClientResponse) -> dict[str, str]: """Create response header.""" headers = {} diff --git a/supervisor/api/jobs.py b/supervisor/api/jobs.py index fe0e016f6..10ed860a8 100644 --- a/supervisor/api/jobs.py +++ b/supervisor/api/jobs.py @@ -1,6 +1,6 @@ """Init file for Supervisor Jobs RESTful API.""" import logging -from typing import Any, Dict +from typing import Any from aiohttp import web import voluptuous as vol @@ -20,7 +20,7 @@ class APIJobs(CoreSysAttributes): """Handle RESTful API for OS functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return JobManager information.""" return { ATTR_IGNORE_CONDITIONS: self.sys_jobs.ignore_conditions, diff --git a/supervisor/api/multicast.py b/supervisor/api/multicast.py index 6e7b23062..c70a3b50c 100644 --- a/supervisor/api/multicast.py +++ b/supervisor/api/multicast.py @@ -1,7 +1,7 @@ """Init file for Supervisor Multicast RESTful API.""" import asyncio import logging -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from aiohttp import web import voluptuous as vol @@ -34,7 +34,7 @@ class APIMulticast(CoreSysAttributes): """Handle RESTful API for Multicast functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return Multicast information.""" return { ATTR_VERSION: self.sys_plugins.multicast.version, @@ -43,7 +43,7 @@ class APIMulticast(CoreSysAttributes): } @api_process - async def stats(self, request: web.Request) -> Dict[str, Any]: + async def stats(self, request: web.Request) -> dict[str, Any]: """Return resource information.""" stats = await self.sys_plugins.multicast.stats() diff --git a/supervisor/api/network.py b/supervisor/api/network.py index 627176544..cc42ee1f2 100644 --- a/supervisor/api/network.py +++ b/supervisor/api/network.py @@ -1,7 +1,7 @@ """REST API for network.""" import asyncio from ipaddress import ip_address, ip_interface -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from aiohttp import web import attr @@ -82,7 +82,7 @@ SCHEMA_UPDATE = vol.Schema( ) -def ipconfig_struct(config: IpConfig) -> Dict[str, Any]: +def ipconfig_struct(config: IpConfig) -> dict[str, Any]: """Return a dict with information about ip configuration.""" return { ATTR_METHOD: config.method, @@ -92,7 +92,7 @@ def ipconfig_struct(config: IpConfig) -> Dict[str, Any]: } -def wifi_struct(config: WifiConfig) -> Dict[str, Any]: +def wifi_struct(config: WifiConfig) -> dict[str, Any]: """Return a dict with information about wifi configuration.""" return { ATTR_MODE: config.mode, @@ -102,7 +102,7 @@ def wifi_struct(config: WifiConfig) -> Dict[str, Any]: } -def vlan_struct(config: VlanConfig) -> Dict[str, Any]: +def vlan_struct(config: VlanConfig) -> dict[str, Any]: """Return a dict with information about VLAN configuration.""" return { ATTR_ID: config.id, @@ -110,7 +110,7 @@ def vlan_struct(config: VlanConfig) -> Dict[str, Any]: } -def interface_struct(interface: Interface) -> Dict[str, Any]: +def interface_struct(interface: Interface) -> dict[str, Any]: """Return a dict with information of a interface to be used in th API.""" return { ATTR_INTERFACE: interface.name, @@ -125,7 +125,7 @@ def interface_struct(interface: Interface) -> Dict[str, Any]: } -def accesspoint_struct(accesspoint: AccessPoint) -> Dict[str, Any]: +def accesspoint_struct(accesspoint: AccessPoint) -> dict[str, Any]: """Return a dict for AccessPoint.""" return { ATTR_MODE: accesspoint.mode, @@ -158,7 +158,7 @@ class APINetwork(CoreSysAttributes): raise APIError(f"Interface {name} does not exist") from None @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return network information.""" return { ATTR_INTERFACES: [ @@ -176,7 +176,7 @@ class APINetwork(CoreSysAttributes): } @api_process - async def interface_info(self, request: web.Request) -> Dict[str, Any]: + async def interface_info(self, request: web.Request) -> dict[str, Any]: """Return network information for a interface.""" interface = self._get_interface(request.match_info.get(ATTR_INTERFACE)) @@ -223,7 +223,7 @@ class APINetwork(CoreSysAttributes): return asyncio.shield(self.sys_host.network.update()) @api_process - async def scan_accesspoints(self, request: web.Request) -> Dict[str, Any]: + async def scan_accesspoints(self, request: web.Request) -> dict[str, Any]: """Scan and return a list of available networks.""" interface = self._get_interface(request.match_info.get(ATTR_INTERFACE)) diff --git a/supervisor/api/observer.py b/supervisor/api/observer.py index 5ecfda5d9..c350098be 100644 --- a/supervisor/api/observer.py +++ b/supervisor/api/observer.py @@ -1,7 +1,7 @@ """Init file for Supervisor Observer RESTful API.""" import asyncio import logging -from typing import Any, Dict +from typing import Any from aiohttp import web import voluptuous as vol @@ -33,7 +33,7 @@ class APIObserver(CoreSysAttributes): """Handle RESTful API for Observer functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return HA Observer information.""" return { ATTR_HOST: str(self.sys_docker.network.observer), @@ -43,7 +43,7 @@ class APIObserver(CoreSysAttributes): } @api_process - async def stats(self, request: web.Request) -> Dict[str, Any]: + async def stats(self, request: web.Request) -> dict[str, Any]: """Return resource information.""" stats = await self.sys_plugins.observer.stats() diff --git a/supervisor/api/os.py b/supervisor/api/os.py index ae1b4939d..8ba8d193d 100644 --- a/supervisor/api/os.py +++ b/supervisor/api/os.py @@ -2,7 +2,7 @@ import asyncio import logging from pathlib import Path -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from aiohttp import web import voluptuous as vol @@ -30,7 +30,7 @@ class APIOS(CoreSysAttributes): """Handle RESTful API for OS functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return OS information.""" return { ATTR_VERSION: self.sys_os.version, @@ -62,7 +62,7 @@ class APIOS(CoreSysAttributes): await asyncio.shield(self.sys_os.datadisk.migrate_disk(body[ATTR_DEVICE])) @api_process - async def list_data(self, request: web.Request) -> Dict[str, Any]: + async def list_data(self, request: web.Request) -> dict[str, Any]: """Return possible data targets.""" return { ATTR_DEVICES: self.sys_os.datadisk.available_disks, diff --git a/supervisor/api/resolution.py b/supervisor/api/resolution.py index e9033d536..5785e851d 100644 --- a/supervisor/api/resolution.py +++ b/supervisor/api/resolution.py @@ -1,6 +1,6 @@ """Handle REST API for resoulution.""" import asyncio -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from aiohttp import web import attr @@ -26,7 +26,7 @@ class APIResoulution(CoreSysAttributes): """Handle REST API for resoulution.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return network information.""" return { ATTR_UNSUPPORTED: self.sys_resolution.unsupported, diff --git a/supervisor/api/security.py b/supervisor/api/security.py index 431e1b6cb..681be2061 100644 --- a/supervisor/api/security.py +++ b/supervisor/api/security.py @@ -1,6 +1,6 @@ """Init file for Supervisor Security RESTful API.""" import logging -from typing import Any, Dict +from typing import Any from aiohttp import web import voluptuous as vol @@ -25,7 +25,7 @@ class APISecurity(CoreSysAttributes): """Handle RESTful API for Security functions.""" @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return Security information.""" return { ATTR_CONTENT_TRUST: self.sys_security.content_trust, diff --git a/supervisor/api/store.py b/supervisor/api/store.py index a4740d96f..b820f3e35 100644 --- a/supervisor/api/store.py +++ b/supervisor/api/store.py @@ -1,6 +1,6 @@ """Init file for Supervisor Home Assistant RESTful API.""" import asyncio -from typing import Any, Awaitable, Dict, List +from typing import Any, Awaitable from aiohttp import web @@ -67,7 +67,7 @@ class APIStore(CoreSysAttributes): return repository - def _generate_addon_information(self, addon: AddonStore) -> Dict[str, Any]: + def _generate_addon_information(self, addon: AddonStore) -> dict[str, Any]: """Generate addon information.""" return { ATTR_ADVANCED: addon.advanced, @@ -90,7 +90,7 @@ class APIStore(CoreSysAttributes): def _generate_repository_information( self, repository: Repository - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Generate repository information.""" return { ATTR_SLUG: repository.slug, @@ -106,7 +106,7 @@ class APIStore(CoreSysAttributes): await asyncio.shield(self.sys_store.reload()) @api_process - async def store_info(self, request: web.Request) -> Dict[str, Any]: + async def store_info(self, request: web.Request) -> dict[str, Any]: """Return store information.""" return { ATTR_ADDONS: [ @@ -120,7 +120,7 @@ class APIStore(CoreSysAttributes): } @api_process - async def addons_list(self, request: web.Request) -> List[Dict[str, Any]]: + async def addons_list(self, request: web.Request) -> list[dict[str, Any]]: """Return all store add-ons.""" return [ self._generate_addon_information(self.sys_addons.store[addon]) @@ -142,13 +142,13 @@ class APIStore(CoreSysAttributes): return asyncio.shield(addon.update()) @api_process - async def addons_addon_info(self, request: web.Request) -> Dict[str, Any]: + async def addons_addon_info(self, request: web.Request) -> dict[str, Any]: """Return add-on information.""" addon: AddonStore = self._extract_addon(request) return self._generate_addon_information(addon) @api_process - async def repositories_list(self, request: web.Request) -> List[Dict[str, Any]]: + async def repositories_list(self, request: web.Request) -> list[dict[str, Any]]: """Return all repositories.""" return [ self._generate_repository_information(repository) @@ -158,7 +158,7 @@ class APIStore(CoreSysAttributes): @api_process async def repositories_repository_info( self, request: web.Request - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Return repository information.""" repository: Repository = self._extract_repository(request) return self._generate_repository_information(repository) diff --git a/supervisor/api/supervisor.py b/supervisor/api/supervisor.py index 10df92ff3..1e190c7c8 100644 --- a/supervisor/api/supervisor.py +++ b/supervisor/api/supervisor.py @@ -1,7 +1,7 @@ """Init file for Supervisor Supervisor RESTful API.""" import asyncio import logging -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from aiohttp import web import voluptuous as vol @@ -82,7 +82,7 @@ class APISupervisor(CoreSysAttributes): return True @api_process - async def info(self, request: web.Request) -> Dict[str, Any]: + async def info(self, request: web.Request) -> dict[str, Any]: """Return host information.""" list_addons = [] for addon in self.sys_addons.installed: @@ -178,7 +178,7 @@ class APISupervisor(CoreSysAttributes): await self.sys_resolution.evaluate.evaluate_system() @api_process - async def stats(self, request: web.Request) -> Dict[str, Any]: + async def stats(self, request: web.Request) -> dict[str, Any]: """Return resource information.""" stats = await self.sys_supervisor.stats() diff --git a/supervisor/api/utils.py b/supervisor/api/utils.py index 9f12a9591..b97f84ba5 100644 --- a/supervisor/api/utils.py +++ b/supervisor/api/utils.py @@ -1,6 +1,6 @@ """Init file for Supervisor util for RESTful API.""" import json -from typing import Any, Dict, List, Optional +from typing import Any, Optional from aiohttp import web from aiohttp.hdrs import AUTHORIZATION @@ -46,7 +46,7 @@ def excract_supervisor_token(request: web.Request) -> Optional[str]: return None -def json_loads(data: Any) -> Dict[str, Any]: +def json_loads(data: Any) -> dict[str, Any]: """Extract json from string with support for '' and None.""" if not data: return {} @@ -135,7 +135,7 @@ def api_return_error( ) -def api_return_ok(data: Optional[Dict[str, Any]] = None) -> web.Response: +def api_return_ok(data: Optional[dict[str, Any]] = None) -> web.Response: """Return an API ok answer.""" return web.json_response( {JSON_RESULT: RESULT_OK, JSON_DATA: data or {}}, @@ -144,10 +144,10 @@ def api_return_ok(data: Optional[Dict[str, Any]] = None) -> web.Response: async def api_validate( - schema: vol.Schema, request: web.Request, origin: Optional[List[str]] = None -) -> Dict[str, Any]: + schema: vol.Schema, request: web.Request, origin: Optional[list[str]] = None +) -> dict[str, Any]: """Validate request data with schema.""" - data: Dict[str, Any] = await request.json(loads=json_loads) + data: dict[str, Any] = await request.json(loads=json_loads) try: data_validated = schema(data) except vol.Invalid as ex: diff --git a/supervisor/arch.py b/supervisor/arch.py index 705b5b05a..adbeb996e 100644 --- a/supervisor/arch.py +++ b/supervisor/arch.py @@ -2,7 +2,6 @@ import logging from pathlib import Path import platform -from typing import List from .coresys import CoreSys, CoreSysAttributes from .exceptions import ConfigurationFileError, HassioArchNotFound @@ -28,7 +27,7 @@ class CpuArch(CoreSysAttributes): def __init__(self, coresys: CoreSys) -> None: """Initialize CPU Architecture handler.""" self.coresys = coresys - self._supported_arch: List[str] = [] + self._supported_arch: list[str] = [] self._default_arch: str @property @@ -42,7 +41,7 @@ class CpuArch(CoreSysAttributes): return self.sys_supervisor.arch @property - def supported(self) -> List[str]: + def supported(self) -> list[str]: """Return support arch by CPU/Machine.""" return self._supported_arch @@ -71,11 +70,11 @@ class CpuArch(CoreSysAttributes): if native_support not in self._supported_arch: self._supported_arch.append(native_support) - def is_supported(self, arch_list: List[str]) -> bool: + def is_supported(self, arch_list: list[str]) -> bool: """Return True if there is a supported arch by this platform.""" return not set(self.supported).isdisjoint(set(arch_list)) - def match(self, arch_list: List[str]) -> str: + def match(self, arch_list: list[str]) -> str: """Return best match for this CPU/Platform.""" for self_arch in self.supported: if self_arch in arch_list: diff --git a/supervisor/auth.py b/supervisor/auth.py index 9bb11865d..13a8dfc0d 100644 --- a/supervisor/auth.py +++ b/supervisor/auth.py @@ -2,7 +2,7 @@ import asyncio import hashlib import logging -from typing import Dict, Optional +from typing import Optional from .addons.addon import Addon from .const import ATTR_ADDON, ATTR_PASSWORD, ATTR_USERNAME, FILE_HASSIO_AUTH @@ -22,7 +22,7 @@ class Auth(FileConfiguration, CoreSysAttributes): super().__init__(FILE_HASSIO_AUTH, SCHEMA_AUTH_CONFIG) self.coresys: CoreSys = coresys - self._running: Dict[str, asyncio.Task] = {} + self._running: dict[str, asyncio.Task] = {} def _check_cache(self, username: str, password: str) -> Optional[bool]: """Check password in cache.""" diff --git a/supervisor/backups/backup.py b/supervisor/backups/backup.py index 8603967e8..e56d0b0e8 100644 --- a/supervisor/backups/backup.py +++ b/supervisor/backups/backup.py @@ -5,7 +5,7 @@ import logging from pathlib import Path import tarfile from tempfile import TemporaryDirectory -from typing import Any, Dict, List, Optional, Set +from typing import Any, Optional from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import padding @@ -70,7 +70,7 @@ class Backup(CoreSysAttributes): """Initialize a backup.""" self.coresys: CoreSys = coresys self._tarfile: Path = tar_file - self._data: Dict[str, Any] = {} + self._data: dict[str, Any] = {} self._tmp = None self._key: Optional[bytes] = None self._aes: Optional[Cipher] = None @@ -311,9 +311,9 @@ class Backup(CoreSysAttributes): finally: self._tmp.cleanup() - async def store_addons(self, addon_list: Optional[List[Addon]] = None): + async def store_addons(self, addon_list: Optional[list[Addon]] = None): """Add a list of add-ons into backup.""" - addon_list: List[Addon] = addon_list or self.sys_addons.installed + addon_list: list[Addon] = addon_list or self.sys_addons.installed async def _addon_save(addon: Addon): """Task to store an add-on into backup.""" @@ -346,9 +346,9 @@ class Backup(CoreSysAttributes): except Exception as err: # pylint: disable=broad-except _LOGGER.warning("Can't save Add-on %s: %s", addon.slug, err) - async def restore_addons(self, addon_list: Optional[List[str]] = None): + async def restore_addons(self, addon_list: Optional[list[str]] = None): """Restore a list add-on from backup.""" - addon_list: List[str] = addon_list or self.addon_list + addon_list: list[str] = addon_list or self.addon_list async def _addon_restore(addon_slug: str): """Task to restore an add-on into backup.""" @@ -375,9 +375,9 @@ class Backup(CoreSysAttributes): except Exception as err: # pylint: disable=broad-except _LOGGER.warning("Can't restore Add-on %s: %s", slug, err) - async def store_folders(self, folder_list: Optional[List[str]] = None): + async def store_folders(self, folder_list: Optional[list[str]] = None): """Backup Supervisor data into backup.""" - folder_list: Set[str] = set(folder_list or ALL_FOLDERS) + folder_list: set[str] = set(folder_list or ALL_FOLDERS) def _folder_save(name: str): """Take backup of a folder.""" @@ -414,9 +414,9 @@ class Backup(CoreSysAttributes): except Exception as err: # pylint: disable=broad-except _LOGGER.warning("Can't save folder %s: %s", folder, err) - async def restore_folders(self, folder_list: Optional[List[str]] = None): + async def restore_folders(self, folder_list: Optional[list[str]] = None): """Backup Supervisor data into backup.""" - folder_list: Set[str] = set(folder_list or self.folders) + folder_list: set[str] = set(folder_list or self.folders) def _folder_restore(name: str): """Intenal function to restore a folder.""" diff --git a/supervisor/backups/manager.py b/supervisor/backups/manager.py index 185fa0db9..db54c7a7c 100644 --- a/supervisor/backups/manager.py +++ b/supervisor/backups/manager.py @@ -2,7 +2,7 @@ import asyncio import logging from pathlib import Path -from typing import Awaitable, Set +from typing import Awaitable from awesomeversion.awesomeversion import AwesomeVersion from awesomeversion.exceptions import AwesomeVersionCompare @@ -29,7 +29,7 @@ class BackupManager(CoreSysAttributes): self.lock = asyncio.Lock() @property - def list_backups(self) -> Set[Backup]: + def list_backups(self) -> set[Backup]: """Return a list of all backup objects.""" return set(self._backups.values()) diff --git a/supervisor/bus.py b/supervisor/bus.py index 83534d1b1..794d188e1 100644 --- a/supervisor/bus.py +++ b/supervisor/bus.py @@ -2,7 +2,7 @@ from __future__ import annotations import logging -from typing import Any, Awaitable, Callable, Dict, List +from typing import Any, Awaitable, Callable import attr @@ -18,7 +18,7 @@ class Bus(CoreSysAttributes): def __init__(self, coresys: CoreSys): """Initialize bus backend.""" self.coresys = coresys - self._listeners: Dict[BusEvent, List[EventListener]] = {} + self._listeners: dict[BusEvent, list[EventListener]] = {} def register_event( self, event: BusEvent, callback: Callable[[Any], Awaitable[None]] diff --git a/supervisor/config.py b/supervisor/config.py index 40a5d6aa9..35c092f6c 100644 --- a/supervisor/config.py +++ b/supervisor/config.py @@ -3,7 +3,7 @@ from datetime import datetime import logging import os from pathlib import Path, PurePath -from typing import List, Optional +from typing import Optional from awesomeversion import AwesomeVersion @@ -294,7 +294,7 @@ class CoreConfig(FileConfiguration): return PurePath(self.path_extern_supervisor, MEDIA_DATA) @property - def addons_repositories(self) -> List[str]: + def addons_repositories(self) -> list[str]: """Return list of custom Add-on repositories.""" return self._data[ATTR_ADDONS_CUSTOM_LIST] diff --git a/supervisor/core.py b/supervisor/core.py index 67c18f8e4..997adc8a2 100644 --- a/supervisor/core.py +++ b/supervisor/core.py @@ -3,7 +3,7 @@ import asyncio from contextlib import suppress from datetime import timedelta import logging -from typing import Awaitable, List, Optional +from typing import Awaitable, Optional import async_timeout @@ -105,7 +105,7 @@ class Core(CoreSysAttributes): await self.sys_supervisor.check_connectivity() # Order can be important! - setup_loads: List[Awaitable[None]] = [ + setup_loads: list[Awaitable[None]] = [ # rest api views self.sys_api.load(), # Load Host Hardware diff --git a/supervisor/coresys.py b/supervisor/coresys.py index 7e269fa27..764c5b056 100644 --- a/supervisor/coresys.py +++ b/supervisor/coresys.py @@ -6,7 +6,7 @@ from datetime import datetime import logging import os from types import MappingProxyType -from typing import TYPE_CHECKING, Any, Callable, Coroutine, Optional, TypeVar +from typing import TYPE_CHECKING, Any, Callable, Coroutine, TypeVar import aiohttp import sentry_sdk @@ -54,8 +54,8 @@ class CoreSys: def __init__(self): """Initialize coresys.""" # Static attributes protected - self._machine_id: Optional[str] = None - self._machine: Optional[str] = None + self._machine_id: str | None = None + self._machine: str | None = None # External objects self._loop: asyncio.BaseEventLoop = asyncio.get_running_loop() @@ -66,30 +66,30 @@ class CoreSys: self._docker: DockerAPI = DockerAPI() # Internal objects pointers - self._core: Optional[Core] = None - self._arch: Optional[CpuArch] = None - self._auth: Optional[Auth] = None - self._homeassistant: Optional[HomeAssistant] = None - self._supervisor: Optional[Supervisor] = None - self._addons: Optional[AddonManager] = None - self._api: Optional[RestAPI] = None - self._updater: Optional[Updater] = None - self._backups: Optional[BackupManager] = None - self._tasks: Optional[Tasks] = None - self._host: Optional[HostManager] = None - self._ingress: Optional[Ingress] = None - self._dbus: Optional[DBusManager] = None - self._os: Optional[OSManager] = None - self._services: Optional[ServiceManager] = None - self._scheduler: Optional[Scheduler] = None - self._store: Optional[StoreManager] = None - self._discovery: Optional[Discovery] = None - self._hardware: Optional[HardwareManager] = None - self._plugins: Optional[PluginManager] = None - self._resolution: Optional[ResolutionManager] = None - self._jobs: Optional[JobManager] = None - self._security: Optional[Security] = None - self._bus: Optional[Bus] = None + self._core: Core | None = None + self._arch: CpuArch | None = None + self._auth: Auth | None = None + self._homeassistant: HomeAssistant | None = None + self._supervisor: Supervisor | None = None + self._addons: AddonManager | None = None + self._api: RestAPI | None = None + self._updater: Updater | None = None + self._backups: BackupManager | None = None + self._tasks: Tasks | None = None + self._host: HostManager | None = None + self._ingress: Ingress | None = None + self._dbus: DBusManager | None = None + self._os: OSManager | None = None + self._services: ServiceManager | None = None + self._scheduler: Scheduler | None = None + self._store: StoreManager | None = None + self._discovery: Discovery | None = None + self._hardware: HardwareManager | None = None + self._plugins: PluginManager | None = None + self._resolution: ResolutionManager | None = None + self._jobs: JobManager | None = None + self._security: Security | None = None + self._bus: Bus | None = None # Set default header for aiohttp self._websession._default_headers = MappingProxyType( @@ -467,7 +467,7 @@ class CoreSys: self._jobs = value @property - def machine(self) -> Optional[str]: + def machine(self) -> str | None: """Return machine type string.""" return self._machine @@ -479,7 +479,7 @@ class CoreSys: self._machine = value @property - def machine_id(self) -> Optional[str]: + def machine_id(self) -> str | None: """Return machine-id type string.""" return self._machine_id @@ -520,7 +520,7 @@ class CoreSysAttributes: return self.coresys.timezone @property - def sys_machine(self) -> Optional[str]: + def sys_machine(self) -> str | None: """Return running machine type of the Supervisor system.""" return self.coresys.machine diff --git a/supervisor/dbus/agent/__init__.py b/supervisor/dbus/agent/__init__.py index de567f2b8..497d6a9a6 100644 --- a/supervisor/dbus/agent/__init__.py +++ b/supervisor/dbus/agent/__init__.py @@ -1,7 +1,7 @@ """OS-Agent implementation for DBUS.""" import asyncio import logging -from typing import Any, Dict +from typing import Any from awesomeversion import AwesomeVersion @@ -30,7 +30,7 @@ class OSAgent(DBusInterface): def __init__(self) -> None: """Initialize Properties.""" - self.properties: Dict[str, Any] = {} + self.properties: dict[str, Any] = {} self._cgroup: CGroup = CGroup() self._apparmor: AppArmor = AppArmor() diff --git a/supervisor/dbus/agent/apparmor.py b/supervisor/dbus/agent/apparmor.py index 17433d29c..749493cc0 100644 --- a/supervisor/dbus/agent/apparmor.py +++ b/supervisor/dbus/agent/apparmor.py @@ -1,6 +1,6 @@ """AppArmor object for OS-Agent.""" from pathlib import Path -from typing import Any, Dict +from typing import Any from awesomeversion import AwesomeVersion @@ -20,7 +20,7 @@ class AppArmor(DBusInterface): def __init__(self) -> None: """Initialize Properties.""" - self.properties: Dict[str, Any] = {} + self.properties: dict[str, Any] = {} @property @dbus_property diff --git a/supervisor/dbus/agent/datadisk.py b/supervisor/dbus/agent/datadisk.py index 461c96cb9..90ff3abf3 100644 --- a/supervisor/dbus/agent/datadisk.py +++ b/supervisor/dbus/agent/datadisk.py @@ -1,6 +1,6 @@ """DataDisk object for OS-Agent.""" from pathlib import Path -from typing import Any, Dict +from typing import Any from ...utils.gdbus import DBus from ..const import ( @@ -18,7 +18,7 @@ class DataDisk(DBusInterface): def __init__(self) -> None: """Initialize Properties.""" - self.properties: Dict[str, Any] = {} + self.properties: dict[str, Any] = {} @property @dbus_property diff --git a/supervisor/dbus/hostname.py b/supervisor/dbus/hostname.py index ff97981c3..b489d8251 100644 --- a/supervisor/dbus/hostname.py +++ b/supervisor/dbus/hostname.py @@ -1,6 +1,6 @@ """D-Bus interface for hostname.""" import logging -from typing import Any, Dict, Optional +from typing import Any, Optional from ..exceptions import DBusError, DBusInterfaceError from ..utils.gdbus import DBus @@ -27,7 +27,7 @@ class Hostname(DBusInterface): def __init__(self): """Initialize Properties.""" - self.properties: Dict[str, Any] = {} + self.properties: dict[str, Any] = {} async def connect(self): """Connect to system's D-Bus.""" diff --git a/supervisor/dbus/interface.py b/supervisor/dbus/interface.py index 64de23b2e..0bb2d311a 100644 --- a/supervisor/dbus/interface.py +++ b/supervisor/dbus/interface.py @@ -1,7 +1,7 @@ """Interface class for D-Bus wrappers.""" from abc import ABC, abstractmethod from functools import wraps -from typing import Any, Dict, Optional +from typing import Any, Optional from ..utils.gdbus import DBus @@ -44,7 +44,7 @@ class DBusInterfaceProxy(ABC): dbus: Optional[DBus] = None object_path: Optional[str] = None - properties: Optional[Dict[str, Any]] = None + properties: Optional[dict[str, Any]] = None @abstractmethod async def connect(self): diff --git a/supervisor/dbus/manager.py b/supervisor/dbus/manager.py index 8ea61dc02..a9185a99b 100644 --- a/supervisor/dbus/manager.py +++ b/supervisor/dbus/manager.py @@ -1,6 +1,5 @@ """D-Bus interface objects.""" import logging -from typing import List from ..const import SOCKET_DBUS from ..coresys import CoreSys, CoreSysAttributes @@ -74,7 +73,7 @@ class DBusManager(CoreSysAttributes): ) return - dbus_loads: List[DBusInterface] = [ + dbus_loads: list[DBusInterface] = [ self.agent, self.systemd, self.logind, diff --git a/supervisor/dbus/network/__init__.py b/supervisor/dbus/network/__init__.py index aa54b1a42..e5d9adfd5 100644 --- a/supervisor/dbus/network/__init__.py +++ b/supervisor/dbus/network/__init__.py @@ -1,6 +1,6 @@ """Network Manager implementation for DBUS.""" import logging -from typing import Any, Awaitable, Dict +from typing import Any, Awaitable from awesomeversion import AwesomeVersion, AwesomeVersionException import sentry_sdk @@ -42,9 +42,9 @@ class NetworkManager(DBusInterface): """Initialize Properties.""" self._dns: NetworkManagerDNS = NetworkManagerDNS() self._settings: NetworkManagerSettings = NetworkManagerSettings() - self._interfaces: Dict[str, NetworkInterface] = {} + self._interfaces: dict[str, NetworkInterface] = {} - self.properties: Dict[str, Any] = {} + self.properties: dict[str, Any] = {} @property def dns(self) -> NetworkManagerDNS: @@ -57,7 +57,7 @@ class NetworkManager(DBusInterface): return self._settings @property - def interfaces(self) -> Dict[str, NetworkInterface]: + def interfaces(self) -> dict[str, NetworkInterface]: """Return a dictionary of active interfaces.""" return self._interfaces diff --git a/supervisor/dbus/network/configuration.py b/supervisor/dbus/network/configuration.py index 887e3caf3..fb679d1bc 100644 --- a/supervisor/dbus/network/configuration.py +++ b/supervisor/dbus/network/configuration.py @@ -1,6 +1,6 @@ """NetworkConnection object4s for Network Manager.""" from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface -from typing import List, Optional, Union +from typing import Optional, Union import attr @@ -10,16 +10,16 @@ class IpConfiguration: """NetworkSettingsIPConfig object for Network Manager.""" gateway: Optional[Union[IPv6Address, IPv6Address]] = attr.ib() - nameservers: List[Union[IPv6Address, IPv6Address]] = attr.ib() - address: List[Union[IPv4Interface, IPv6Interface]] = attr.ib() + nameservers: list[Union[IPv6Address, IPv6Address]] = attr.ib() + address: list[Union[IPv4Interface, IPv6Interface]] = attr.ib() @attr.s(slots=True) class DNSConfiguration: """DNS configuration Object.""" - nameservers: List[Union[IPv4Address, IPv6Address]] = attr.ib() - domains: List[str] = attr.ib() + nameservers: list[Union[IPv4Address, IPv6Address]] = attr.ib() + domains: list[str] = attr.ib() interface: str = attr.ib() priority: int = attr.ib() vpn: bool = attr.ib() diff --git a/supervisor/dbus/network/dns.py b/supervisor/dbus/network/dns.py index 7fa1e21d9..8dbb9ba09 100644 --- a/supervisor/dbus/network/dns.py +++ b/supervisor/dbus/network/dns.py @@ -1,7 +1,7 @@ """D-Bus interface for hostname.""" from ipaddress import ip_address import logging -from typing import List, Optional +from typing import Optional from ...const import ( ATTR_DOMAINS, @@ -34,7 +34,7 @@ class NetworkManagerDNS(DBusInterface): """Initialize Properties.""" self._mode: Optional[str] = None self._rc_manager: Optional[str] = None - self._configuration: List[DNSConfiguration] = [] + self._configuration: list[DNSConfiguration] = [] @property def mode(self) -> Optional[str]: @@ -47,7 +47,7 @@ class NetworkManagerDNS(DBusInterface): return self._rc_manager @property - def configuration(self) -> List[DNSConfiguration]: + def configuration(self) -> list[DNSConfiguration]: """Return Propertie configuraton.""" return self._configuration diff --git a/supervisor/dbus/payloads/generate.py b/supervisor/dbus/payloads/generate.py index caf4ced07..b862c3935 100644 --- a/supervisor/dbus/payloads/generate.py +++ b/supervisor/dbus/payloads/generate.py @@ -4,7 +4,7 @@ from __future__ import annotations from ipaddress import IPv4Address, IPv6Address from pathlib import Path import socket -from typing import TYPE_CHECKING, Optional +from typing import TYPE_CHECKING from uuid import uuid4 import jinja2 @@ -21,7 +21,7 @@ INTERFACE_UPDATE_TEMPLATE: Path = ( def interface_update_payload( - interface: Interface, name: Optional[str] = None, uuid: Optional[str] = None + interface: Interface, name: str | None = None, uuid: str | None = None ) -> str: """Generate a payload for network interface update.""" env = jinja2.Environment() diff --git a/supervisor/dbus/timedate.py b/supervisor/dbus/timedate.py index 6d6f4f38d..24c4a8634 100644 --- a/supervisor/dbus/timedate.py +++ b/supervisor/dbus/timedate.py @@ -1,7 +1,7 @@ """Interface to systemd-timedate over D-Bus.""" from datetime import datetime import logging -from typing import Any, Dict +from typing import Any from ..exceptions import DBusError, DBusInterfaceError from ..utils.dt import utc_from_timestamp @@ -28,7 +28,7 @@ class TimeDate(DBusInterface): def __init__(self) -> None: """Initialize Properties.""" - self.properties: Dict[str, Any] = {} + self.properties: dict[str, Any] = {} @property @dbus_property diff --git a/supervisor/discovery/__init__.py b/supervisor/discovery/__init__.py index dbf806545..b0fbd26cb 100644 --- a/supervisor/discovery/__init__.py +++ b/supervisor/discovery/__init__.py @@ -3,7 +3,7 @@ from __future__ import annotations from contextlib import suppress import logging -from typing import TYPE_CHECKING, Any, Dict, List, Optional +from typing import TYPE_CHECKING, Any from uuid import UUID, uuid4 import attr @@ -31,7 +31,7 @@ class Message: addon: str = attr.ib() service: str = attr.ib() - config: Dict[str, Any] = attr.ib(eq=False) + config: dict[str, Any] = attr.ib(eq=False) uuid: UUID = attr.ib(factory=lambda: uuid4().hex, eq=False) @@ -42,7 +42,7 @@ class Discovery(CoreSysAttributes, FileConfiguration): """Initialize discovery handler.""" super().__init__(FILE_HASSIO_DISCOVERY, SCHEMA_DISCOVERY_CONFIG) self.coresys: CoreSys = coresys - self.message_obj: Dict[str, Message] = {} + self.message_obj: dict[str, Message] = {} async def load(self) -> None: """Load exists discovery message into storage.""" @@ -56,7 +56,7 @@ class Discovery(CoreSysAttributes, FileConfiguration): def save(self) -> None: """Write discovery message into data file.""" - messages: List[Dict[str, Any]] = [] + messages: list[dict[str, Any]] = [] for message in self.list_messages: messages.append(attr.asdict(message)) @@ -64,16 +64,16 @@ class Discovery(CoreSysAttributes, FileConfiguration): self._data[ATTR_DISCOVERY].extend(messages) self.save_data() - def get(self, uuid: str) -> Optional[Message]: + def get(self, uuid: str) -> Message | None: """Return discovery message.""" return self.message_obj.get(uuid) @property - def list_messages(self) -> List[Message]: + def list_messages(self) -> list[Message]: """Return list of available discovery messages.""" return list(self.message_obj.values()) - def send(self, addon: Addon, service: str, config: Dict[str, Any]) -> Message: + def send(self, addon: Addon, service: str, config: dict[str, Any]) -> Message: """Send a discovery message to Home Assistant.""" try: config = valid_discovery_config(service, config) diff --git a/supervisor/docker/__init__.py b/supervisor/docker/__init__.py index 085212305..fd9d99a7a 100644 --- a/supervisor/docker/__init__.py +++ b/supervisor/docker/__init__.py @@ -4,7 +4,7 @@ from ipaddress import IPv4Address import logging import os from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Optional import attr from awesomeversion import AwesomeVersion, AwesomeVersionCompare @@ -47,7 +47,7 @@ class DockerInfo: logging: str = attr.ib() @staticmethod - def new(data: Dict[str, Any]): + def new(data: dict[str, Any]): """Create a object from docker info.""" return DockerInfo( AwesomeVersion(data["ServerVersion"]), data["Driver"], data["LoggingDriver"] @@ -77,7 +77,7 @@ class DockerConfig(FileConfiguration): super().__init__(FILE_HASSIO_DOCKER, SCHEMA_DOCKER_CONFIG) @property - def registries(self) -> Dict[str, Any]: + def registries(self) -> dict[str, Any]: """Return credentials for docker registries.""" return self._data.get(ATTR_REGISTRIES, {}) @@ -91,7 +91,7 @@ class DockerAPI: def __init__(self): """Initialize Docker base wrapper.""" self.docker: docker.DockerClient = docker.DockerClient( - base_url="unix:/{}".format(str(SOCKET_DOCKER)), version="auto", timeout=900 + base_url=f"unix:/{str(SOCKET_DOCKER)}", version="auto", timeout=900 ) self.network: DockerNetwork = DockerNetwork(self.docker) self._info: DockerInfo = DockerInfo.new(self.docker.info()) diff --git a/supervisor/docker/addon.py b/supervisor/docker/addon.py index 1844cfd32..3aa0b566b 100644 --- a/supervisor/docker/addon.py +++ b/supervisor/docker/addon.py @@ -6,7 +6,7 @@ from ipaddress import IPv4Address, ip_address import logging import os from pathlib import Path -from typing import TYPE_CHECKING, Awaitable, Dict, List, Optional, Set, Union +from typing import TYPE_CHECKING, Awaitable from awesomeversion import AwesomeVersion import docker @@ -55,7 +55,7 @@ class DockerAddon(DockerInterface): self.addon = addon @property - def image(self) -> Optional[str]: + def image(self) -> str | None: """Return name of Docker image.""" return self.addon.image @@ -98,7 +98,7 @@ class DockerAddon(DockerInterface): return f"addon_{self.addon.slug}" @property - def environment(self) -> Dict[str, Optional[str]]: + def environment(self) -> dict[str, str | None]: """Return environment for Docker add-on.""" addon_env = self.addon.environment or {} @@ -118,7 +118,7 @@ class DockerAddon(DockerInterface): } @property - def cgroups_rules(self) -> Optional[List[str]]: + def cgroups_rules(self) -> list[str] | None: """Return a list of needed cgroups permission.""" rules = set() @@ -177,7 +177,7 @@ class DockerAddon(DockerInterface): return None @property - def ports(self) -> Optional[Dict[str, Union[str, int, None]]]: + def ports(self) -> dict[str, str | int | None] | None: """Filter None from add-on ports.""" if self.addon.host_network or not self.addon.ports: return None @@ -189,7 +189,7 @@ class DockerAddon(DockerInterface): } @property - def security_opt(self) -> List[str]: + def security_opt(self) -> list[str]: """Control security options.""" security = super().security_opt @@ -203,7 +203,7 @@ class DockerAddon(DockerInterface): return security @property - def tmpfs(self) -> Optional[Dict[str, str]]: + def tmpfs(self) -> dict[str, str] | None: """Return tmpfs for Docker add-on.""" tmpfs = {} @@ -219,7 +219,7 @@ class DockerAddon(DockerInterface): return None @property - def network_mapping(self) -> Dict[str, IPv4Address]: + def network_mapping(self) -> dict[str, IPv4Address]: """Return hosts mapping.""" return { "supervisor": self.sys_docker.network.supervisor, @@ -227,23 +227,23 @@ class DockerAddon(DockerInterface): } @property - def network_mode(self) -> Optional[str]: + def network_mode(self) -> str | None: """Return network mode for add-on.""" if self.addon.host_network: return "host" return None @property - def pid_mode(self) -> Optional[str]: + def pid_mode(self) -> str | None: """Return PID mode for add-on.""" if not self.addon.protected and self.addon.host_pid: return "host" return None @property - def capabilities(self) -> Optional[List[str]]: + def capabilities(self) -> list[str] | None: """Generate needed capabilities.""" - capabilities: Set[Capabilities] = set(self.addon.privileged) + capabilities: set[Capabilities] = set(self.addon.privileged) # Need work with kernel modules if self.addon.with_kernel_modules: @@ -259,9 +259,9 @@ class DockerAddon(DockerInterface): return None @property - def ulimits(self) -> Optional[List[docker.types.Ulimit]]: + def ulimits(self) -> list[docker.types.Ulimit] | None: """Generate ulimits for add-on.""" - limits: List[docker.types.Ulimit] = [] + limits: list[docker.types.Ulimit] = [] # Need schedule functions if self.addon.with_realtime: @@ -277,7 +277,7 @@ class DockerAddon(DockerInterface): return None @property - def cpu_rt_runtime(self) -> Optional[int]: + def cpu_rt_runtime(self) -> int | None: """Limit CPU real-time runtime in microseconds.""" if not self.sys_docker.info.support_cpu_realtime: return None @@ -288,7 +288,7 @@ class DockerAddon(DockerInterface): return None @property - def volumes(self) -> Dict[str, Dict[str, str]]: + def volumes(self) -> dict[str, dict[str, str]]: """Generate volumes for mappings.""" addon_mapping = self.addon.map_volumes @@ -494,7 +494,7 @@ class DockerAddon(DockerInterface): self.sys_capture_exception(err) def _install( - self, version: AwesomeVersion, image: Optional[str] = None, latest: bool = False + self, version: AwesomeVersion, image: str | None = None, latest: bool = False ) -> None: """Pull Docker image or build it. diff --git a/supervisor/docker/audio.py b/supervisor/docker/audio.py index ec0d67f62..5ede19352 100644 --- a/supervisor/docker/audio.py +++ b/supervisor/docker/audio.py @@ -1,6 +1,6 @@ """Audio docker object.""" import logging -from typing import Dict, List, Optional +from typing import Optional import docker @@ -34,7 +34,7 @@ class DockerAudio(DockerInterface, CoreSysAttributes): return AUDIO_DOCKER_NAME @property - def volumes(self) -> Dict[str, Dict[str, str]]: + def volumes(self) -> dict[str, dict[str, str]]: """Return Volumes for the mount.""" volumes = { "/dev": {"bind": "/dev", "mode": "ro"}, @@ -50,19 +50,19 @@ class DockerAudio(DockerInterface, CoreSysAttributes): return volumes @property - def cgroups_rules(self) -> List[str]: + def cgroups_rules(self) -> list[str]: """Return a list of needed cgroups permission.""" return self.sys_hardware.policy.get_cgroups_rules( PolicyGroup.AUDIO ) + self.sys_hardware.policy.get_cgroups_rules(PolicyGroup.BLUETOOTH) @property - def capabilities(self) -> List[str]: + def capabilities(self) -> list[str]: """Generate needed capabilities.""" return [cap.value for cap in (Capabilities.SYS_NICE, Capabilities.SYS_RESOURCE)] @property - def ulimits(self) -> List[docker.types.Ulimit]: + def ulimits(self) -> list[docker.types.Ulimit]: """Generate ulimits for audio.""" # Pulseaudio by default tries to use real-time scheduling with priority of 5. return [docker.types.Ulimit(name="rtprio", soft=10, hard=10)] diff --git a/supervisor/docker/homeassistant.py b/supervisor/docker/homeassistant.py index 467738a71..24083e0f4 100644 --- a/supervisor/docker/homeassistant.py +++ b/supervisor/docker/homeassistant.py @@ -1,7 +1,7 @@ """Init file for Supervisor Docker object.""" from ipaddress import IPv4Address import logging -from typing import Awaitable, Dict, List, Optional +from typing import Awaitable, Optional from awesomeversion import AwesomeVersion, AwesomeVersionCompare import docker @@ -51,7 +51,7 @@ class DockerHomeAssistant(DockerInterface): return self.sys_docker.network.gateway @property - def cgroups_rules(self) -> List[str]: + def cgroups_rules(self) -> list[str]: """Return a list of needed cgroups permission.""" return ( self.sys_hardware.policy.get_cgroups_rules(PolicyGroup.UART) @@ -61,7 +61,7 @@ class DockerHomeAssistant(DockerInterface): ) @property - def volumes(self) -> Dict[str, Dict[str, str]]: + def volumes(self) -> dict[str, dict[str, str]]: """Return Volumes for the mount.""" volumes = { "/dev": {"bind": "/dev", "mode": "ro"}, diff --git a/supervisor/docker/interface.py b/supervisor/docker/interface.py index 00c0f2afd..0b42ba3d9 100644 --- a/supervisor/docker/interface.py +++ b/supervisor/docker/interface.py @@ -3,7 +3,7 @@ import asyncio from contextlib import suppress import logging import re -from typing import Any, Awaitable, Dict, List, Optional +from typing import Any, Awaitable, Optional from awesomeversion import AwesomeVersion from awesomeversion.strategy import AwesomeVersionStrategy @@ -44,7 +44,7 @@ class DockerInterface(CoreSysAttributes): def __init__(self, coresys: CoreSys): """Initialize Docker base wrapper.""" self.coresys: CoreSys = coresys - self._meta: Optional[Dict[str, Any]] = None + self._meta: Optional[dict[str, Any]] = None self.lock: asyncio.Lock = asyncio.Lock() @property @@ -58,21 +58,21 @@ class DockerInterface(CoreSysAttributes): return None @property - def meta_config(self) -> Dict[str, Any]: + def meta_config(self) -> dict[str, Any]: """Return meta data of configuration for container/image.""" if not self._meta: return {} return self._meta.get("Config", {}) @property - def meta_host(self) -> Dict[str, Any]: + def meta_host(self) -> dict[str, Any]: """Return meta data of configuration for host.""" if not self._meta: return {} return self._meta.get("HostConfig", {}) @property - def meta_labels(self) -> Dict[str, str]: + def meta_labels(self) -> dict[str, str]: """Return meta data of labels for container/image.""" return self.meta_config.get("Labels") or {} @@ -102,7 +102,7 @@ class DockerInterface(CoreSysAttributes): return self.lock.locked() @property - def security_opt(self) -> List[str]: + def security_opt(self) -> list[str]: """Control security options.""" # Disable Seccomp / We don't support it official and it # causes problems on some types of host systems. @@ -565,7 +565,7 @@ class DockerInterface(CoreSysAttributes): Need run inside executor. """ - available_version: List[AwesomeVersion] = [] + available_version: list[AwesomeVersion] = [] try: for image in self.sys_docker.images.list(self.image): for tag in image.tags: diff --git a/supervisor/docker/multicast.py b/supervisor/docker/multicast.py index 44c818392..bbd4bf17b 100644 --- a/supervisor/docker/multicast.py +++ b/supervisor/docker/multicast.py @@ -1,6 +1,5 @@ """HA Cli docker object.""" import logging -from typing import List from ..const import ENV_TIME from ..coresys import CoreSysAttributes @@ -26,7 +25,7 @@ class DockerMulticast(DockerInterface, CoreSysAttributes): return MULTICAST_DOCKER_NAME @property - def capabilities(self) -> List[str]: + def capabilities(self) -> list[str]: """Generate needed capabilities.""" return [Capabilities.NET_ADMIN.value] diff --git a/supervisor/docker/network.py b/supervisor/docker/network.py index cc3d31ee1..0d37043c1 100644 --- a/supervisor/docker/network.py +++ b/supervisor/docker/network.py @@ -2,7 +2,7 @@ from contextlib import suppress from ipaddress import IPv4Address import logging -from typing import List, Optional +from typing import Optional import docker import requests @@ -30,9 +30,9 @@ class DockerNetwork: return DOCKER_NETWORK @property - def containers(self) -> List[docker.models.containers.Container]: + def containers(self) -> list[docker.models.containers.Container]: """Return of connected containers from network.""" - containers: List[docker.models.containers.Container] = [] + containers: list[docker.models.containers.Container] = [] for cid, _ in self.network.attrs.get("Containers", {}).items(): try: containers.append(self.docker.containers.get(cid)) @@ -99,7 +99,7 @@ class DockerNetwork: def attach_container( self, container: docker.models.containers.Container, - alias: Optional[List[str]] = None, + alias: Optional[list[str]] = None, ipv4: Optional[IPv4Address] = None, ) -> None: """Attach container to Supervisor network. diff --git a/supervisor/hardware/data.py b/supervisor/hardware/data.py index d7867a705..ecd230cf1 100644 --- a/supervisor/hardware/data.py +++ b/supervisor/hardware/data.py @@ -2,7 +2,6 @@ from __future__ import annotations from pathlib import Path -from typing import Dict, List, Optional import attr import pyudev @@ -16,10 +15,10 @@ class Device: path: Path = attr.ib(eq=False) sysfs: Path = attr.ib(eq=True) subsystem: str = attr.ib(eq=False) - parent: Optional[Path] = attr.ib(eq=False) - links: List[Path] = attr.ib(eq=False) - attributes: Dict[str, str] = attr.ib(eq=False) - children: List[Path] = attr.ib(eq=False) + parent: Path | None = attr.ib(eq=False) + links: list[Path] = attr.ib(eq=False) + attributes: dict[str, str] = attr.ib(eq=False) + children: list[Path] = attr.ib(eq=False) @property def major(self) -> int: @@ -32,7 +31,7 @@ class Device: return int(self.attributes.get("MINOR", 0)) @property - def by_id(self) -> Optional[Path]: + def by_id(self) -> Path | None: """Return path by-id.""" for link in self.links: if not link.match("/dev/*/by-id/*"): diff --git a/supervisor/hardware/manager.py b/supervisor/hardware/manager.py index 079ca01bc..973718e40 100644 --- a/supervisor/hardware/manager.py +++ b/supervisor/hardware/manager.py @@ -1,7 +1,7 @@ """Hardware Manager of Supervisor.""" import logging from pathlib import Path -from typing import Dict, List, Optional +from typing import Optional import pyudev @@ -24,7 +24,7 @@ class HardwareManager(CoreSysAttributes): def __init__(self, coresys: CoreSys): """Initialize Hardware Monitor object.""" self.coresys: CoreSys = coresys - self._devices: Dict[str, Device] = {} + self._devices: dict[str, Device] = {} self._udev = pyudev.Context() self._montior: HwMonitor = HwMonitor(coresys) @@ -53,7 +53,7 @@ class HardwareManager(CoreSysAttributes): return self._disk @property - def devices(self) -> List[Device]: + def devices(self) -> list[Device]: """Return List of devices.""" return list(self._devices.values()) @@ -66,7 +66,7 @@ class HardwareManager(CoreSysAttributes): return device raise HardwareNotFound() - def filter_devices(self, subsystem: Optional[UdevSubsystem] = None) -> List[Device]: + def filter_devices(self, subsystem: Optional[UdevSubsystem] = None) -> list[Device]: """Return a filtered list.""" devices = set() for device in self.devices: diff --git a/supervisor/hardware/policy.py b/supervisor/hardware/policy.py index dc7da28c8..45042e7da 100644 --- a/supervisor/hardware/policy.py +++ b/supervisor/hardware/policy.py @@ -1,6 +1,5 @@ """Policy / cgroups management of local host.""" import logging -from typing import Dict, List from ..coresys import CoreSys, CoreSysAttributes from .const import PolicyGroup, UdevSubsystem @@ -12,7 +11,7 @@ _LOGGER: logging.Logger = logging.getLogger(__name__) # fmt: off # https://www.kernel.org/doc/Documentation/admin-guide/devices.txt -_CGROUPS: Dict[PolicyGroup, List[int]] = { +_CGROUPS: dict[PolicyGroup, list[int]] = { PolicyGroup.UART: [ 204, # ttyAMA / ttySAC (tty) 188, # ttyUSB (tty) @@ -37,7 +36,7 @@ _CGROUPS: Dict[PolicyGroup, List[int]] = { ] } -_CGROUPS_DYNAMIC_MAJOR: Dict[PolicyGroup, List[UdevSubsystem]] = { +_CGROUPS_DYNAMIC_MAJOR: dict[PolicyGroup, list[UdevSubsystem]] = { PolicyGroup.USB: [ UdevSubsystem.HIDRAW ], @@ -54,7 +53,7 @@ _CGROUPS_DYNAMIC_MAJOR: Dict[PolicyGroup, List[UdevSubsystem]] = { ] } -_CGROUPS_DYNAMIC_MINOR: Dict[PolicyGroup, List[UdevSubsystem]] = { +_CGROUPS_DYNAMIC_MINOR: dict[PolicyGroup, list[UdevSubsystem]] = { PolicyGroup.UART: [ UdevSubsystem.SERIAL ] @@ -74,9 +73,9 @@ class HwPolicy(CoreSysAttributes): """Return true if device is in cgroup Policy.""" return device.major in _CGROUPS.get(group, []) - def get_cgroups_rules(self, group: PolicyGroup) -> List[str]: + def get_cgroups_rules(self, group: PolicyGroup) -> list[str]: """Generate cgroups rules for a policy group.""" - cgroups: List[str] = [f"c {dev}:* rwm" for dev in _CGROUPS[group]] + cgroups: list[str] = [f"c {dev}:* rwm" for dev in _CGROUPS[group]] # Lookup dynamic device groups from host if group in _CGROUPS_DYNAMIC_MAJOR: diff --git a/supervisor/homeassistant/api.py b/supervisor/homeassistant/api.py index 9ae6e2f60..e12184827 100644 --- a/supervisor/homeassistant/api.py +++ b/supervisor/homeassistant/api.py @@ -3,7 +3,7 @@ import asyncio from contextlib import asynccontextmanager, suppress from datetime import datetime, timedelta import logging -from typing import Any, AsyncContextManager, Dict, Optional +from typing import Any, AsyncContextManager, Optional import aiohttp from aiohttp import hdrs @@ -64,12 +64,12 @@ class HomeAssistantAPI(CoreSysAttributes): self, method: str, path: str, - json: Optional[Dict[str, Any]] = None, + json: Optional[dict[str, Any]] = None, content_type: Optional[str] = None, data: Any = None, timeout: int = 30, - params: Optional[Dict[str, str]] = None, - headers: Optional[Dict[str, str]] = None, + params: Optional[dict[str, str]] = None, + headers: Optional[dict[str, str]] = None, ) -> AsyncContextManager[aiohttp.ClientResponse]: """Async context manager to make a request with right auth.""" url = f"{self.sys_homeassistant.api_url}/{path}" @@ -105,7 +105,7 @@ class HomeAssistantAPI(CoreSysAttributes): raise HomeAssistantAPIError() - async def get_config(self) -> Dict[str, Any]: + async def get_config(self) -> dict[str, Any]: """Return Home Assistant config.""" async with self.make_request("get", "api/config") as resp: if resp.status in (200, 201): diff --git a/supervisor/homeassistant/secrets.py b/supervisor/homeassistant/secrets.py index b29d6dda6..b92ecbbb1 100644 --- a/supervisor/homeassistant/secrets.py +++ b/supervisor/homeassistant/secrets.py @@ -2,7 +2,7 @@ from datetime import timedelta import logging from pathlib import Path -from typing import Dict, Optional, Union +from typing import Optional, Union from ..coresys import CoreSys, CoreSysAttributes from ..exceptions import YamlFileError @@ -19,7 +19,7 @@ class HomeAssistantSecrets(CoreSysAttributes): def __init__(self, coresys: CoreSys): """Initialize secret manager.""" self.coresys: CoreSys = coresys - self.secrets: Dict[str, Union[bool, float, int, str]] = {} + self.secrets: dict[str, Union[bool, float, int, str]] = {} @property def path_secrets(self) -> Path: diff --git a/supervisor/homeassistant/websocket.py b/supervisor/homeassistant/websocket.py index c58a5fdfb..aa711e365 100644 --- a/supervisor/homeassistant/websocket.py +++ b/supervisor/homeassistant/websocket.py @@ -1,7 +1,7 @@ """Home Assistant Websocket API.""" import asyncio import logging -from typing import Any, Dict, Optional +from typing import Any, Optional import aiohttp from awesomeversion import AwesomeVersion @@ -30,7 +30,7 @@ class WSClient: self.message_id: int = 0 self._lock: asyncio.Lock = asyncio.Lock() - async def async_send_command(self, message: Dict[str, Any]): + async def async_send_command(self, message: dict[str, Any]): """Send a websocket command.""" async with self._lock: self.message_id += 1 @@ -103,7 +103,7 @@ class HomeAssistantWebSocket(CoreSysAttributes): return client - async def async_send_command(self, message: Dict[str, Any]): + async def async_send_command(self, message: dict[str, Any]): """Send a command with the WS client.""" if self.sys_core.state in CLOSING_STATES: raise HomeAssistantWSNotSupported( @@ -136,7 +136,7 @@ class HomeAssistantWebSocket(CoreSysAttributes): raise HomeAssistantWSError from err async def async_supervisor_update_event( - self, key: str, data: Optional[Dict[str, Any]] = None + self, key: str, data: Optional[dict[str, Any]] = None ): """Send a supervisor/event command.""" try: @@ -155,13 +155,13 @@ class HomeAssistantWebSocket(CoreSysAttributes): except HomeAssistantWSError as err: _LOGGER.error(err) - def supervisor_update_event(self, key: str, data: Optional[Dict[str, Any]] = None): + def supervisor_update_event(self, key: str, data: Optional[dict[str, Any]] = None): """Send a supervisor/event command.""" if self.sys_core.state in CLOSING_STATES: return self.sys_create_task(self.async_supervisor_update_event(key, data)) - def send_command(self, message: Dict[str, Any]): + def send_command(self, message: dict[str, Any]): """Send a supervisor/event command.""" if self.sys_core.state in CLOSING_STATES: return diff --git a/supervisor/host/manager.py b/supervisor/host/manager.py index 52e1b2ed6..64ce046f6 100644 --- a/supervisor/host/manager.py +++ b/supervisor/host/manager.py @@ -2,7 +2,6 @@ from contextlib import suppress from functools import lru_cache import logging -from typing import List from ..const import BusEvent from ..coresys import CoreSys, CoreSysAttributes @@ -65,12 +64,12 @@ class HostManager(CoreSysAttributes): return self._sound @property - def features(self) -> List[HostFeature]: + def features(self) -> list[HostFeature]: """Return a list of host features.""" return self.supported_features() @lru_cache - def supported_features(self) -> List[HostFeature]: + def supported_features(self) -> list[HostFeature]: """Return a list of supported host features.""" features = [] diff --git a/supervisor/host/network.py b/supervisor/host/network.py index d90409e44..bd2047edf 100644 --- a/supervisor/host/network.py +++ b/supervisor/host/network.py @@ -4,7 +4,6 @@ from __future__ import annotations import asyncio from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface import logging -from typing import List, Optional, Union import attr @@ -40,10 +39,10 @@ class NetworkManager(CoreSysAttributes): def __init__(self, coresys: CoreSys): """Initialize system center handling.""" self.coresys: CoreSys = coresys - self._connectivity: Optional[bool] = None + self._connectivity: bool | None = None @property - def connectivity(self) -> Optional[bool]: + def connectivity(self) -> bool | None: """Return true current connectivity state.""" return self._connectivity @@ -58,19 +57,19 @@ class NetworkManager(CoreSysAttributes): ) @property - def interfaces(self) -> List[Interface]: + def interfaces(self) -> list[Interface]: """Return a dictionary of active interfaces.""" - interfaces: List[Interface] = [] + interfaces: list[Interface] = [] for inet in self.sys_dbus.network.interfaces.values(): interfaces.append(Interface.from_dbus_interface(inet)) return interfaces @property - def dns_servers(self) -> List[str]: + def dns_servers(self) -> list[str]: """Return a list of local DNS servers.""" # Read all local dns servers - servers: List[str] = [] + servers: list[str] = [] for config in self.sys_dbus.network.dns.configuration: if config.vpn or not config.nameservers: continue @@ -183,7 +182,7 @@ class NetworkManager(CoreSysAttributes): ) await self.update() - async def scan_wifi(self, interface: Interface) -> List[AccessPoint]: + async def scan_wifi(self, interface: Interface) -> list[AccessPoint]: """Scan on Interface for AccessPoint.""" inet = self.sys_dbus.network.interfaces.get(interface.name) @@ -202,7 +201,7 @@ class NetworkManager(CoreSysAttributes): await asyncio.sleep(5) # Process AP - accesspoints: List[AccessPoint] = [] + accesspoints: list[AccessPoint] = [] for ap_object in (await inet.wireless.get_all_accesspoints())[0]: accesspoint = NetworkWirelessAP(ap_object) @@ -241,9 +240,9 @@ class IpConfig: """Represent a IP configuration.""" method: InterfaceMethod = attr.ib() - address: List[Union[IPv4Interface, IPv6Interface]] = attr.ib() - gateway: Optional[Union[IPv4Address, IPv6Address]] = attr.ib() - nameservers: List[Union[IPv4Address, IPv6Address]] = attr.ib() + address: list[IPv4Interface | IPv6Interface] = attr.ib() + gateway: IPv4Address | IPv6Address | None = attr.ib() + nameservers: list[IPv4Address | IPv6Address] = attr.ib() @attr.s(slots=True) @@ -253,8 +252,8 @@ class WifiConfig: mode: WifiMode = attr.ib() ssid: str = attr.ib() auth: AuthMethod = attr.ib() - psk: Optional[str] = attr.ib() - signal: Optional[int] = attr.ib() + psk: str | None = attr.ib() + signal: int | None = attr.ib() @attr.s(slots=True) @@ -274,10 +273,10 @@ class Interface: connected: bool = attr.ib() primary: bool = attr.ib() type: InterfaceType = attr.ib() - ipv4: Optional[IpConfig] = attr.ib() - ipv6: Optional[IpConfig] = attr.ib() - wifi: Optional[WifiConfig] = attr.ib() - vlan: Optional[VlanConfig] = attr.ib() + ipv4: IpConfig | None = attr.ib() + ipv6: IpConfig | None = attr.ib() + wifi: WifiConfig | None = attr.ib() + vlan: VlanConfig | None = attr.ib() @staticmethod def from_dbus_interface(inet: NetworkInterface) -> Interface: @@ -320,7 +319,7 @@ class Interface: return mapping.get(method, InterfaceMethod.DISABLED) @staticmethod - def _map_nm_connected(connection: Optional[NetworkConnection]) -> bool: + def _map_nm_connected(connection: NetworkConnection | None) -> bool: """Map connectivity state.""" if not connection: return False @@ -340,7 +339,7 @@ class Interface: return mapping[device_type] @staticmethod - def _map_nm_wifi(inet: NetworkInterface) -> Optional[WifiConfig]: + def _map_nm_wifi(inet: NetworkInterface) -> WifiConfig | None: """Create mapping to nm wifi property.""" if inet.type != DeviceType.WIRELESS or not inet.settings: return None @@ -376,7 +375,7 @@ class Interface: ) @staticmethod - def _map_nm_vlan(inet: NetworkInterface) -> Optional[WifiConfig]: + def _map_nm_vlan(inet: NetworkInterface) -> WifiConfig | None: """Create mapping to nm vlan property.""" if inet.type != DeviceType.VLAN or not inet.settings: return None diff --git a/supervisor/host/sound.py b/supervisor/host/sound.py index 9c4748d09..adfd66e64 100644 --- a/supervisor/host/sound.py +++ b/supervisor/host/sound.py @@ -2,7 +2,7 @@ from datetime import timedelta from enum import Enum import logging -from typing import List, Optional +from typing import Optional import attr from pulsectl import Pulse, PulseError, PulseIndexError, PulseOperationFailed @@ -48,7 +48,7 @@ class AudioStream: mute: bool = attr.ib() default: bool = attr.ib() card: Optional[int] = attr.ib() - applications: List[AudioApplication] = attr.ib() + applications: list[AudioApplication] = attr.ib() @attr.s(frozen=True) @@ -67,7 +67,7 @@ class SoundCard: name: str = attr.ib() index: int = attr.ib() driver: str = attr.ib() - profiles: List[SoundProfile] = attr.ib() + profiles: list[SoundProfile] = attr.ib() class SoundControl(CoreSysAttributes): @@ -76,28 +76,28 @@ class SoundControl(CoreSysAttributes): def __init__(self, coresys: CoreSys) -> None: """Initialize PulseAudio sound control.""" self.coresys: CoreSys = coresys - self._cards: List[SoundCard] = [] - self._inputs: List[AudioStream] = [] - self._outputs: List[AudioStream] = [] - self._applications: List[AudioApplication] = [] + self._cards: list[SoundCard] = [] + self._inputs: list[AudioStream] = [] + self._outputs: list[AudioStream] = [] + self._applications: list[AudioApplication] = [] @property - def cards(self) -> List[SoundCard]: + def cards(self) -> list[SoundCard]: """Return a list of available sound cards and profiles.""" return self._cards @property - def inputs(self) -> List[AudioStream]: + def inputs(self) -> list[AudioStream]: """Return a list of available input streams.""" return self._inputs @property - def outputs(self) -> List[AudioStream]: + def outputs(self) -> list[AudioStream]: """Return a list of available output streams.""" return self._outputs @property - def applications(self) -> List[AudioApplication]: + def applications(self) -> list[AudioApplication]: """Return a list of available application streams.""" return self._applications @@ -311,7 +311,7 @@ class SoundControl(CoreSysAttributes): # Update Sound Card self._cards.clear() for card in pulse.card_list(): - sound_profiles: List[SoundProfile] = [] + sound_profiles: list[SoundProfile] = [] # Generate profiles for profile in card.profile_list: diff --git a/supervisor/ingress.py b/supervisor/ingress.py index a5279c71e..62e93151d 100644 --- a/supervisor/ingress.py +++ b/supervisor/ingress.py @@ -3,7 +3,7 @@ from datetime import timedelta import logging import random import secrets -from typing import Dict, List, Optional +from typing import Optional from .addons.addon import Addon from .const import ATTR_PORTS, ATTR_SESSION, FILE_HASSIO_INGRESS @@ -23,7 +23,7 @@ class Ingress(FileConfiguration, CoreSysAttributes): """Initialize updater.""" super().__init__(FILE_HASSIO_INGRESS, SCHEMA_INGRESS_CONFIG) self.coresys: CoreSys = coresys - self.tokens: Dict[str, str] = {} + self.tokens: dict[str, str] = {} def get(self, token: str) -> Optional[Addon]: """Return addon they have this ingress token.""" @@ -32,17 +32,17 @@ class Ingress(FileConfiguration, CoreSysAttributes): return self.sys_addons.get(self.tokens[token], local_only=True) @property - def sessions(self) -> Dict[str, float]: + def sessions(self) -> dict[str, float]: """Return sessions.""" return self._data[ATTR_SESSION] @property - def ports(self) -> Dict[str, int]: + def ports(self) -> dict[str, int]: """Return list of dynamic ports.""" return self._data[ATTR_PORTS] @property - def addons(self) -> List[Addon]: + def addons(self) -> list[Addon]: """Return list of ingress Add-ons.""" addons = [] for addon in self.sys_addons.installed: diff --git a/supervisor/jobs/__init__.py b/supervisor/jobs/__init__.py index 8a2e40afa..3157c67bb 100644 --- a/supervisor/jobs/__init__.py +++ b/supervisor/jobs/__init__.py @@ -1,6 +1,6 @@ """Supervisor job manager.""" import logging -from typing import Dict, List, Optional +from typing import Optional from ..coresys import CoreSys, CoreSysAttributes from ..utils.common import FileConfiguration @@ -56,20 +56,20 @@ class JobManager(FileConfiguration, CoreSysAttributes): """Initialize the JobManager class.""" super().__init__(FILE_CONFIG_JOBS, SCHEMA_JOBS_CONFIG) self.coresys: CoreSys = coresys - self._jobs: Dict[str, SupervisorJob] = {} + self._jobs: dict[str, SupervisorJob] = {} @property - def jobs(self) -> List[SupervisorJob]: + def jobs(self) -> list[SupervisorJob]: """Return a list of current jobs.""" return self._jobs @property - def ignore_conditions(self) -> List[JobCondition]: + def ignore_conditions(self) -> list[JobCondition]: """Return a list of ingore condition.""" return self._data[ATTR_IGNORE_CONDITIONS] @ignore_conditions.setter - def ignore_conditions(self, value: List[JobCondition]) -> None: + def ignore_conditions(self, value: list[JobCondition]) -> None: """Set a list of ignored condition.""" self._data[ATTR_IGNORE_CONDITIONS] = value diff --git a/supervisor/jobs/decorator.py b/supervisor/jobs/decorator.py index ca1466b89..bc610c2f8 100644 --- a/supervisor/jobs/decorator.py +++ b/supervisor/jobs/decorator.py @@ -3,7 +3,7 @@ import asyncio from datetime import datetime, timedelta from functools import wraps import logging -from typing import Any, List, Optional, Tuple +from typing import Any, Optional import sentry_sdk @@ -23,7 +23,7 @@ class Job(CoreSysAttributes): def __init__( self, name: Optional[str] = None, - conditions: Optional[List[JobCondition]] = None, + conditions: Optional[list[JobCondition]] = None, cleanup: bool = True, on_condition: Optional[JobException] = None, limit: Optional[JobExecutionLimit] = None, @@ -47,7 +47,7 @@ class Job(CoreSysAttributes): ): raise RuntimeError("Using Job without a Throttle period!") - def _post_init(self, args: Tuple[Any]) -> None: + def _post_init(self, args: tuple[Any]) -> None: """Runtime init.""" if self.name is None: self.name = str(self._method.__qualname__).lower().replace(".", "_") diff --git a/supervisor/misc/scheduler.py b/supervisor/misc/scheduler.py index 5a29bffe2..ef1803b2b 100644 --- a/supervisor/misc/scheduler.py +++ b/supervisor/misc/scheduler.py @@ -2,7 +2,7 @@ import asyncio from datetime import date, datetime, time, timedelta import logging -from typing import Awaitable, Callable, List, Optional, Union +from typing import Awaitable, Callable, Optional, Union from uuid import UUID, uuid4 import async_timeout @@ -32,7 +32,7 @@ class Scheduler(CoreSysAttributes): def __init__(self, coresys: CoreSys): """Initialize task schedule.""" self.coresys: CoreSys = coresys - self._tasks: List[_Task] = [] + self._tasks: list[_Task] = [] def register_task( self, @@ -96,7 +96,7 @@ class Scheduler(CoreSysAttributes): async def shutdown(self, timeout=10) -> None: """Shutdown all task inside the scheduler.""" - running: List[asyncio.tasks.Task] = [] + running: list[asyncio.tasks.Task] = [] # Cancel next task / get running list _LOGGER.info("Shutting down scheduled tasks") diff --git a/supervisor/os/data_disk.py b/supervisor/os/data_disk.py index 6299babe1..31e30a574 100644 --- a/supervisor/os/data_disk.py +++ b/supervisor/os/data_disk.py @@ -1,7 +1,7 @@ """Home Assistant Operating-System DataDisk.""" import logging from pathlib import Path -from typing import List, Optional +from typing import Optional from awesomeversion import AwesomeVersion @@ -34,9 +34,9 @@ class DataDisk(CoreSysAttributes): return self.sys_dbus.agent.datadisk.current_device @property - def available_disks(self) -> List[Path]: + def available_disks(self) -> list[Path]: """Return a list of possible new disk locations.""" - device_paths: List[Path] = [] + device_paths: list[Path] = [] for device in self.sys_hardware.devices: # Filter devices out which can't be a target if ( diff --git a/supervisor/plugins/dns.py b/supervisor/plugins/dns.py index f31b62f94..64cdf6864 100644 --- a/supervisor/plugins/dns.py +++ b/supervisor/plugins/dns.py @@ -7,7 +7,7 @@ from contextlib import suppress from ipaddress import IPv4Address import logging from pathlib import Path -from typing import List, Optional +from typing import Optional import attr from awesomeversion import AwesomeVersion @@ -43,7 +43,7 @@ class HostEntry: """Single entry in hosts.""" ip_address: IPv4Address = attr.ib() - names: List[str] = attr.ib() + names: list[str] = attr.ib() class PluginDns(PluginBase): @@ -58,7 +58,7 @@ class PluginDns(PluginBase): self.resolv_template: Optional[jinja2.Template] = None self.hosts_template: Optional[jinja2.Template] = None - self._hosts: List[HostEntry] = [] + self._hosts: list[HostEntry] = [] self._loop: bool = False @property @@ -72,9 +72,9 @@ class PluginDns(PluginBase): return Path(self.sys_config.path_dns, "coredns.json") @property - def locals(self) -> List[str]: + def locals(self) -> list[str]: """Return list of local system DNS servers.""" - servers: List[str] = [] + servers: list[str] = [] for server in [ f"dns://{server!s}" for server in self.sys_host.network.dns_servers ]: @@ -84,12 +84,12 @@ class PluginDns(PluginBase): return servers @property - def servers(self) -> List[str]: + def servers(self) -> list[str]: """Return list of DNS servers.""" return self._data[ATTR_SERVERS] @servers.setter - def servers(self, value: List[str]) -> None: + def servers(self, value: list[str]) -> None: """Return list of DNS servers.""" self._data[ATTR_SERVERS] = value @@ -259,8 +259,8 @@ class PluginDns(PluginBase): def _write_config(self) -> None: """Write CoreDNS config.""" debug: bool = self.sys_config.logging == LogLevel.DEBUG - dns_servers: List[str] = [] - dns_locals: List[str] = [] + dns_servers: list[str] = [] + dns_locals: list[str] = [] # Prepare DNS serverlist: Prio 1 Manual, Prio 2 Local, Prio 3 Fallback if not self._loop: @@ -317,12 +317,12 @@ class PluginDns(PluginBase): _LOGGER.error("Can't update hosts: %s", err) raise CoreDNSError() from err - def add_host(self, ipv4: IPv4Address, names: List[str], write: bool = True) -> None: + def add_host(self, ipv4: IPv4Address, names: list[str], write: bool = True) -> None: """Add a new host entry.""" if not ipv4 or ipv4 == IPv4Address("0.0.0.0"): return - hostnames: List[str] = [] + hostnames: list[str] = [] for name in names: hostnames.append(name) hostnames.append(f"{name}.{DNS_SUFFIX}") @@ -355,7 +355,7 @@ class PluginDns(PluginBase): if write: self.write_hosts() - def _search_host(self, names: List[str]) -> Optional[HostEntry]: + def _search_host(self, names: list[str]) -> Optional[HostEntry]: """Search a host entry.""" for entry in self._hosts: for name in names: diff --git a/supervisor/resolution/check.py b/supervisor/resolution/check.py index f8425c5c9..d1bad2b78 100644 --- a/supervisor/resolution/check.py +++ b/supervisor/resolution/check.py @@ -1,7 +1,7 @@ """Helpers to checks the system.""" from importlib import import_module import logging -from typing import Any, Dict, List +from typing import Any from ..const import ATTR_CHECKS from ..coresys import CoreSys, CoreSysAttributes @@ -18,17 +18,17 @@ class ResolutionCheck(CoreSysAttributes): def __init__(self, coresys: CoreSys) -> None: """Initialize the checks class.""" self.coresys = coresys - self._checks: Dict[str, CheckBase] = {} + self._checks: dict[str, CheckBase] = {} self._load() @property - def data(self) -> Dict[str, Any]: + def data(self) -> dict[str, Any]: """Return data.""" return self.sys_resolution.data[ATTR_CHECKS] @property - def all_checks(self) -> List[CheckBase]: + def all_checks(self) -> list[CheckBase]: """Return all list of all checks.""" return list(self._checks.values()) diff --git a/supervisor/resolution/checks/addon_pwned.py b/supervisor/resolution/checks/addon_pwned.py index 9d69b30e0..32fd2d4ac 100644 --- a/supervisor/resolution/checks/addon_pwned.py +++ b/supervisor/resolution/checks/addon_pwned.py @@ -1,7 +1,7 @@ """Helpers to check core security.""" from datetime import timedelta import logging -from typing import List, Optional +from typing import Optional from ...const import AddonState, CoreState from ...coresys import CoreSys @@ -99,6 +99,6 @@ class CheckAddonPwned(CheckBase): return ContextType.ADDON @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this check can run.""" return [CoreState.RUNNING] diff --git a/supervisor/resolution/checks/base.py b/supervisor/resolution/checks/base.py index f9743735d..acfc0f079 100644 --- a/supervisor/resolution/checks/base.py +++ b/supervisor/resolution/checks/base.py @@ -1,7 +1,7 @@ """Baseclass for system checks.""" from abc import ABC, abstractmethod import logging -from typing import List, Optional +from typing import Optional from ...const import ATTR_ENABLED, CoreState from ...coresys import CoreSys, CoreSysAttributes @@ -70,7 +70,7 @@ class CheckBase(ABC, CoreSysAttributes): """Return a ContextType enum.""" @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this check can run.""" return [] diff --git a/supervisor/resolution/checks/core_security.py b/supervisor/resolution/checks/core_security.py index b488c1c38..e6740c6fa 100644 --- a/supervisor/resolution/checks/core_security.py +++ b/supervisor/resolution/checks/core_security.py @@ -1,7 +1,7 @@ """Helpers to check core security.""" from enum import Enum from pathlib import Path -from typing import List, Optional +from typing import Optional from awesomeversion import AwesomeVersion, AwesomeVersionException @@ -64,6 +64,6 @@ class CheckCoreSecurity(CheckBase): return ContextType.CORE @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this check can run.""" return [CoreState.RUNNING, CoreState.STARTUP] diff --git a/supervisor/resolution/checks/core_trust.py b/supervisor/resolution/checks/core_trust.py index 075187526..9f636f2fe 100644 --- a/supervisor/resolution/checks/core_trust.py +++ b/supervisor/resolution/checks/core_trust.py @@ -1,6 +1,6 @@ """Helpers to check core trust.""" import logging -from typing import List, Optional +from typing import Optional from ...const import CoreState from ...coresys import CoreSys @@ -54,6 +54,6 @@ class CheckCoreTrust(CheckBase): return ContextType.CORE @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this check can run.""" return [CoreState.RUNNING, CoreState.STARTUP] diff --git a/supervisor/resolution/checks/free_space.py b/supervisor/resolution/checks/free_space.py index b46f2a64c..387ee14a9 100644 --- a/supervisor/resolution/checks/free_space.py +++ b/supervisor/resolution/checks/free_space.py @@ -1,5 +1,5 @@ """Helpers to check and fix issues with free space.""" -from typing import List, Optional +from typing import Optional from ...backups.const import BackupType from ...const import CoreState @@ -33,7 +33,7 @@ class CheckFreeSpace(CheckBase): ) return - suggestions: List[SuggestionType] = [] + suggestions: list[SuggestionType] = [] if ( len( [ @@ -67,6 +67,6 @@ class CheckFreeSpace(CheckBase): return ContextType.SYSTEM @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this check can run.""" return [CoreState.RUNNING, CoreState.STARTUP] diff --git a/supervisor/resolution/checks/plugin_trust.py b/supervisor/resolution/checks/plugin_trust.py index f4487f6fb..25f61c3db 100644 --- a/supervisor/resolution/checks/plugin_trust.py +++ b/supervisor/resolution/checks/plugin_trust.py @@ -1,6 +1,6 @@ """Helpers to check plugin trust.""" import logging -from typing import List, Optional +from typing import Optional from ...const import CoreState from ...coresys import CoreSys @@ -72,6 +72,6 @@ class CheckPluginTrust(CheckBase): return ContextType.PLUGIN @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this check can run.""" return [CoreState.RUNNING, CoreState.STARTUP] diff --git a/supervisor/resolution/checks/supervisor_trust.py b/supervisor/resolution/checks/supervisor_trust.py index eff2f6188..427760f0e 100644 --- a/supervisor/resolution/checks/supervisor_trust.py +++ b/supervisor/resolution/checks/supervisor_trust.py @@ -1,6 +1,6 @@ """Helpers to check supervisor trust.""" import logging -from typing import List, Optional +from typing import Optional from ...const import CoreState from ...coresys import CoreSys @@ -54,6 +54,6 @@ class CheckSupervisorTrust(CheckBase): return ContextType.SUPERVISOR @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this check can run.""" return [CoreState.RUNNING, CoreState.STARTUP] diff --git a/supervisor/resolution/evaluate.py b/supervisor/resolution/evaluate.py index 1509762fe..c6b180413 100644 --- a/supervisor/resolution/evaluate.py +++ b/supervisor/resolution/evaluate.py @@ -1,7 +1,6 @@ """Helpers to evaluate the system.""" from importlib import import_module import logging -from typing import Dict, List, Set from ..coresys import CoreSys, CoreSysAttributes from ..exceptions import ResolutionNotFound @@ -25,13 +24,13 @@ class ResolutionEvaluation(CoreSysAttributes): def __init__(self, coresys: CoreSys) -> None: """Initialize the evaluation class.""" self.coresys = coresys - self.cached_images: Set[str] = set() - self._evalutions: Dict[str, EvaluateBase] = {} + self.cached_images: set[str] = set() + self._evalutions: dict[str, EvaluateBase] = {} self._load() @property - def all_evaluations(self) -> List[EvaluateBase]: + def all_evaluations(self) -> list[EvaluateBase]: """Return all list of all checks.""" return list(self._evalutions.values()) diff --git a/supervisor/resolution/evaluations/apparmor.py b/supervisor/resolution/evaluations/apparmor.py index 86e58b62d..496c59376 100644 --- a/supervisor/resolution/evaluations/apparmor.py +++ b/supervisor/resolution/evaluations/apparmor.py @@ -1,6 +1,5 @@ """Evaluation class for AppArmor.""" from pathlib import Path -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -29,7 +28,7 @@ class EvaluateAppArmor(EvaluateBase): return "AppArmor is required for Home Assistant." @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.INITIALIZE] diff --git a/supervisor/resolution/evaluations/base.py b/supervisor/resolution/evaluations/base.py index db1c374d7..7c633857a 100644 --- a/supervisor/resolution/evaluations/base.py +++ b/supervisor/resolution/evaluations/base.py @@ -1,7 +1,6 @@ """Baseclass for system evaluations.""" from abc import ABC, abstractmethod import logging -from typing import List from ...const import CoreState from ...coresys import CoreSys, CoreSysAttributes @@ -54,6 +53,6 @@ class EvaluateBase(ABC, CoreSysAttributes): """Return a string that is printed when self.evaluate is False.""" @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [] diff --git a/supervisor/resolution/evaluations/container.py b/supervisor/resolution/evaluations/container.py index c1dffb878..c4a23eb2c 100644 --- a/supervisor/resolution/evaluations/container.py +++ b/supervisor/resolution/evaluations/container.py @@ -1,6 +1,6 @@ """Evaluation class for container.""" import logging -from typing import Any, List +from typing import Any from docker.errors import DockerException from requests import RequestException @@ -43,7 +43,7 @@ class EvaluateContainer(EvaluateBase): return f"Found images: {self._images} which are not supported, remove these from the host!" @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.SETUP, CoreState.RUNNING, CoreState.INITIALIZE] @@ -68,7 +68,7 @@ class EvaluateContainer(EvaluateBase): self._images.add(image_name) return len(self._images) != 0 - def _get_images(self) -> List[Any]: + def _get_images(self) -> list[Any]: """Return a list of images.""" images = [] diff --git a/supervisor/resolution/evaluations/content_trust.py b/supervisor/resolution/evaluations/content_trust.py index 2a9fb4538..eb1f4f78d 100644 --- a/supervisor/resolution/evaluations/content_trust.py +++ b/supervisor/resolution/evaluations/content_trust.py @@ -1,5 +1,4 @@ """Evaluation class for Content Trust.""" -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -26,7 +25,7 @@ class EvaluateContentTrust(EvaluateBase): return "System run with disabled trusted content security." @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.INITIALIZE, CoreState.SETUP, CoreState.RUNNING] diff --git a/supervisor/resolution/evaluations/dbus.py b/supervisor/resolution/evaluations/dbus.py index 5ca70e2fc..6dc612bcf 100644 --- a/supervisor/resolution/evaluations/dbus.py +++ b/supervisor/resolution/evaluations/dbus.py @@ -1,5 +1,4 @@ """Evaluation class for dbus.""" -from typing import List from ...const import SOCKET_DBUS, CoreState from ...coresys import CoreSys @@ -26,7 +25,7 @@ class EvaluateDbus(EvaluateBase): return "D-Bus is required for Home Assistant." @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.INITIALIZE] diff --git a/supervisor/resolution/evaluations/docker_configuration.py b/supervisor/resolution/evaluations/docker_configuration.py index cf7350cfd..0aa328ebb 100644 --- a/supervisor/resolution/evaluations/docker_configuration.py +++ b/supervisor/resolution/evaluations/docker_configuration.py @@ -1,6 +1,5 @@ """Evaluation class for docker configuration.""" import logging -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -32,7 +31,7 @@ class EvaluateDockerConfiguration(EvaluateBase): return "The configuration of Docker is not supported" @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.INITIALIZE] diff --git a/supervisor/resolution/evaluations/docker_version.py b/supervisor/resolution/evaluations/docker_version.py index 5f5c404cb..afe75f959 100644 --- a/supervisor/resolution/evaluations/docker_version.py +++ b/supervisor/resolution/evaluations/docker_version.py @@ -1,5 +1,4 @@ """Evaluation class for docker version.""" -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -26,7 +25,7 @@ class EvaluateDockerVersion(EvaluateBase): return f"Docker version '{self.sys_docker.info.version}' is not supported by the Supervisor!" @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.INITIALIZE] diff --git a/supervisor/resolution/evaluations/job_conditions.py b/supervisor/resolution/evaluations/job_conditions.py index 9599dce04..b688e5c28 100644 --- a/supervisor/resolution/evaluations/job_conditions.py +++ b/supervisor/resolution/evaluations/job_conditions.py @@ -1,5 +1,4 @@ """Evaluation class for Job Conditions.""" -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -26,7 +25,7 @@ class EvaluateJobConditions(EvaluateBase): return "Found unsupported job conditions settings." @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.INITIALIZE, CoreState.SETUP, CoreState.RUNNING] diff --git a/supervisor/resolution/evaluations/lxc.py b/supervisor/resolution/evaluations/lxc.py index ca68944a0..f18ac966a 100644 --- a/supervisor/resolution/evaluations/lxc.py +++ b/supervisor/resolution/evaluations/lxc.py @@ -1,7 +1,6 @@ """Evaluation class for lxc.""" from contextlib import suppress from pathlib import Path -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -28,7 +27,7 @@ class EvaluateLxc(EvaluateBase): return "Detected Docker running inside LXC." @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.INITIALIZE] diff --git a/supervisor/resolution/evaluations/network_manager.py b/supervisor/resolution/evaluations/network_manager.py index 8c92ec7a5..fb2e509e0 100644 --- a/supervisor/resolution/evaluations/network_manager.py +++ b/supervisor/resolution/evaluations/network_manager.py @@ -1,5 +1,4 @@ """Evaluation class for network manager.""" -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -27,7 +26,7 @@ class EvaluateNetworkManager(EvaluateBase): return "NetworkManager is not correctly configured" @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.SETUP, CoreState.RUNNING] diff --git a/supervisor/resolution/evaluations/operating_system.py b/supervisor/resolution/evaluations/operating_system.py index 80ff781a2..429bb6efc 100644 --- a/supervisor/resolution/evaluations/operating_system.py +++ b/supervisor/resolution/evaluations/operating_system.py @@ -1,5 +1,4 @@ """Evaluation class for operating system.""" -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -28,7 +27,7 @@ class EvaluateOperatingSystem(EvaluateBase): return f"Detected unsupported OS: {self.sys_host.info.operating_system}" @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.SETUP] diff --git a/supervisor/resolution/evaluations/privileged.py b/supervisor/resolution/evaluations/privileged.py index 520cdf09c..8c1252034 100644 --- a/supervisor/resolution/evaluations/privileged.py +++ b/supervisor/resolution/evaluations/privileged.py @@ -1,5 +1,4 @@ """Evaluation class for privileged.""" -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -26,7 +25,7 @@ class EvaluatePrivileged(EvaluateBase): return "Supervisor does not run in Privileged mode." @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.INITIALIZE] diff --git a/supervisor/resolution/evaluations/source_mods.py b/supervisor/resolution/evaluations/source_mods.py index ceffb8ba8..b8ef574bc 100644 --- a/supervisor/resolution/evaluations/source_mods.py +++ b/supervisor/resolution/evaluations/source_mods.py @@ -1,7 +1,6 @@ """Evaluation class for Content Trust.""" import logging from pathlib import Path -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -32,7 +31,7 @@ class EvaluateSourceMods(EvaluateBase): return "System detect unauthorized source code modifications." @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.RUNNING] diff --git a/supervisor/resolution/evaluations/systemd.py b/supervisor/resolution/evaluations/systemd.py index 2501c9232..e0db4c43d 100644 --- a/supervisor/resolution/evaluations/systemd.py +++ b/supervisor/resolution/evaluations/systemd.py @@ -1,5 +1,4 @@ """Evaluation class for systemd.""" -from typing import List from ...const import CoreState from ...coresys import CoreSys @@ -27,7 +26,7 @@ class EvaluateSystemd(EvaluateBase): return "Systemd is not correctly working" @property - def states(self) -> List[CoreState]: + def states(self) -> list[CoreState]: """Return a list of valid states when this evaluation can run.""" return [CoreState.SETUP] diff --git a/supervisor/resolution/fixup.py b/supervisor/resolution/fixup.py index 71788ef6a..9c09d206e 100644 --- a/supervisor/resolution/fixup.py +++ b/supervisor/resolution/fixup.py @@ -1,6 +1,5 @@ """Helpers to fixup the system.""" import logging -from typing import List from ..coresys import CoreSys, CoreSysAttributes from ..jobs.const import JobCondition @@ -30,7 +29,7 @@ class ResolutionFixup(CoreSysAttributes): self._store_execute_remove = FixupStoreExecuteRemove(coresys) @property - def all_fixes(self) -> List[FixupBase]: + def all_fixes(self) -> list[FixupBase]: """Return a list of all fixups. Order can be important! diff --git a/supervisor/resolution/fixups/base.py b/supervisor/resolution/fixups/base.py index 893b0e722..39a1e8b13 100644 --- a/supervisor/resolution/fixups/base.py +++ b/supervisor/resolution/fixups/base.py @@ -1,7 +1,7 @@ """Baseclass for system fixup.""" from abc import ABC, abstractmethod import logging -from typing import List, Optional +from typing import Optional from ...coresys import CoreSys, CoreSysAttributes from ...exceptions import ResolutionFixupError @@ -63,7 +63,7 @@ class FixupBase(ABC, CoreSysAttributes): """Return a ContextType enum.""" @property - def issues(self) -> List[IssueType]: + def issues(self) -> list[IssueType]: """Return a IssueType enum list.""" return [] diff --git a/supervisor/resolution/fixups/clear_full_backup.py b/supervisor/resolution/fixups/clear_full_backup.py index 4754dce2b..543f65ceb 100644 --- a/supervisor/resolution/fixups/clear_full_backup.py +++ b/supervisor/resolution/fixups/clear_full_backup.py @@ -1,6 +1,6 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import List, Optional +from typing import Optional from ...backups.const import BackupType from ..const import MINIMUM_FULL_BACKUPS, ContextType, IssueType, SuggestionType @@ -36,6 +36,6 @@ class FixupClearFullBackup(FixupBase): return ContextType.SYSTEM @property - def issues(self) -> List[IssueType]: + def issues(self) -> list[IssueType]: """Return a IssueType enum list.""" return [IssueType.FREE_SPACE] diff --git a/supervisor/resolution/fixups/store_execute_reload.py b/supervisor/resolution/fixups/store_execute_reload.py index a73f1e014..0c893ee28 100644 --- a/supervisor/resolution/fixups/store_execute_reload.py +++ b/supervisor/resolution/fixups/store_execute_reload.py @@ -1,6 +1,6 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import List, Optional +from typing import Optional from ...exceptions import ( ResolutionFixupError, @@ -50,7 +50,7 @@ class FixupStoreExecuteReload(FixupBase): return ContextType.STORE @property - def issues(self) -> List[IssueType]: + def issues(self) -> list[IssueType]: """Return a IssueType enum list.""" return [IssueType.FATAL_ERROR] diff --git a/supervisor/resolution/fixups/store_execute_remove.py b/supervisor/resolution/fixups/store_execute_remove.py index 2dcf3524b..e9e4a2826 100644 --- a/supervisor/resolution/fixups/store_execute_remove.py +++ b/supervisor/resolution/fixups/store_execute_remove.py @@ -1,6 +1,6 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import List, Optional +from typing import Optional from supervisor.exceptions import ResolutionFixupError, StoreError, StoreNotFound @@ -44,7 +44,7 @@ class FixupStoreExecuteRemove(FixupBase): return ContextType.STORE @property - def issues(self) -> List[IssueType]: + def issues(self) -> list[IssueType]: """Return a IssueType enum list.""" return [IssueType.CORRUPT_REPOSITORY] diff --git a/supervisor/resolution/fixups/store_execute_reset.py b/supervisor/resolution/fixups/store_execute_reset.py index 89fb8b843..1b6daad9c 100644 --- a/supervisor/resolution/fixups/store_execute_reset.py +++ b/supervisor/resolution/fixups/store_execute_reset.py @@ -1,6 +1,6 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import List, Optional +from typing import Optional from ...exceptions import ( ResolutionFixupError, @@ -52,7 +52,7 @@ class FixupStoreExecuteReset(FixupBase): return ContextType.STORE @property - def issues(self) -> List[IssueType]: + def issues(self) -> list[IssueType]: """Return a IssueType enum list.""" return [IssueType.CORRUPT_REPOSITORY, IssueType.FATAL_ERROR] diff --git a/supervisor/resolution/module.py b/supervisor/resolution/module.py index 09ff51359..0a6fc2dff 100644 --- a/supervisor/resolution/module.py +++ b/supervisor/resolution/module.py @@ -1,6 +1,6 @@ """Supervisor resolution center.""" import logging -from typing import Any, Dict, List, Optional +from typing import Any, Optional from ..coresys import CoreSys, CoreSysAttributes from ..exceptions import ResolutionError, ResolutionNotFound @@ -37,13 +37,13 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes): self._fixup = ResolutionFixup(coresys) self._notify = ResolutionNotify(coresys) - self._suggestions: List[Suggestion] = [] - self._issues: List[Issue] = [] - self._unsupported: List[UnsupportedReason] = [] - self._unhealthy: List[UnhealthyReason] = [] + self._suggestions: list[Suggestion] = [] + self._issues: list[Issue] = [] + self._unsupported: list[UnsupportedReason] = [] + self._unhealthy: list[UnhealthyReason] = [] @property - def data(self) -> Dict[str, Any]: + def data(self) -> dict[str, Any]: """Return data.""" return self._data @@ -68,7 +68,7 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes): return self._notify @property - def issues(self) -> List[Issue]: + def issues(self) -> list[Issue]: """Return a list of issues.""" return self._issues @@ -83,7 +83,7 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes): self._issues.append(issue) @property - def suggestions(self) -> List[Suggestion]: + def suggestions(self) -> list[Suggestion]: """Return a list of suggestions that can handled.""" return self._suggestions @@ -101,7 +101,7 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes): self._suggestions.append(suggestion) @property - def unsupported(self) -> List[UnsupportedReason]: + def unsupported(self) -> list[UnsupportedReason]: """Return a list of unsupported reasons.""" return self._unsupported @@ -112,7 +112,7 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes): self._unsupported.append(reason) @property - def unhealthy(self) -> List[UnhealthyReason]: + def unhealthy(self) -> list[UnhealthyReason]: """Return a list of unsupported reasons.""" return self._unhealthy @@ -143,7 +143,7 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes): issue: IssueType, context: ContextType, reference: Optional[str] = None, - suggestions: Optional[List[SuggestionType]] = None, + suggestions: Optional[list[SuggestionType]] = None, ) -> None: """Create issues and suggestion.""" self.issues = Issue(issue, context, reference) diff --git a/supervisor/resolution/validate.py b/supervisor/resolution/validate.py index 793f65a7d..5c0de5a33 100644 --- a/supervisor/resolution/validate.py +++ b/supervisor/resolution/validate.py @@ -1,13 +1,12 @@ """Validate resolution configuration schema.""" from pathlib import Path -from typing import List import voluptuous as vol from ..const import ATTR_CHECKS, ATTR_ENABLED -def get_valid_modules(folder) -> List[str]: +def get_valid_modules(folder) -> list[str]: """Validate check name.""" module_files = Path(__file__).parent.joinpath(folder) if not module_files.exists(): diff --git a/supervisor/services/__init__.py b/supervisor/services/__init__.py index f85fdef14..7aa29153b 100644 --- a/supervisor/services/__init__.py +++ b/supervisor/services/__init__.py @@ -1,5 +1,5 @@ """Handle internal services discovery.""" -from typing import Dict, List, Optional +from typing import Optional from ..coresys import CoreSys, CoreSysAttributes from .const import SERVICE_MQTT, SERVICE_MYSQL @@ -18,10 +18,10 @@ class ServiceManager(CoreSysAttributes): """Initialize Services handler.""" self.coresys: CoreSys = coresys self.data: ServicesData = ServicesData() - self.services_obj: Dict[str, ServiceInterface] = {} + self.services_obj: dict[str, ServiceInterface] = {} @property - def list_services(self) -> List[ServiceInterface]: + def list_services(self) -> list[ServiceInterface]: """Return a list of services.""" return list(self.services_obj.values()) diff --git a/supervisor/services/data.py b/supervisor/services/data.py index 83e8ec755..a16fc9816 100644 --- a/supervisor/services/data.py +++ b/supervisor/services/data.py @@ -1,5 +1,5 @@ """Handle service data for persistent supervisor reboot.""" -from typing import Any, Dict +from typing import Any from ..const import FILE_HASSIO_SERVICES from ..utils.common import FileConfiguration @@ -15,11 +15,11 @@ class ServicesData(FileConfiguration): super().__init__(FILE_HASSIO_SERVICES, SCHEMA_SERVICES_CONFIG) @property - def mqtt(self) -> Dict[str, Any]: + def mqtt(self) -> dict[str, Any]: """Return settings for MQTT service.""" return self._data[SERVICE_MQTT] @property - def mysql(self) -> Dict[str, Any]: + def mysql(self) -> dict[str, Any]: """Return settings for MySQL service.""" return self._data[SERVICE_MYSQL] diff --git a/supervisor/services/interface.py b/supervisor/services/interface.py index 768078a72..4bfdd2598 100644 --- a/supervisor/services/interface.py +++ b/supervisor/services/interface.py @@ -1,6 +1,6 @@ """Interface for single service.""" from abc import ABC, abstractmethod -from typing import Any, Dict, List, Optional +from typing import Any, Optional import voluptuous as vol @@ -23,7 +23,7 @@ class ServiceInterface(CoreSysAttributes, ABC): @property @abstractmethod - def _data(self) -> Dict[str, Any]: + def _data(self) -> dict[str, Any]: """Return data of this service.""" @property @@ -32,7 +32,7 @@ class ServiceInterface(CoreSysAttributes, ABC): """Return data schema of this service.""" @property - def providers(self) -> List[str]: + def providers(self) -> list[str]: """Return name of service providers addon.""" addons = [] for addon in self.sys_addons.installed: @@ -42,7 +42,7 @@ class ServiceInterface(CoreSysAttributes, ABC): @property @abstractmethod - def active(self) -> List[str]: + def active(self) -> list[str]: """Return list of addon slug they have enable that.""" @property @@ -54,14 +54,14 @@ class ServiceInterface(CoreSysAttributes, ABC): """Save changes.""" self.sys_services.data.save_data() - def get_service_data(self) -> Optional[Dict[str, Any]]: + def get_service_data(self) -> Optional[dict[str, Any]]: """Return the requested service data.""" if self.enabled: return self._data return None @abstractmethod - def set_service_data(self, addon: Addon, data: Dict[str, Any]) -> None: + def set_service_data(self, addon: Addon, data: dict[str, Any]) -> None: """Write the data into service object.""" @abstractmethod diff --git a/supervisor/services/modules/mqtt.py b/supervisor/services/modules/mqtt.py index c1af59b08..22a1b4dd5 100644 --- a/supervisor/services/modules/mqtt.py +++ b/supervisor/services/modules/mqtt.py @@ -1,6 +1,6 @@ """Provide the MQTT Service.""" import logging -from typing import Any, Dict, List +from typing import Any import voluptuous as vol @@ -51,7 +51,7 @@ class MQTTService(ServiceInterface): return SERVICE_MQTT @property - def _data(self) -> Dict[str, Any]: + def _data(self) -> dict[str, Any]: """Return data of this service.""" return self.sys_services.data.mqtt @@ -61,13 +61,13 @@ class MQTTService(ServiceInterface): return SCHEMA_SERVICE_MQTT @property - def active(self) -> List[str]: + def active(self) -> list[str]: """Return list of addon slug they have enable that.""" if not self.enabled: return [] return [self._data[ATTR_ADDON]] - def set_service_data(self, addon: Addon, data: Dict[str, Any]) -> None: + def set_service_data(self, addon: Addon, data: dict[str, Any]) -> None: """Write the data into service object.""" if self.enabled: _LOGGER.error( diff --git a/supervisor/services/modules/mysql.py b/supervisor/services/modules/mysql.py index fb9252902..cef530e44 100644 --- a/supervisor/services/modules/mysql.py +++ b/supervisor/services/modules/mysql.py @@ -1,6 +1,6 @@ """Provide the MySQL Service.""" import logging -from typing import Any, Dict, List +from typing import Any import voluptuous as vol @@ -45,7 +45,7 @@ class MySQLService(ServiceInterface): return SERVICE_MYSQL @property - def _data(self) -> Dict[str, Any]: + def _data(self) -> dict[str, Any]: """Return data of this service.""" return self.sys_services.data.mysql @@ -55,13 +55,13 @@ class MySQLService(ServiceInterface): return SCHEMA_SERVICE_MYSQL @property - def active(self) -> List[str]: + def active(self) -> list[str]: """Return list of addon slug they have enable that.""" if not self.enabled: return [] return [self._data[ATTR_ADDON]] - def set_service_data(self, addon: Addon, data: Dict[str, Any]) -> None: + def set_service_data(self, addon: Addon, data: dict[str, Any]) -> None: """Write the data into service object.""" if self.enabled: _LOGGER.error( diff --git a/supervisor/store/__init__.py b/supervisor/store/__init__.py index 4526d0575..9a1eaa0f0 100644 --- a/supervisor/store/__init__.py +++ b/supervisor/store/__init__.py @@ -1,7 +1,6 @@ """Add-on Store handler.""" import asyncio import logging -from typing import Dict, List from ..const import URL_HASSIO_ADDONS from ..coresys import CoreSys, CoreSysAttributes @@ -25,10 +24,10 @@ class StoreManager(CoreSysAttributes): """Initialize Docker base wrapper.""" self.coresys: CoreSys = coresys self.data = StoreData(coresys) - self.repositories: Dict[str, Repository] = {} + self.repositories: dict[str, Repository] = {} @property - def all(self) -> List[Repository]: + def all(self) -> list[Repository]: """Return list of add-on repositories.""" return list(self.repositories.values()) diff --git a/supervisor/store/data.py b/supervisor/store/data.py index 8528eb9a6..aad67b9a2 100644 --- a/supervisor/store/data.py +++ b/supervisor/store/data.py @@ -1,7 +1,7 @@ """Init file for Supervisor add-on data.""" import logging from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Optional import voluptuous as vol from voluptuous.humanize import humanize_error @@ -34,8 +34,8 @@ class StoreData(CoreSysAttributes): def __init__(self, coresys: CoreSys): """Initialize data holder.""" self.coresys: CoreSys = coresys - self.repositories: Dict[str, Any] = {} - self.addons: Dict[str, Any] = {} + self.repositories: dict[str, Any] = {} + self.addons: dict[str, Any] = {} def update(self) -> None: """Read data from add-on repository.""" @@ -86,7 +86,7 @@ class StoreData(CoreSysAttributes): self.repositories[slug] = repository_info self._read_addons_folder(path, slug) - def _find_addons(self, path: Path, repository: Dict) -> Optional[list[Path]]: + def _find_addons(self, path: Path, repository: dict) -> Optional[list[Path]]: """Find add-ons in the path.""" try: # Generate a list without artefact, safe for corruptions @@ -116,7 +116,7 @@ class StoreData(CoreSysAttributes): return None return addon_list - def _read_addons_folder(self, path: Path, repository: Dict) -> None: + def _read_addons_folder(self, path: Path, repository: dict) -> None: """Read data from add-ons folder.""" if not (addon_list := self._find_addons(path, repository)): return diff --git a/supervisor/store/git.py b/supervisor/store/git.py index ab67a329d..c5149d52c 100644 --- a/supervisor/store/git.py +++ b/supervisor/store/git.py @@ -3,7 +3,7 @@ import asyncio import functools as ft import logging from pathlib import Path -from typing import Dict, Optional +from typing import Optional import git @@ -31,7 +31,7 @@ class GitRepo(CoreSysAttributes): self.path: Path = path self.lock: asyncio.Lock = asyncio.Lock() - self.data: Dict[str, str] = RE_REPOSITORY.match(url).groupdict() + self.data: dict[str, str] = RE_REPOSITORY.match(url).groupdict() def __repr__(self) -> str: """Return internal representation.""" diff --git a/supervisor/store/repository.py b/supervisor/store/repository.py index c5e474baf..c693a0769 100644 --- a/supervisor/store/repository.py +++ b/supervisor/store/repository.py @@ -1,7 +1,7 @@ """Represent a Supervisor repository.""" import logging from pathlib import Path -from typing import Dict, Optional +from typing import Optional import voluptuous as vol @@ -55,7 +55,7 @@ class Repository(CoreSysAttributes): return self._type @property - def data(self) -> Dict: + def data(self) -> dict: """Return data struct repository.""" return self.sys_store.data.repositories.get(self.slug, {}) diff --git a/supervisor/utils/__init__.py b/supervisor/utils/__init__.py index 70fd1277d..e351c9a18 100644 --- a/supervisor/utils/__init__.py +++ b/supervisor/utils/__init__.py @@ -6,7 +6,7 @@ import os from pathlib import Path import re import socket -from typing import Any, Dict +from typing import Any _LOGGER: logging.Logger = logging.getLogger(__name__) @@ -103,7 +103,7 @@ async def remove_folder(folder: Path, content_only: bool = False) -> None: _LOGGER.error("Can't remove folder %s: %s", folder, error_msg) -def clean_env() -> Dict[str, str]: +def clean_env() -> dict[str, str]: """Return a clean env from system.""" new_env = {} for key in ("HOME", "PATH", "PWD", "CWD", "SHLVL"): diff --git a/supervisor/utils/codenotary.py b/supervisor/utils/codenotary.py index 652a24f72..deeb2c484 100644 --- a/supervisor/utils/codenotary.py +++ b/supervisor/utils/codenotary.py @@ -5,7 +5,7 @@ import json import logging from pathlib import Path import shlex -from typing import Optional, Set, Tuple, Union +from typing import Optional, Union import async_timeout @@ -15,7 +15,7 @@ from ..exceptions import CodeNotaryBackendError, CodeNotaryError, CodeNotaryUntr _LOGGER: logging.Logger = logging.getLogger(__name__) _VCN_CMD: str = "vcn authenticate --silent --output json" -_CACHE: Set[Tuple[str, Path, str, str]] = set() +_CACHE: set[tuple[str, Path, str, str]] = set() _ATTR_ERROR = "error" diff --git a/supervisor/utils/common.py b/supervisor/utils/common.py index 495abe64b..595a0ee3c 100644 --- a/supervisor/utils/common.py +++ b/supervisor/utils/common.py @@ -1,7 +1,7 @@ """Common utils.""" import logging from pathlib import Path -from typing import Any, Dict, List +from typing import Any import voluptuous as vol from voluptuous.humanize import humanize_error @@ -12,10 +12,10 @@ from .yaml import read_yaml_file, write_yaml_file _LOGGER: logging.Logger = logging.getLogger(__name__) -_DEFAULT: Dict[str, Any] = {} +_DEFAULT: dict[str, Any] = {} -def find_one_filetype(path: Path, filename: str, filetypes: List[str]) -> Path: +def find_one_filetype(path: Path, filename: str, filetypes: list[str]) -> Path: """Find first file matching filetypes.""" for file in path.glob(f"**/{filename}.*"): if file.suffix in filetypes: @@ -52,7 +52,7 @@ class FileConfiguration: """Initialize hass object.""" self._file: Path = file_path self._schema: vol.Schema = schema - self._data: Dict[str, Any] = _DEFAULT + self._data: dict[str, Any] = _DEFAULT self.read_data() diff --git a/supervisor/utils/dt.py b/supervisor/utils/dt.py index cd37a56ab..27cffe189 100644 --- a/supervisor/utils/dt.py +++ b/supervisor/utils/dt.py @@ -2,7 +2,7 @@ from contextlib import suppress from datetime import datetime, timedelta, timezone, tzinfo import re -from typing import Any, Dict, Optional +from typing import Any, Optional import zoneinfo import ciso8601 @@ -38,7 +38,7 @@ def parse_datetime(dt_str): match = DATETIME_RE.match(dt_str) if not match: return None - kws: Dict[str, Any] = match.groupdict() + kws: dict[str, Any] = match.groupdict() if kws["microsecond"]: kws["microsecond"] = kws["microsecond"].ljust(6, "0") tzinfo_str = kws.pop("tzinfo") diff --git a/supervisor/utils/gdbus.py b/supervisor/utils/gdbus.py index c054cab67..1c07e93ea 100644 --- a/supervisor/utils/gdbus.py +++ b/supervisor/utils/gdbus.py @@ -7,7 +7,7 @@ import logging import re import shlex from signal import SIGINT -from typing import Any, Dict, List, Optional, Set +from typing import Any import xml.etree.ElementTree as ET import sentry_sdk @@ -55,7 +55,7 @@ RE_MONITOR_OUTPUT: re.Pattern[Any] = re.compile( ) # Map GDBus to errors -MAP_GDBUS_ERROR: Dict[str, Any] = { +MAP_GDBUS_ERROR: dict[str, Any] = { "GDBus.Error:org.freedesktop.DBus.Error.ServiceUnknown": DBusInterfaceError, "GDBus.Error:org.freedesktop.DBus.Error.Spawn.ChildExited": DBusFatalError, "No such file or directory": DBusNotConnectedError, @@ -91,8 +91,8 @@ class DBus: """Initialize dbus object.""" self.bus_name: str = bus_name self.object_path: str = object_path - self.methods: Set[str] = set() - self.signals: Set[str] = set() + self.methods: set[str] = set() + self.signals: set[str] = set() @staticmethod async def connect(bus_name: str, object_path: str) -> DBus: @@ -186,12 +186,12 @@ class DBus: raise DBusParseError() from err @staticmethod - def gvariant_args(args: List[Any]) -> str: + def gvariant_args(args: list[Any]) -> str: """Convert args into gvariant.""" gvariant = "" for arg in args: if isinstance(arg, bool): - gvariant += " {}".format(str(arg).lower()) + gvariant += f" {str(arg).lower()}" elif isinstance(arg, (int, float)): gvariant += f" {arg}" elif isinstance(arg, str): @@ -201,7 +201,7 @@ class DBus: return gvariant.lstrip() - async def call_dbus(self, method: str, *args: List[Any]) -> str: + async def call_dbus(self, method: str, *args: list[Any]) -> str: """Call a dbus method.""" command = shlex.split( CALL.format( @@ -219,7 +219,7 @@ class DBus: # Parse and return data return self.parse_gvariant(data) - async def get_properties(self, interface: str) -> Dict[str, Any]: + async def get_properties(self, interface: str) -> dict[str, Any]: """Read all properties from interface.""" try: return (await self.call_dbus(DBUS_METHOD_GETALL, interface))[0] @@ -229,7 +229,7 @@ class DBus: async def set_property( self, interface: str, name: str, value: Any - ) -> Dict[str, Any]: + ) -> dict[str, Any]: """Set a property from interface.""" try: return (await self.call_dbus(DBUS_METHOD_SET, interface, name, value))[0] @@ -237,7 +237,7 @@ class DBus: _LOGGER.error("No Set attribute %s for %s", name, interface) raise DBusFatalError() from err - async def _send(self, command: List[str], silent=False) -> str: + async def _send(self, command: list[str], silent=False) -> str: """Send command over dbus.""" # Run command _LOGGER.debug("Send D-Bus command: %s", command) @@ -319,11 +319,11 @@ class DBusCallWrapper: class DBusSignalWrapper: """Process Signals.""" - def __init__(self, dbus: DBus, signals: Optional[str] = None): + def __init__(self, dbus: DBus, signals: str | None = None): """Initialize dbus signal wrapper.""" self.dbus: DBus = dbus - self._signals: Optional[str] = signals - self._proc: Optional[asyncio.Process] = None + self._signals: str | None = signals + self._proc: asyncio.Process | None = None async def __aenter__(self): """Start monitor events.""" diff --git a/supervisor/utils/pwned.py b/supervisor/utils/pwned.py index 912bc0f0a..ea8663a8a 100644 --- a/supervisor/utils/pwned.py +++ b/supervisor/utils/pwned.py @@ -2,7 +2,6 @@ import asyncio import io import logging -from typing import Set import aiohttp @@ -11,7 +10,7 @@ from ..exceptions import PwnedConnectivityError, PwnedError, PwnedSecret _LOGGER: logging.Logger = logging.getLogger(__name__) _API_CALL: str = "https://api.pwnedpasswords.com/range/{hash}" -_CACHE: Set[str] = set() +_CACHE: set[str] = set() async def check_pwned_password(websession: aiohttp.ClientSession, sha1_pw: str) -> None: diff --git a/supervisor/utils/tar.py b/supervisor/utils/tar.py index a371ec04f..ecde2e569 100644 --- a/supervisor/utils/tar.py +++ b/supervisor/utils/tar.py @@ -4,7 +4,7 @@ import logging import os from pathlib import Path, PurePath import tarfile -from typing import IO, Generator, List, Optional +from typing import IO, Generator, Optional from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import padding @@ -142,7 +142,7 @@ def secure_path(tar: tarfile.TarFile) -> Generator[tarfile.TarInfo, None, None]: yield member -def _is_excluded_by_filter(path: PurePath, exclude_list: List[str]) -> bool: +def _is_excluded_by_filter(path: PurePath, exclude_list: list[str]) -> bool: """Filter to filter excludes.""" for exclude in exclude_list: @@ -157,7 +157,7 @@ def _is_excluded_by_filter(path: PurePath, exclude_list: List[str]) -> bool: def atomic_contents_add( tar_file: tarfile.TarFile, origin_path: Path, - excludes: List[str], + excludes: list[str], arcname: str = ".", ) -> None: """Append directories and/or files to the TarFile if excludes wont filter.""" diff --git a/tests/docker/test_addon.py b/tests/docker/test_addon.py index 9d440a2c6..d789464fb 100644 --- a/tests/docker/test_addon.py +++ b/tests/docker/test_addon.py @@ -1,5 +1,4 @@ """Test docker addon setup.""" -from typing import Dict from unittest.mock import MagicMock, PropertyMock, patch import pytest @@ -15,7 +14,7 @@ from ..common import load_json_fixture @pytest.fixture(name="addonsdata_system") -def fixture_addonsdata_system() -> Dict[str, Data]: +def fixture_addonsdata_system() -> dict[str, Data]: """Mock AddonsData.system.""" with patch( "supervisor.addons.data.AddonsData.system", new_callable=PropertyMock @@ -24,7 +23,7 @@ def fixture_addonsdata_system() -> Dict[str, Data]: @pytest.fixture(name="addonsdata_user", autouse=True) -def fixture_addonsdata_user() -> Dict[str, Data]: +def fixture_addonsdata_user() -> dict[str, Data]: """Mock AddonsData.user.""" with patch( "supervisor.addons.data.AddonsData.user", new_callable=PropertyMock @@ -41,7 +40,7 @@ def fixture_os_environ(): def get_docker_addon( - coresys: CoreSys, addonsdata_system: Dict[str, Data], config_file: str + coresys: CoreSys, addonsdata_system: dict[str, Data], config_file: str ): """Make and return docker addon object.""" config = vd.SCHEMA_ADDON_CONFIG(load_json_fixture(config_file)) @@ -53,7 +52,7 @@ def get_docker_addon( return docker_addon -def test_base_volumes_included(coresys: CoreSys, addonsdata_system: Dict[str, Data]): +def test_base_volumes_included(coresys: CoreSys, addonsdata_system: dict[str, Data]): """Dev and data volumes always included.""" docker_addon = get_docker_addon( coresys, addonsdata_system, "basic-addon-config.json" @@ -73,7 +72,7 @@ def test_base_volumes_included(coresys: CoreSys, addonsdata_system: Dict[str, Da def test_addon_map_folder_defaults( - coresys: CoreSys, addonsdata_system: Dict[str, Data] + coresys: CoreSys, addonsdata_system: dict[str, Data] ): """Validate defaults for mapped folders in addons.""" docker_addon = get_docker_addon( @@ -97,7 +96,7 @@ def test_addon_map_folder_defaults( assert str(docker_addon.sys_config.path_extern_share) not in volumes -def test_journald_addon(coresys: CoreSys, addonsdata_system: Dict[str, Data]): +def test_journald_addon(coresys: CoreSys, addonsdata_system: dict[str, Data]): """Validate volume for journald option.""" docker_addon = get_docker_addon( coresys, addonsdata_system, "journald-addon-config.json" @@ -116,7 +115,7 @@ def test_journald_addon(coresys: CoreSys, addonsdata_system: Dict[str, Data]): assert volumes.get(str(SYSTEMD_JOURNAL_VOLATILE)).get("mode") == "ro" -def test_not_journald_addon(coresys: CoreSys, addonsdata_system: Dict[str, Data]): +def test_not_journald_addon(coresys: CoreSys, addonsdata_system: dict[str, Data]): """Validate journald option defaults off.""" docker_addon = get_docker_addon( coresys, addonsdata_system, "basic-addon-config.json" diff --git a/tests/hardware/test_module.py b/tests/hardware/test_module.py index 92f7ad0e7..e68a160fc 100644 --- a/tests/hardware/test_module.py +++ b/tests/hardware/test_module.py @@ -120,19 +120,13 @@ def test_device_filter(coresys): coresys.hardware.update_device(device) assert sorted( - [device.path for device in coresys.hardware.filter_devices()] - ) == sorted([device.path for device in coresys.hardware.devices]) + device.path for device in coresys.hardware.filter_devices() + ) == sorted(device.path for device in coresys.hardware.devices) assert sorted( - [ - device.path - for device in coresys.hardware.filter_devices( - subsystem=UdevSubsystem.SERIAL - ) - ] + device.path + for device in coresys.hardware.filter_devices(subsystem=UdevSubsystem.SERIAL) ) == sorted( - [ - device.path - for device in coresys.hardware.devices - if device.subsystem == UdevSubsystem.SERIAL - ] + device.path + for device in coresys.hardware.devices + if device.subsystem == UdevSubsystem.SERIAL )