diff --git a/.devcontainer/devcontainer.json b/.devcontainer/devcontainer.json index 86ca2dab1..77618694e 100644 --- a/.devcontainer/devcontainer.json +++ b/.devcontainer/devcontainer.json @@ -10,7 +10,7 @@ "visualstudioexptteam.vscodeintellicode", "esbenp.prettier-vscode" ], - "mounts": [ "type=volume,target=/var/lib/docker" ], + "mounts": ["type=volume,target=/var/lib/docker"], "settings": { "terminal.integrated.profiles.linux": { "zsh": { @@ -26,7 +26,7 @@ "python.linting.pylintEnabled": true, "python.linting.enabled": true, "python.formatting.provider": "black", - "python.formatting.blackArgs": ["--target-version", "py39"], + "python.formatting.blackArgs": ["--target-version", "py310"], "python.formatting.blackPath": "/usr/local/bin/black", "python.linting.banditPath": "/usr/local/bin/bandit", "python.linting.flake8Path": "/usr/local/bin/flake8", diff --git a/.github/workflows/builder.yml b/.github/workflows/builder.yml index f11701724..eb78563a0 100644 --- a/.github/workflows/builder.yml +++ b/.github/workflows/builder.yml @@ -33,10 +33,9 @@ on: - setup.py env: - DEFAULT_PYTHON: 3.9 + DEFAULT_PYTHON: "3.10" BUILD_NAME: supervisor BUILD_TYPE: supervisor - WHEELS_TAG: 3.9-alpine3.14 jobs: init: @@ -88,18 +87,26 @@ jobs: uses: actions/checkout@v3.0.2 with: fetch-depth: 0 + + - name: Write env-file + if: needs.init.outputs.requirements == 'true' + run: | + ( + # Fix out of memory issues with rust + echo "CARGO_NET_GIT_FETCH_WITH_CLI=true" + ) > .env_file - name: Build wheels if: needs.init.outputs.requirements == 'true' - uses: home-assistant/wheels@2022.01.2 + uses: home-assistant/wheels@2022.06.7 with: - tag: ${{ env.WHEELS_TAG }} + abi: cp310 + tag: musllinux_1_2 arch: ${{ matrix.arch }} - wheels-host: wheels.hass.io wheels-key: ${{ secrets.WHEELS_KEY }} - wheels-user: wheels - apk: "build-base;libffi-dev;openssl-dev;cargo" + apk: "libffi-dev;openssl-dev" skip-binary: aiohttp + env-file: true requirements: "requirements.txt" - name: Set version diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index ebc008d5f..6e010b184 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -8,7 +8,7 @@ on: pull_request: ~ env: - DEFAULT_PYTHON: 3.9 + DEFAULT_PYTHON: "3.10" PRE_COMMIT_HOME: ~/.cache/pre-commit DEFAULT_CAS: v1.0.2 @@ -19,7 +19,7 @@ jobs: runs-on: ubuntu-latest strategy: matrix: - python-version: [3.9] + python-version: ["3.10"] name: Prepare Python ${{ matrix.python-version }} dependencies steps: - name: Check out code from GitHub @@ -341,7 +341,7 @@ jobs: needs: prepare strategy: matrix: - python-version: [3.9] + python-version: ["3.10"] name: Run tests Python ${{ matrix.python-version }} steps: - name: Check out code from GitHub diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index ead296ac2..bce42c248 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -7,7 +7,7 @@ repos: - --safe - --quiet - --target-version - - py39 + - py310 files: ^((supervisor|tests)/.+)?[^/]+\.py$ - repo: https://gitlab.com/pycqa/flake8 rev: 3.8.3 @@ -31,4 +31,4 @@ repos: rev: v2.32.1 hooks: - id: pyupgrade - args: [--py39-plus] + args: [--py310-plus] diff --git a/Dockerfile b/Dockerfile index c88893610..0d8f56297 100644 --- a/Dockerfile +++ b/Dockerfile @@ -6,7 +6,6 @@ ENV \ SUPERVISOR_API=http://localhost ARG \ - BUILD_ARCH \ CAS_VERSION # Install base @@ -40,7 +39,7 @@ COPY requirements.txt . RUN \ export MAKEFLAGS="-j$(nproc)" \ && pip3 install --no-cache-dir --no-index --only-binary=:all: --find-links \ - "https://wheels.home-assistant.io/alpine-$(cut -d '.' -f 1-2 < /etc/alpine-release)/${BUILD_ARCH}/" \ + "https://wheels.home-assistant.io/musllinux/" \ -r ./requirements.txt \ && rm -f requirements.txt diff --git a/build.yaml b/build.yaml index 58750a30e..0c9355a89 100644 --- a/build.yaml +++ b/build.yaml @@ -1,11 +1,11 @@ image: homeassistant/{arch}-hassio-supervisor shadow_repository: ghcr.io/home-assistant build_from: - aarch64: ghcr.io/home-assistant/aarch64-base-python:3.9-alpine3.14 - armhf: ghcr.io/home-assistant/armhf-base-python:3.9-alpine3.14 - armv7: ghcr.io/home-assistant/armv7-base-python:3.9-alpine3.14 - amd64: ghcr.io/home-assistant/amd64-base-python:3.9-alpine3.14 - i386: ghcr.io/home-assistant/i386-base-python:3.9-alpine3.14 + aarch64: ghcr.io/home-assistant/aarch64-base-python:3.10-alpine3.16 + armhf: ghcr.io/home-assistant/armhf-base-python:3.10-alpine3.16 + armv7: ghcr.io/home-assistant/armv7-base-python:3.10-alpine3.16 + amd64: ghcr.io/home-assistant/amd64-base-python:3.10-alpine3.16 + i386: ghcr.io/home-assistant/i386-base-python:3.10-alpine3.16 codenotary: signer: notary@home-assistant.io base_image: notary@home-assistant.io diff --git a/supervisor/addons/__init__.py b/supervisor/addons/__init__.py index ee3a8e7ac..c7c7463cd 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 Optional, Union +from typing import Union from ..const import AddonBoot, AddonStartup, AddonState from ..coresys import CoreSys, CoreSysAttributes @@ -53,7 +53,7 @@ class AddonManager(CoreSysAttributes): """Return a list of all installed add-ons.""" return list(self.local.values()) - def get(self, addon_slug: str, local_only: bool = False) -> Optional[AnyAddon]: + def get(self, addon_slug: str, local_only: bool = False) -> AnyAddon | None: """Return an add-on from slug. Prio: @@ -66,7 +66,7 @@ class AddonManager(CoreSysAttributes): return self.store.get(addon_slug) return None - def from_token(self, token: str) -> Optional[Addon]: + def from_token(self, token: str) -> Addon | None: """Return an add-on from Supervisor token.""" for addon in self.installed: if token == addon.supervisor_token: @@ -246,7 +246,7 @@ class AddonManager(CoreSysAttributes): conditions=ADDON_UPDATE_CONDITIONS, on_condition=AddonsJobError, ) - async def update(self, slug: str, backup: Optional[bool] = False) -> None: + async def update(self, slug: str, backup: bool | None = False) -> None: """Update add-on.""" if slug not in self.local: raise AddonsError(f"Add-on {slug} is not installed", _LOGGER.error) diff --git a/supervisor/addons/addon.py b/supervisor/addons/addon.py index 53259924a..523771752 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, Final, Optional +from typing import Any, Awaitable, Final import aiohttp from deepmerge import Merger @@ -240,7 +240,7 @@ class Addon(AddonModel): return self._available(self.data_store) @property - def version(self) -> Optional[str]: + def version(self) -> str | None: """Return installed version.""" return self.persist[ATTR_VERSION] @@ -264,7 +264,7 @@ class Addon(AddonModel): ) @options.setter - def options(self, value: Optional[dict[str, Any]]) -> None: + def options(self, value: dict[str, Any] | None) -> None: """Store user add-on options.""" self.persist[ATTR_OPTIONS] = {} if value is None else deepcopy(value) @@ -309,17 +309,17 @@ class Addon(AddonModel): return self.persist[ATTR_UUID] @property - def supervisor_token(self) -> Optional[str]: + def supervisor_token(self) -> str | None: """Return access token for Supervisor API.""" return self.persist.get(ATTR_ACCESS_TOKEN) @property - def ingress_token(self) -> Optional[str]: + def ingress_token(self) -> str | None: """Return access token for Supervisor API.""" return self.persist.get(ATTR_INGRESS_TOKEN) @property - def ingress_entry(self) -> Optional[str]: + def ingress_entry(self) -> str | None: """Return ingress external URL.""" if self.with_ingress: return f"/api/hassio_ingress/{self.ingress_token}" @@ -341,12 +341,12 @@ class Addon(AddonModel): self.persist[ATTR_PROTECTED] = value @property - def ports(self) -> Optional[dict[str, Optional[int]]]: + def ports(self) -> dict[str, int | None] | None: """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: dict[str, int | None] | None) -> None: """Set custom ports of add-on.""" if value is None: self.persist.pop(ATTR_NETWORK, None) @@ -361,7 +361,7 @@ class Addon(AddonModel): self.persist[ATTR_NETWORK] = new_ports @property - def ingress_url(self) -> Optional[str]: + def ingress_url(self) -> str | None: """Return URL to ingress url.""" if not self.with_ingress: return None @@ -372,7 +372,7 @@ class Addon(AddonModel): return url @property - def webui(self) -> Optional[str]: + def webui(self) -> str | None: """Return URL to webui or None.""" url = super().webui if not url: @@ -400,7 +400,7 @@ class Addon(AddonModel): return f"{proto}://[HOST]:{port}{s_suffix}" @property - def ingress_port(self) -> Optional[int]: + def ingress_port(self) -> int | None: """Return Ingress port.""" if not self.with_ingress: return None @@ -411,7 +411,7 @@ class Addon(AddonModel): return port @property - def ingress_panel(self) -> Optional[bool]: + def ingress_panel(self) -> bool | None: """Return True if the add-on access support ingress.""" if not self.with_ingress: return None @@ -424,19 +424,19 @@ class Addon(AddonModel): self.persist[ATTR_INGRESS_PANEL] = value @property - def audio_output(self) -> Optional[str]: + def audio_output(self) -> str | None: """Return a pulse profile for output or None.""" if not self.with_audio: return None return self.persist.get(ATTR_AUDIO_OUTPUT) @audio_output.setter - def audio_output(self, value: Optional[str]): + def audio_output(self, value: str | None): """Set audio output profile settings.""" self.persist[ATTR_AUDIO_OUTPUT] = value @property - def audio_input(self) -> Optional[str]: + def audio_input(self) -> str | None: """Return pulse profile for input or None.""" if not self.with_audio: return None @@ -444,12 +444,12 @@ class Addon(AddonModel): return self.persist.get(ATTR_AUDIO_INPUT) @audio_input.setter - def audio_input(self, value: Optional[str]) -> None: + def audio_input(self, value: str | None) -> None: """Set audio input settings.""" self.persist[ATTR_AUDIO_INPUT] = value @property - def image(self) -> Optional[str]: + def image(self) -> str | None: """Return image name of add-on.""" return self.persist.get(ATTR_IMAGE) diff --git a/supervisor/addons/model.py b/supervisor/addons/model.py index 160ee87fe..117d52f02 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, Optional +from typing import Any, Awaitable from awesomeversion import AwesomeVersion, AwesomeVersionException @@ -123,7 +123,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_BOOT] @property - def auto_update(self) -> Optional[bool]: + def auto_update(self) -> bool | None: """Return if auto update is enable.""" return None @@ -148,22 +148,22 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_TIMEOUT] @property - def uuid(self) -> Optional[str]: + def uuid(self) -> str | None: """Return an API token for this add-on.""" return None @property - def supervisor_token(self) -> Optional[str]: + def supervisor_token(self) -> str | None: """Return access token for Supervisor API.""" return None @property - def ingress_token(self) -> Optional[str]: + def ingress_token(self) -> str | None: """Return access token for Supervisor API.""" return None @property - def ingress_entry(self) -> Optional[str]: + def ingress_entry(self) -> str | None: """Return ingress external URL.""" return None @@ -173,7 +173,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_DESCRIPTON] @property - def long_description(self) -> Optional[str]: + def long_description(self) -> str | None: """Return README.md as long_description.""" readme = Path(self.path_location, "README.md") @@ -243,32 +243,32 @@ class AddonModel(CoreSysAttributes, ABC): return self.data.get(ATTR_DISCOVERY, []) @property - def ports_description(self) -> Optional[dict[str, str]]: + def ports_description(self) -> dict[str, str] | None: """Return descriptions of ports.""" return self.data.get(ATTR_PORTS_DESCRIPTION) @property - def ports(self) -> Optional[dict[str, Optional[int]]]: + def ports(self) -> dict[str, int | None] | None: """Return ports of add-on.""" return self.data.get(ATTR_PORTS) @property - def ingress_url(self) -> Optional[str]: + def ingress_url(self) -> str | None: """Return URL to ingress url.""" return None @property - def webui(self) -> Optional[str]: + def webui(self) -> str | None: """Return URL to webui or None.""" return self.data.get(ATTR_WEBUI) @property - def watchdog(self) -> Optional[str]: + def watchdog(self) -> str | None: """Return URL to for watchdog or None.""" return self.data.get(ATTR_WATCHDOG) @property - def ingress_port(self) -> Optional[int]: + def ingress_port(self) -> int | None: """Return Ingress port.""" return None @@ -313,7 +313,7 @@ class AddonModel(CoreSysAttributes, ABC): return [Path(node) for node in self.data.get(ATTR_DEVICES, [])] @property - def environment(self) -> Optional[dict[str, str]]: + def environment(self) -> dict[str, str] | None: """Return environment of add-on.""" return self.data.get(ATTR_ENVIRONMENT) @@ -362,12 +362,12 @@ class AddonModel(CoreSysAttributes, ABC): return self.data.get(ATTR_BACKUP_EXCLUDE, []) @property - def backup_pre(self) -> Optional[str]: + def backup_pre(self) -> str | None: """Return pre-backup command.""" return self.data.get(ATTR_BACKUP_PRE) @property - def backup_post(self) -> Optional[str]: + def backup_post(self) -> str | None: """Return post-backup command.""" return self.data.get(ATTR_BACKUP_POST) @@ -392,7 +392,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_INGRESS] @property - def ingress_panel(self) -> Optional[bool]: + def ingress_panel(self) -> bool | None: """Return True if the add-on access support ingress.""" return None @@ -442,7 +442,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_DEVICETREE] @property - def with_tmpfs(self) -> Optional[str]: + def with_tmpfs(self) -> str | None: """Return if tmp is in memory of add-on.""" return self.data[ATTR_TMPFS] @@ -462,12 +462,12 @@ class AddonModel(CoreSysAttributes, ABC): return self.data[ATTR_VIDEO] @property - def homeassistant_version(self) -> Optional[str]: + def homeassistant_version(self) -> str | None: """Return min Home Assistant version they needed by Add-on.""" return self.data.get(ATTR_HOMEASSISTANT) @property - def url(self) -> Optional[str]: + def url(self) -> str | None: """Return URL of add-on.""" return self.data.get(ATTR_URL) @@ -510,7 +510,7 @@ class AddonModel(CoreSysAttributes, ABC): return self.sys_arch.default @property - def image(self) -> Optional[str]: + def image(self) -> str | None: """Generate image name from data.""" return self._image(self.data) @@ -571,7 +571,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) -> list[dict[any, any]] | None: """Create a UI schema for add-on options.""" raw_schema = self.data[ATTR_SCHEMA] @@ -590,7 +590,7 @@ class AddonModel(CoreSysAttributes, ABC): return ATTR_CODENOTARY in self.data @property - def codenotary(self) -> Optional[str]: + def codenotary(self) -> str | None: """Return Signer email address for CAS.""" return self.data.get(ATTR_CODENOTARY) @@ -614,7 +614,7 @@ class AddonModel(CoreSysAttributes, ABC): return False # Home Assistant - version: Optional[AwesomeVersion] = config.get(ATTR_HOMEASSISTANT) + version: AwesomeVersion | None = config.get(ATTR_HOMEASSISTANT) try: return self.sys_homeassistant.version >= version except (AwesomeVersionException, TypeError): @@ -638,7 +638,7 @@ class AddonModel(CoreSysAttributes, ABC): """Uninstall this add-on.""" return self.sys_addons.uninstall(self.slug) - def update(self, backup: Optional[bool] = False) -> Awaitable[None]: + def update(self, backup: bool | None = False) -> Awaitable[None]: """Update this add-on.""" return self.sys_addons.update(self.slug, backup=backup) diff --git a/supervisor/addons/options.py b/supervisor/addons/options.py index b9daded7e..5837cbb9b 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, Union +from typing import Any import voluptuous as vol @@ -293,7 +293,7 @@ class UiOptions(CoreSysAttributes): multiple: bool = False, ) -> None: """Validate a single element.""" - ui_node: dict[str, Union[str, bool, float, list[str]]] = {"name": key} + ui_node: dict[str, str | bool | float | list[str]] = {"name": key} # If multiple if multiple: diff --git a/supervisor/api/__init__.py b/supervisor/api/__init__.py index 73bc27f48..57df58769 100644 --- a/supervisor/api/__init__.py +++ b/supervisor/api/__init__.py @@ -1,7 +1,7 @@ """Init file for Supervisor RESTful API.""" import logging from pathlib import Path -from typing import Any, Optional +from typing import Any from aiohttp import web @@ -63,7 +63,7 @@ class RestAPI(CoreSysAttributes): # service stuff self._runner: web.AppRunner = web.AppRunner(self.webapp) - self._site: Optional[web.TCPSite] = None + self._site: web.TCPSite | None = None async def load(self) -> None: """Register REST API Calls.""" diff --git a/supervisor/api/ingress.py b/supervisor/api/ingress.py index 6461c20d8..74df1a8ac 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, Union +from typing import Any import aiohttp from aiohttp import ClientTimeout, hdrs, web @@ -86,7 +86,7 @@ class APIIngress(CoreSysAttributes): @require_home_assistant async def handler( self, request: web.Request - ) -> Union[web.Response, web.StreamResponse, web.WebSocketResponse]: + ) -> web.Response | web.StreamResponse | web.WebSocketResponse: """Route data to Supervisor ingress service.""" # Check Ingress Session @@ -157,7 +157,7 @@ class APIIngress(CoreSysAttributes): async def _handle_request( self, request: web.Request, addon: Addon, path: str - ) -> Union[web.Response, web.StreamResponse]: + ) -> web.Response | web.StreamResponse: """Ingress route for request.""" url = self._create_url(addon, path) source_header = _init_header(request, addon) @@ -216,9 +216,7 @@ class APIIngress(CoreSysAttributes): return response -def _init_header( - request: web.Request, addon: str -) -> Union[CIMultiDict, dict[str, str]]: +def _init_header(request: web.Request, addon: str) -> CIMultiDict | dict[str, str]: """Create initial header.""" headers = {} diff --git a/supervisor/api/utils.py b/supervisor/api/utils.py index 8fb1dfe3b..bc73c3d8c 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, Optional +from typing import Any from aiohttp import web from aiohttp.hdrs import AUTHORIZATION @@ -25,7 +25,7 @@ from ..utils.log_format import format_message from .const import CONTENT_TYPE_BINARY, HEADER_TOKEN, HEADER_TOKEN_OLD -def excract_supervisor_token(request: web.Request) -> Optional[str]: +def excract_supervisor_token(request: web.Request) -> str | None: """Extract Supervisor token from request.""" if supervisor_token := request.headers.get(HEADER_TOKEN): return supervisor_token @@ -112,7 +112,7 @@ def api_process_raw(content): def api_return_error( - error: Optional[Exception] = None, message: Optional[str] = None + error: Exception | None = None, message: str | None = None ) -> web.Response: """Return an API error message.""" if error and not message: @@ -130,7 +130,7 @@ def api_return_error( ) -def api_return_ok(data: Optional[dict[str, Any]] = None) -> web.Response: +def api_return_ok(data: dict[str, Any] | None = None) -> web.Response: """Return an API ok answer.""" return web.json_response( {JSON_RESULT: RESULT_OK, JSON_DATA: data or {}}, @@ -139,7 +139,7 @@ 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 + schema: vol.Schema, request: web.Request, origin: list[str] | None = None ) -> dict[str, Any]: """Validate request data with schema.""" data: dict[str, Any] = await request.json(loads=json_loads) diff --git a/supervisor/auth.py b/supervisor/auth.py index 0bfdeb618..f7b1952b9 100644 --- a/supervisor/auth.py +++ b/supervisor/auth.py @@ -2,7 +2,6 @@ import asyncio import hashlib import logging -from typing import Optional from .addons.addon import Addon from .const import ATTR_ADDON, ATTR_PASSWORD, ATTR_USERNAME, FILE_HASSIO_AUTH @@ -24,7 +23,7 @@ class Auth(FileConfiguration, CoreSysAttributes): self._running: dict[str, asyncio.Task] = {} - def _check_cache(self, username: str, password: str) -> Optional[bool]: + def _check_cache(self, username: str, password: str) -> bool | None: """Check password in cache.""" username_h = self._rehash(username) password_h = self._rehash(password, username) diff --git a/supervisor/backups/backup.py b/supervisor/backups/backup.py index c6ac3a736..219451438 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, Awaitable, Optional +from typing import Any, Awaitable from awesomeversion import AwesomeVersion, AwesomeVersionCompareException from cryptography.hazmat.backends import default_backend @@ -57,8 +57,8 @@ class Backup(CoreSysAttributes): self._tarfile: Path = tar_file self._data: dict[str, Any] = {} self._tmp = None - self._key: Optional[bytes] = None - self._aes: Optional[Cipher] = None + self._key: bytes | None = None + self._aes: Cipher | None = None @property def version(self) -> int: diff --git a/supervisor/config.py b/supervisor/config.py index ba693ec85..359ff2051 100644 --- a/supervisor/config.py +++ b/supervisor/config.py @@ -3,7 +3,6 @@ from datetime import datetime import logging import os from pathlib import Path, PurePath -from typing import Optional from awesomeversion import AwesomeVersion @@ -63,7 +62,7 @@ class CoreConfig(FileConfiguration): super().__init__(FILE_HASSIO_CONFIG, SCHEMA_SUPERVISOR_CONFIG) @property - def timezone(self) -> Optional[str]: + def timezone(self) -> str | None: """Return system timezone.""" timezone = self._data.get(ATTR_TIMEZONE) if timezone != _UTC: @@ -89,7 +88,7 @@ class CoreConfig(FileConfiguration): self._data[ATTR_VERSION] = value @property - def image(self) -> Optional[str]: + def image(self) -> str | None: """Return supervisor image.""" return self._data.get(ATTR_IMAGE) @@ -129,7 +128,7 @@ class CoreConfig(FileConfiguration): self._data[ATTR_DEBUG_BLOCK] = value @property - def diagnostics(self) -> Optional[bool]: + def diagnostics(self) -> bool | None: """Return bool if diagnostics is set otherwise None.""" return self._data[ATTR_DIAGNOSTICS] diff --git a/supervisor/core.py b/supervisor/core.py index 206f197b6..456434ff9 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, Optional +from typing import Awaitable import async_timeout @@ -31,7 +31,7 @@ class Core(CoreSysAttributes): def __init__(self, coresys: CoreSys): """Initialize Supervisor object.""" self.coresys: CoreSys = coresys - self._state: Optional[CoreState] = None + self._state: CoreState | None = None self.exit_code: int = 0 @property diff --git a/supervisor/dbus/hostname.py b/supervisor/dbus/hostname.py index c978ebba1..328a21b51 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, Optional +from typing import Any from ..exceptions import DBusError, DBusInterfaceError from ..utils.dbus import DBus @@ -43,37 +43,37 @@ class Hostname(DBusInterface): @property @dbus_property - def hostname(self) -> Optional[str]: + def hostname(self) -> str | None: """Return local hostname.""" return self.properties[DBUS_ATTR_STATIC_HOSTNAME] @property @dbus_property - def chassis(self) -> Optional[str]: + def chassis(self) -> str | None: """Return local chassis type.""" return self.properties[DBUS_ATTR_CHASSIS] @property @dbus_property - def deployment(self) -> Optional[str]: + def deployment(self) -> str | None: """Return local deployment type.""" return self.properties[DBUS_ATTR_DEPLOYMENT] @property @dbus_property - def kernel(self) -> Optional[str]: + def kernel(self) -> str | None: """Return local kernel version.""" return self.properties[DBUS_ATTR_KERNEL_RELEASE] @property @dbus_property - def operating_system(self) -> Optional[str]: + def operating_system(self) -> str | None: """Return local operating system.""" return self.properties[DBUS_ATTR_OPERATING_SYSTEM_PRETTY_NAME] @property @dbus_property - def cpe(self) -> Optional[str]: + def cpe(self) -> str | None: """Return local CPE.""" return self.properties[DBUS_ATTR_STATIC_OPERATING_SYSTEM_CPE_NAME] diff --git a/supervisor/dbus/interface.py b/supervisor/dbus/interface.py index e22075cbb..39e52c27b 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, Optional +from typing import Any from ..utils.dbus import DBus @@ -22,8 +22,8 @@ def dbus_property(func): class DBusInterface(ABC): """Handle D-Bus interface for hostname/system.""" - dbus: Optional[DBus] = None - name: Optional[str] = None + dbus: DBus | None = None + name: str | None = None @property def is_connected(self): @@ -42,9 +42,9 @@ class DBusInterface(ABC): class DBusInterfaceProxy(ABC): """Handle D-Bus interface proxy.""" - dbus: Optional[DBus] = None - object_path: Optional[str] = None - properties: Optional[dict[str, Any]] = None + dbus: DBus | None = None + object_path: str | None = None + properties: dict[str, Any] | None = None @abstractmethod async def connect(self): diff --git a/supervisor/dbus/network/configuration.py b/supervisor/dbus/network/configuration.py index fb679d1bc..704679400 100644 --- a/supervisor/dbus/network/configuration.py +++ b/supervisor/dbus/network/configuration.py @@ -1,6 +1,5 @@ """NetworkConnection object4s for Network Manager.""" from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface -from typing import Optional, Union import attr @@ -9,16 +8,16 @@ import attr 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() + gateway: IPv6Address | IPv6Address | None = attr.ib() + nameservers: list[IPv6Address | IPv6Address] = attr.ib() + address: list[IPv4Interface | IPv6Interface] = attr.ib() @attr.s(slots=True) class DNSConfiguration: """DNS configuration Object.""" - nameservers: list[Union[IPv4Address, IPv6Address]] = attr.ib() + nameservers: list[IPv4Address | IPv6Address] = attr.ib() domains: list[str] = attr.ib() interface: str = attr.ib() priority: int = attr.ib() @@ -29,48 +28,48 @@ class DNSConfiguration: class ConnectionProperties: """Connection Properties object for Network Manager.""" - id: Optional[str] = attr.ib() - uuid: Optional[str] = attr.ib() - type: Optional[str] = attr.ib() - interface_name: Optional[str] = attr.ib() + id: str | None = attr.ib() + uuid: str | None = attr.ib() + type: str | None = attr.ib() + interface_name: str | None = attr.ib() @attr.s(slots=True) class WirelessProperties: """Wireless Properties object for Network Manager.""" - ssid: Optional[str] = attr.ib() - assigned_mac: Optional[str] = attr.ib() - mode: Optional[str] = attr.ib() - powersave: Optional[int] = attr.ib() + ssid: str | None = attr.ib() + assigned_mac: str | None = attr.ib() + mode: str | None = attr.ib() + powersave: int | None = attr.ib() @attr.s(slots=True) class WirelessSecurityProperties: """Wireless Security Properties object for Network Manager.""" - auth_alg: Optional[str] = attr.ib() - key_mgmt: Optional[str] = attr.ib() - psk: Optional[str] = attr.ib() + auth_alg: str | None = attr.ib() + key_mgmt: str | None = attr.ib() + psk: str | None = attr.ib() @attr.s(slots=True) class EthernetProperties: """Ethernet properties object for Network Manager.""" - assigned_mac: Optional[str] = attr.ib() + assigned_mac: str | None = attr.ib() @attr.s(slots=True) class VlanProperties: """Ethernet properties object for Network Manager.""" - id: Optional[int] = attr.ib() - parent: Optional[str] = attr.ib() + id: int | None = attr.ib() + parent: str | None = attr.ib() @attr.s(slots=True) class IpProperties: """IP properties object for Network Manager.""" - method: Optional[str] = attr.ib() + method: str | None = attr.ib() diff --git a/supervisor/dbus/network/connection.py b/supervisor/dbus/network/connection.py index 46593b9d6..316b132d2 100644 --- a/supervisor/dbus/network/connection.py +++ b/supervisor/dbus/network/connection.py @@ -1,6 +1,5 @@ """Connection object for Network Manager.""" from ipaddress import ip_address, ip_interface -from typing import Optional from ...const import ATTR_ADDRESS, ATTR_PREFIX from ...utils.dbus import DBus @@ -39,8 +38,8 @@ class NetworkConnection(DBusInterfaceProxy): self.object_path = object_path self.properties = {} - self._ipv4: Optional[IpConfiguration] = None - self._ipv6: Optional[IpConfiguration] = None + self._ipv4: IpConfiguration | None = None + self._ipv6: IpConfiguration | None = None @property def id(self) -> str: @@ -68,12 +67,12 @@ class NetworkConnection(DBusInterfaceProxy): return self.properties[DBUS_ATTR_CONNECTION] @property - def ipv4(self) -> Optional[IpConfiguration]: + def ipv4(self) -> IpConfiguration | None: """Return a ip4 configuration object for the connection.""" return self._ipv4 @property - def ipv6(self) -> Optional[IpConfiguration]: + def ipv6(self) -> IpConfiguration | None: """Return a ip6 configuration object for the connection.""" return self._ipv6 diff --git a/supervisor/dbus/network/dns.py b/supervisor/dbus/network/dns.py index 46d4aad44..370be8641 100644 --- a/supervisor/dbus/network/dns.py +++ b/supervisor/dbus/network/dns.py @@ -1,7 +1,6 @@ """D-Bus interface for hostname.""" from ipaddress import ip_address import logging -from typing import Optional from ...const import ( ATTR_DOMAINS, @@ -35,17 +34,17 @@ class NetworkManagerDNS(DBusInterface): def __init__(self) -> None: """Initialize Properties.""" - self._mode: Optional[str] = None - self._rc_manager: Optional[str] = None + self._mode: str | None = None + self._rc_manager: str | None = None self._configuration: list[DNSConfiguration] = [] @property - def mode(self) -> Optional[str]: + def mode(self) -> str | None: """Return Propertie mode.""" return self._mode @property - def rc_manager(self) -> Optional[str]: + def rc_manager(self) -> str | None: """Return Propertie RcManager.""" return self._rc_manager diff --git a/supervisor/dbus/network/interface.py b/supervisor/dbus/network/interface.py index e08d419d0..340d03820 100644 --- a/supervisor/dbus/network/interface.py +++ b/supervisor/dbus/network/interface.py @@ -1,6 +1,4 @@ """NetworkInterface object for Network Manager.""" -from typing import Optional - from ...utils.dbus import DBus from ..const import ( DBUS_ATTR_ACTIVE_CONNECTION, @@ -32,9 +30,9 @@ class NetworkInterface(DBusInterfaceProxy): self.primary = False - self._connection: Optional[NetworkConnection] = None - self._settings: Optional[NetworkSetting] = None - self._wireless: Optional[NetworkWireless] = None + self._connection: NetworkConnection | None = None + self._settings: NetworkSetting | None = None + self._wireless: NetworkWireless | None = None self._nm_dbus: DBus = nm_dbus @property @@ -58,17 +56,17 @@ class NetworkInterface(DBusInterfaceProxy): return self.properties[DBUS_ATTR_MANAGED] @property - def connection(self) -> Optional[NetworkConnection]: + def connection(self) -> NetworkConnection | None: """Return the connection used for this interface.""" return self._connection @property - def settings(self) -> Optional[NetworkSetting]: + def settings(self) -> NetworkSetting | None: """Return the connection settings used for this interface.""" return self._settings @property - def wireless(self) -> Optional[NetworkWireless]: + def wireless(self) -> NetworkWireless | None: """Return the wireless data for this interface.""" return self._wireless diff --git a/supervisor/dbus/network/setting/__init__.py b/supervisor/dbus/network/setting/__init__.py index 3e2a0583a..eff2286a2 100644 --- a/supervisor/dbus/network/setting/__init__.py +++ b/supervisor/dbus/network/setting/__init__.py @@ -1,6 +1,6 @@ """Connection object for Network Manager.""" import logging -from typing import Any, Awaitable, Optional +from typing import Any, Awaitable from ....const import ATTR_METHOD, ATTR_MODE, ATTR_PSK, ATTR_SSID from ....utils.dbus import DBus @@ -59,46 +59,46 @@ class NetworkSetting(DBusInterfaceProxy): self.object_path = object_path self.properties = {} - self._connection: Optional[ConnectionProperties] = None - self._wireless: Optional[WirelessProperties] = None - self._wireless_security: Optional[WirelessSecurityProperties] = None - self._ethernet: Optional[EthernetProperties] = None - self._vlan: Optional[VlanProperties] = None - self._ipv4: Optional[IpProperties] = None - self._ipv6: Optional[IpProperties] = None + self._connection: ConnectionProperties | None = None + self._wireless: WirelessProperties | None = None + self._wireless_security: WirelessSecurityProperties | None = None + self._ethernet: EthernetProperties | None = None + self._vlan: VlanProperties | None = None + self._ipv4: IpProperties | None = None + self._ipv6: IpProperties | None = None @property - def connection(self) -> Optional[ConnectionProperties]: + def connection(self) -> ConnectionProperties | None: """Return connection properties if any.""" return self._connection @property - def wireless(self) -> Optional[WirelessProperties]: + def wireless(self) -> WirelessProperties | None: """Return wireless properties if any.""" return self._wireless @property - def wireless_security(self) -> Optional[WirelessSecurityProperties]: + def wireless_security(self) -> WirelessSecurityProperties | None: """Return wireless security properties if any.""" return self._wireless_security @property - def ethernet(self) -> Optional[EthernetProperties]: + def ethernet(self) -> EthernetProperties | None: """Return Ethernet properties if any.""" return self._ethernet @property - def vlan(self) -> Optional[VlanProperties]: + def vlan(self) -> VlanProperties | None: """Return Vlan properties if any.""" return self._vlan @property - def ipv4(self) -> Optional[IpProperties]: + def ipv4(self) -> IpProperties | None: """Return ipv4 properties if any.""" return self._ipv4 @property - def ipv6(self) -> Optional[IpProperties]: + def ipv6(self) -> IpProperties | None: """Return ipv6 properties if any.""" return self._ipv6 diff --git a/supervisor/dbus/network/wireless.py b/supervisor/dbus/network/wireless.py index b15a14791..e697bb2ba 100644 --- a/supervisor/dbus/network/wireless.py +++ b/supervisor/dbus/network/wireless.py @@ -1,5 +1,5 @@ """Connection object for Network Manager.""" -from typing import Any, Awaitable, Optional +from typing import Any, Awaitable from ...utils.dbus import DBus from ..const import ( @@ -24,10 +24,10 @@ class NetworkWireless(DBusInterfaceProxy): self.object_path = object_path self.properties = {} - self._active: Optional[NetworkWirelessAP] = None + self._active: NetworkWirelessAP | None = None @property - def active(self) -> Optional[NetworkWirelessAP]: + def active(self) -> NetworkWirelessAP | None: """Return details about active connection.""" return self._active diff --git a/supervisor/dbus/rauc.py b/supervisor/dbus/rauc.py index 6de419709..2d34cf6ce 100644 --- a/supervisor/dbus/rauc.py +++ b/supervisor/dbus/rauc.py @@ -1,6 +1,5 @@ """D-Bus interface for rauc.""" import logging -from typing import Optional from ..exceptions import DBusError, DBusInterfaceError from ..utils.dbus import DBus @@ -29,11 +28,11 @@ class Rauc(DBusInterface): def __init__(self): """Initialize Properties.""" - self._operation: Optional[str] = None - self._last_error: Optional[str] = None - self._compatible: Optional[str] = None - self._variant: Optional[str] = None - self._boot_slot: Optional[str] = None + self._operation: str | None = None + self._last_error: str | None = None + self._compatible: str | None = None + self._variant: str | None = None + self._boot_slot: str | None = None async def connect(self): """Connect to D-Bus.""" @@ -45,27 +44,27 @@ class Rauc(DBusInterface): _LOGGER.warning("Host has no rauc support. OTA updates have been disabled.") @property - def operation(self) -> Optional[str]: + def operation(self) -> str | None: """Return the current (global) operation.""" return self._operation @property - def last_error(self) -> Optional[str]: + def last_error(self) -> str | None: """Return the last message of the last error that occurred.""" return self._last_error @property - def compatible(self) -> Optional[str]: + def compatible(self) -> str | None: """Return the system compatible string.""" return self._compatible @property - def variant(self) -> Optional[str]: + def variant(self) -> str | None: """Return the system variant string.""" return self._variant @property - def boot_slot(self) -> Optional[str]: + def boot_slot(self) -> str | None: """Return the used boot slot.""" return self._boot_slot diff --git a/supervisor/docker/audio.py b/supervisor/docker/audio.py index 11667d6dc..e171761b6 100644 --- a/supervisor/docker/audio.py +++ b/supervisor/docker/audio.py @@ -1,6 +1,5 @@ """Audio docker object.""" import logging -from typing import Optional import docker @@ -63,7 +62,7 @@ class DockerAudio(DockerInterface, CoreSysAttributes): return [docker.types.Ulimit(name="rtprio", soft=10, hard=10)] @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 diff --git a/supervisor/docker/homeassistant.py b/supervisor/docker/homeassistant.py index e38579c29..d16b8d5e2 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, Optional +from typing import Awaitable from awesomeversion import AwesomeVersion, AwesomeVersionCompareException import docker @@ -23,7 +23,7 @@ class DockerHomeAssistant(DockerInterface): """Docker Supervisor wrapper for Home Assistant.""" @property - def machine(self) -> Optional[str]: + def machine(self) -> str | None: """Return machine of Home Assistant Docker image.""" if self._meta and LABEL_MACHINE in self._meta["Config"]["Labels"]: return self._meta["Config"]["Labels"][LABEL_MACHINE] diff --git a/supervisor/docker/manager.py b/supervisor/docker/manager.py index ec950adbc..afff6059b 100644 --- a/supervisor/docker/manager.py +++ b/supervisor/docker/manager.py @@ -4,7 +4,7 @@ from ipaddress import IPv4Address import logging import os from pathlib import Path -from typing import Any, Optional +from typing import Any import attr from awesomeversion import AwesomeVersion, AwesomeVersionCompareException @@ -154,16 +154,16 @@ class DockerAPI: image: str, tag: str = "latest", dns: bool = True, - ipv4: Optional[IPv4Address] = None, + ipv4: IPv4Address | None = None, **kwargs: Any, ) -> Container: """Create a Docker container and run it. Need run inside executor. """ - name: Optional[str] = kwargs.get("name") - network_mode: Optional[str] = kwargs.get("network_mode") - hostname: Optional[str] = kwargs.get("hostname") + name: str | None = kwargs.get("name") + network_mode: str | None = kwargs.get("network_mode") + hostname: str | None = kwargs.get("hostname") if "labels" not in kwargs: kwargs["labels"] = {} @@ -242,7 +242,7 @@ class DockerAPI: self, image: str, tag: str = "latest", - command: Optional[str] = None, + command: str | None = None, **kwargs: Any, ) -> CommandReturn: """Create a temporary container and run command. diff --git a/supervisor/docker/monitor.py b/supervisor/docker/monitor.py index 8807e0549..fbbe71d1c 100644 --- a/supervisor/docker/monitor.py +++ b/supervisor/docker/monitor.py @@ -2,7 +2,6 @@ from dataclasses import dataclass import logging from threading import Thread -from typing import Optional from docker.models.containers import Container from docker.types.daemon import CancellableStream @@ -31,7 +30,7 @@ class DockerMonitor(CoreSysAttributes, Thread): """Initialize Docker monitor object.""" super().__init__() self.coresys = coresys - self._events: Optional[CancellableStream] = None + self._events: CancellableStream | None = None self._unlabeled_managed_containers: list[str] = [] def watch_container(self, container: Container): @@ -64,7 +63,7 @@ class DockerMonitor(CoreSysAttributes, Thread): LABEL_MANAGED in attributes or attributes.get("name") in self._unlabeled_managed_containers ): - container_state: Optional[ContainerState] = None + container_state: ContainerState | None = None action: str = event["Action"] if action == "start": diff --git a/supervisor/docker/network.py b/supervisor/docker/network.py index 7b6179bbd..e479581d4 100644 --- a/supervisor/docker/network.py +++ b/supervisor/docker/network.py @@ -2,7 +2,6 @@ from contextlib import suppress from ipaddress import IPv4Address import logging -from typing import Optional import docker import requests @@ -99,8 +98,8 @@ class DockerNetwork: def attach_container( self, container: docker.models.containers.Container, - alias: Optional[list[str]] = None, - ipv4: Optional[IPv4Address] = None, + alias: list[str] | None = None, + ipv4: IPv4Address | None = None, ) -> None: """Attach container to Supervisor network. diff --git a/supervisor/exceptions.py b/supervisor/exceptions.py index 8b516d680..7590b08be 100644 --- a/supervisor/exceptions.py +++ b/supervisor/exceptions.py @@ -1,7 +1,5 @@ """Core Exceptions.""" - - -from typing import Callable, Optional +from typing import Callable class HassioError(Exception): @@ -9,8 +7,8 @@ class HassioError(Exception): def __init__( self, - message: Optional[str] = None, - logger: Optional[Callable[..., None]] = None, + message: str | None = None, + logger: Callable[..., None] | None = None, ) -> None: """Raise & log.""" if logger is not None and message is not None: diff --git a/supervisor/hardware/disk.py b/supervisor/hardware/disk.py index 5941d73c5..67ab696ff 100644 --- a/supervisor/hardware/disk.py +++ b/supervisor/hardware/disk.py @@ -2,7 +2,6 @@ import logging from pathlib import Path import shutil -from typing import Union from ..coresys import CoreSys, CoreSysAttributes from ..exceptions import HardwareNotFound @@ -48,17 +47,17 @@ class HwDisk(CoreSysAttributes): return False - def get_disk_total_space(self, path: Union[str, Path]) -> float: + def get_disk_total_space(self, path: str | Path) -> float: """Return total space (GiB) on disk for path.""" total, _, _ = shutil.disk_usage(path) return round(total / (1024.0**3), 1) - def get_disk_used_space(self, path: Union[str, Path]) -> float: + def get_disk_used_space(self, path: str | Path) -> float: """Return used space (GiB) on disk for path.""" _, used, _ = shutil.disk_usage(path) return round(used / (1024.0**3), 1) - def get_disk_free_space(self, path: Union[str, Path]) -> float: + def get_disk_free_space(self, path: str | Path) -> float: """Return free space (GiB) on disk for path.""" _, _, free = shutil.disk_usage(path) return round(free / (1024.0**3), 1) @@ -112,7 +111,7 @@ class HwDisk(CoreSysAttributes): # Return the pessimistic estimate (0x02 -> 10%-20%, return 20%) return life_time_value * 10.0 - def get_disk_life_time(self, path: Union[str, Path]) -> float: + def get_disk_life_time(self, path: str | Path) -> float: """Return life time estimate of the underlying SSD drive.""" mount_source = self._get_mount_source(str(path)) if mount_source == "overlay": diff --git a/supervisor/hardware/helper.py b/supervisor/hardware/helper.py index f2f70a53d..156692848 100644 --- a/supervisor/hardware/helper.py +++ b/supervisor/hardware/helper.py @@ -3,7 +3,6 @@ from datetime import datetime import logging from pathlib import Path import re -from typing import Optional import pyudev @@ -42,7 +41,7 @@ class HwHelper(CoreSysAttributes): return bool(self.sys_hardware.filter_devices(subsystem=UdevSubsystem.USB)) @property - def last_boot(self) -> Optional[str]: + def last_boot(self) -> str | None: """Return last boot time.""" try: stats: str = _PROC_STAT.read_text(encoding="utf-8") @@ -51,7 +50,7 @@ class HwHelper(CoreSysAttributes): return None # parse stat file - found: Optional[re.Match] = _RE_BOOT_TIME.search(stats) + found: re.Match | None = _RE_BOOT_TIME.search(stats) if not found: _LOGGER.error("Can't found last boot time!") return None diff --git a/supervisor/hardware/manager.py b/supervisor/hardware/manager.py index 686c214e6..6af0481a7 100644 --- a/supervisor/hardware/manager.py +++ b/supervisor/hardware/manager.py @@ -1,7 +1,6 @@ """Hardware Manager of Supervisor.""" import logging from pathlib import Path -from typing import Optional import pyudev @@ -65,7 +64,7 @@ class HardwareManager(CoreSysAttributes): return device raise HardwareNotFound() - def filter_devices(self, subsystem: Optional[UdevSubsystem] = None) -> list[Device]: + def filter_devices(self, subsystem: UdevSubsystem | None = None) -> list[Device]: """Return a filtered list.""" devices = set() for device in self.devices: diff --git a/supervisor/hardware/monitor.py b/supervisor/hardware/monitor.py index 39077f9a0..fcc7da6e3 100644 --- a/supervisor/hardware/monitor.py +++ b/supervisor/hardware/monitor.py @@ -3,7 +3,6 @@ import asyncio import logging from pathlib import Path from pprint import pformat -from typing import Optional import pyudev @@ -24,8 +23,8 @@ class HwMonitor(CoreSysAttributes): """Initialize Hardware Monitor object.""" self.coresys: CoreSys = coresys self.context = pyudev.Context() - self.monitor: Optional[pyudev.Monitor] = None - self.observer: Optional[pyudev.MonitorObserver] = None + self.monitor: pyudev.Monitor | None = None + self.observer: pyudev.MonitorObserver | None = None async def load(self) -> None: """Start hardware monitor.""" @@ -70,8 +69,8 @@ class HwMonitor(CoreSysAttributes): ): return - hw_action: Optional[HardwareAction] = None - device: Optional[Device] = None + hw_action: HardwareAction | None = None + device: Device | None = None ## # Remove diff --git a/supervisor/homeassistant/api.py b/supervisor/homeassistant/api.py index 5a8852a22..f26d8d7d5 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, Optional +from typing import Any, AsyncContextManager import aiohttp from aiohttp import hdrs @@ -26,8 +26,8 @@ class HomeAssistantAPI(CoreSysAttributes): self.coresys: CoreSys = coresys # We don't persist access tokens. Instead we fetch new ones when needed - self.access_token: Optional[str] = None - self._access_token_expires: Optional[datetime] = None + self.access_token: str | None = None + self._access_token_expires: datetime | None = None @Job(limit=JobExecutionLimit.SINGLE_WAIT) async def ensure_access_token(self) -> None: @@ -65,12 +65,12 @@ class HomeAssistantAPI(CoreSysAttributes): self, method: str, path: str, - json: Optional[dict[str, Any]] = None, - content_type: Optional[str] = None, + json: dict[str, Any] | None = None, + content_type: str | None = None, data: Any = None, timeout: int = 30, - params: Optional[dict[str, str]] = None, - headers: Optional[dict[str, str]] = None, + params: dict[str, str] | None = None, + headers: dict[str, str] | None = None, ) -> AsyncContextManager[aiohttp.ClientResponse]: """Async context manager to make a request with right auth.""" url = f"{self.sys_homeassistant.api_url}/{path}" diff --git a/supervisor/homeassistant/core.py b/supervisor/homeassistant/core.py index 7d6792dab..0174e6494 100644 --- a/supervisor/homeassistant/core.py +++ b/supervisor/homeassistant/core.py @@ -5,7 +5,7 @@ import logging import re import secrets import shutil -from typing import Awaitable, Optional +from typing import Awaitable import attr from awesomeversion import AwesomeVersion @@ -189,8 +189,8 @@ class HomeAssistantCore(CoreSysAttributes): ) async def update( self, - version: Optional[AwesomeVersion] = None, - backup: Optional[bool] = False, + version: AwesomeVersion | None = None, + backup: bool | None = False, ) -> None: """Update HomeAssistant version.""" version = version or self.sys_homeassistant.latest_version diff --git a/supervisor/homeassistant/module.py b/supervisor/homeassistant/module.py index 9faa97656..0962c3e99 100644 --- a/supervisor/homeassistant/module.py +++ b/supervisor/homeassistant/module.py @@ -6,7 +6,6 @@ from pathlib import Path import shutil import tarfile from tempfile import TemporaryDirectory -from typing import Optional from uuid import UUID from awesomeversion import AwesomeVersion, AwesomeVersionException @@ -158,7 +157,7 @@ class HomeAssistant(FileConfiguration, CoreSysAttributes): self._data[ATTR_WATCHDOG] = value @property - def latest_version(self) -> Optional[AwesomeVersion]: + def latest_version(self) -> AwesomeVersion | None: """Return last available version of Home Assistant.""" return self.sys_updater.version_homeassistant @@ -170,12 +169,12 @@ class HomeAssistant(FileConfiguration, CoreSysAttributes): return f"ghcr.io/home-assistant/{self.sys_machine}-homeassistant" @image.setter - def image(self, value: Optional[str]) -> None: + def image(self, value: str | None) -> None: """Set image name of Home Assistant container.""" self._data[ATTR_IMAGE] = value @property - def version(self) -> Optional[AwesomeVersion]: + def version(self) -> AwesomeVersion | None: """Return version of local version.""" return self._data.get(ATTR_VERSION) @@ -200,7 +199,7 @@ class HomeAssistant(FileConfiguration, CoreSysAttributes): return self._data[ATTR_UUID] @property - def supervisor_token(self) -> Optional[str]: + def supervisor_token(self) -> str | None: """Return an access token for the Supervisor API.""" return self._data.get(ATTR_ACCESS_TOKEN) @@ -210,12 +209,12 @@ class HomeAssistant(FileConfiguration, CoreSysAttributes): self._data[ATTR_ACCESS_TOKEN] = value @property - def refresh_token(self) -> Optional[str]: + def refresh_token(self) -> str | None: """Return the refresh token to authenticate with Home Assistant.""" return self._data.get(ATTR_REFRESH_TOKEN) @refresh_token.setter - def refresh_token(self, value: Optional[str]): + def refresh_token(self, value: str | None): """Set Home Assistant refresh_token.""" self._data[ATTR_REFRESH_TOKEN] = value @@ -230,22 +229,22 @@ class HomeAssistant(FileConfiguration, CoreSysAttributes): return Path(self.sys_config.path_extern_tmp, "homeassistant_pulse") @property - def audio_output(self) -> Optional[str]: + def audio_output(self) -> str | None: """Return a pulse profile for output or None.""" return self._data[ATTR_AUDIO_OUTPUT] @audio_output.setter - def audio_output(self, value: Optional[str]): + def audio_output(self, value: str | None): """Set audio output profile settings.""" self._data[ATTR_AUDIO_OUTPUT] = value @property - def audio_input(self) -> Optional[str]: + def audio_input(self) -> str | None: """Return pulse profile for input or None.""" return self._data[ATTR_AUDIO_INPUT] @audio_input.setter - def audio_input(self, value: Optional[str]): + def audio_input(self, value: str | None): """Set audio input settings.""" self._data[ATTR_AUDIO_INPUT] = value diff --git a/supervisor/homeassistant/secrets.py b/supervisor/homeassistant/secrets.py index 09f91b7ad..250840df9 100644 --- a/supervisor/homeassistant/secrets.py +++ b/supervisor/homeassistant/secrets.py @@ -2,7 +2,6 @@ from datetime import timedelta import logging from pathlib import Path -from typing import Optional, Union from ..coresys import CoreSys, CoreSysAttributes from ..exceptions import YamlFileError @@ -19,14 +18,14 @@ 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, bool | float | int | str] = {} @property def path_secrets(self) -> Path: """Return path to secret file.""" return Path(self.sys_config.path_homeassistant, "secrets.yaml") - def get(self, secret: str) -> Optional[Union[bool, float, int, str]]: + def get(self, secret: str) -> bool | float | int | str | None: """Get secret from store.""" _LOGGER.info("Request secret %s", secret) return self.secrets.get(secret) diff --git a/supervisor/host/info.py b/supervisor/host/info.py index 85547e2e8..0a840afdd 100644 --- a/supervisor/host/info.py +++ b/supervisor/host/info.py @@ -2,7 +2,6 @@ import asyncio from datetime import datetime import logging -from typing import Optional from ..coresys import CoreSysAttributes from ..dbus.const import MulticastProtocolEnabled @@ -19,86 +18,86 @@ class InfoCenter(CoreSysAttributes): self.coresys = coresys @property - def hostname(self) -> Optional[str]: + def hostname(self) -> str | None: """Return local hostname.""" return self.sys_dbus.hostname.hostname @property - def llmnr_hostname(self) -> Optional[str]: + def llmnr_hostname(self) -> str | None: """Return local llmnr hostname.""" return self.sys_dbus.resolved.llmnr_hostname @property - def broadcast_llmnr(self) -> Optional[bool]: + def broadcast_llmnr(self) -> bool | None: """Host is broadcasting llmnr name.""" if self.sys_dbus.resolved.llmnr: return self.sys_dbus.resolved.llmnr == MulticastProtocolEnabled.YES return None @property - def broadcast_mdns(self) -> Optional[bool]: + def broadcast_mdns(self) -> bool | None: """Host is broadcasting mdns name.""" if self.sys_dbus.resolved.multicast_dns: return self.sys_dbus.resolved.multicast_dns == MulticastProtocolEnabled.YES return None @property - def chassis(self) -> Optional[str]: + def chassis(self) -> str | None: """Return local chassis type.""" return self.sys_dbus.hostname.chassis @property - def deployment(self) -> Optional[str]: + def deployment(self) -> str | None: """Return local deployment type.""" return self.sys_dbus.hostname.deployment @property - def kernel(self) -> Optional[str]: + def kernel(self) -> str | None: """Return local kernel version.""" return self.sys_dbus.hostname.kernel @property - def operating_system(self) -> Optional[str]: + def operating_system(self) -> str | None: """Return local operating system.""" return self.sys_dbus.hostname.operating_system @property - def cpe(self) -> Optional[str]: + def cpe(self) -> str | None: """Return local CPE.""" return self.sys_dbus.hostname.cpe @property - def timezone(self) -> Optional[str]: + def timezone(self) -> str | None: """Return host timezone.""" return self.sys_dbus.timedate.timezone @property - def dt_utc(self) -> Optional[datetime]: + def dt_utc(self) -> datetime | None: """Return host UTC time.""" return self.sys_dbus.timedate.dt_utc @property - def use_rtc(self) -> Optional[bool]: + def use_rtc(self) -> bool | None: """Return true if host have an RTC.""" return self.sys_dbus.timedate.local_rtc @property - def use_ntp(self) -> Optional[bool]: + def use_ntp(self) -> bool | None: """Return true if host using NTP.""" return self.sys_dbus.timedate.ntp @property - def dt_synchronized(self) -> Optional[bool]: + def dt_synchronized(self) -> bool | None: """Return true if host time is syncronized.""" return self.sys_dbus.timedate.ntp_synchronized @property - def startup_time(self) -> Optional[float]: + def startup_time(self) -> float | None: """Return startup time in seconds.""" return self.sys_dbus.systemd.startup_time @property - def boot_timestamp(self) -> Optional[int]: + def boot_timestamp(self) -> int | None: """Return the boot timestamp.""" return self.sys_dbus.systemd.boot_timestamp diff --git a/supervisor/host/sound.py b/supervisor/host/sound.py index 1b359a1b2..abca5f47f 100644 --- a/supervisor/host/sound.py +++ b/supervisor/host/sound.py @@ -2,7 +2,6 @@ from datetime import timedelta from enum import Enum import logging -from typing import Optional import attr from pulsectl import Pulse, PulseError, PulseIndexError, PulseOperationFailed @@ -47,7 +46,7 @@ class AudioStream: volume: float = attr.ib() mute: bool = attr.ib() default: bool = attr.ib() - card: Optional[int] = attr.ib() + card: int | None = attr.ib() applications: list[AudioApplication] = attr.ib() diff --git a/supervisor/ingress.py b/supervisor/ingress.py index 62e93151d..626c42137 100644 --- a/supervisor/ingress.py +++ b/supervisor/ingress.py @@ -3,7 +3,6 @@ from datetime import timedelta import logging import random import secrets -from typing import Optional from .addons.addon import Addon from .const import ATTR_PORTS, ATTR_SESSION, FILE_HASSIO_INGRESS @@ -25,7 +24,7 @@ class Ingress(FileConfiguration, CoreSysAttributes): self.coresys: CoreSys = coresys self.tokens: dict[str, str] = {} - def get(self, token: str) -> Optional[Addon]: + def get(self, token: str) -> Addon | None: """Return addon they have this ingress token.""" if token not in self.tokens: return None diff --git a/supervisor/jobs/__init__.py b/supervisor/jobs/__init__.py index 3157c67bb..6b2bb8d11 100644 --- a/supervisor/jobs/__init__.py +++ b/supervisor/jobs/__init__.py @@ -1,6 +1,5 @@ """Supervisor job manager.""" import logging -from typing import Optional from ..coresys import CoreSys, CoreSysAttributes from ..utils.common import FileConfiguration @@ -18,7 +17,7 @@ class SupervisorJob(CoreSysAttributes): self.coresys: CoreSys = coresys self.name: str = name self._progress: int = 0 - self._stage: Optional[str] = None + self._stage: str | None = None @property def progress(self) -> int: @@ -26,13 +25,11 @@ class SupervisorJob(CoreSysAttributes): return self._progress @property - def stage(self) -> Optional[str]: + def stage(self) -> str | None: """Return the current stage.""" return self._stage - def update( - self, progress: Optional[int] = None, stage: Optional[str] = None - ) -> None: + def update(self, progress: int | None = None, stage: str | None = None) -> None: """Update the job object.""" if progress is not None: if progress >= round(100): diff --git a/supervisor/jobs/decorator.py b/supervisor/jobs/decorator.py index 8069c9e3e..bd62e3c71 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, Optional +from typing import Any import sentry_sdk @@ -22,13 +22,13 @@ class Job(CoreSysAttributes): def __init__( self, - name: Optional[str] = None, - conditions: Optional[list[JobCondition]] = None, + name: str | None = None, + conditions: list[JobCondition] | None = None, cleanup: bool = True, - on_condition: Optional[JobException] = None, - limit: Optional[JobExecutionLimit] = None, - throttle_period: Optional[timedelta] = None, - throttle_max_calls: Optional[int] = None, + on_condition: JobException | None = None, + limit: JobExecutionLimit | None = None, + throttle_period: timedelta | None = None, + throttle_max_calls: int | None = None, ): """Initialize the Job class.""" self.name = name @@ -38,10 +38,10 @@ class Job(CoreSysAttributes): self.limit = limit self.throttle_period = throttle_period self.throttle_max_calls = throttle_max_calls - self._lock: Optional[asyncio.Semaphore] = None + self._lock: asyncio.Semaphore | None = None self._method = None self._last_call = datetime.min - self._rate_limited_calls: Optional[list[datetime]] = None + self._rate_limited_calls: list[datetime] | None = None # Validate Options if ( diff --git a/supervisor/misc/scheduler.py b/supervisor/misc/scheduler.py index ef1803b2b..1fe2a841f 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, Optional, Union +from typing import Awaitable, Callable from uuid import UUID, uuid4 import async_timeout @@ -20,10 +20,10 @@ class _Task: id: UUID = attr.ib() coro_callback: Callable[..., Awaitable[None]] = attr.ib(eq=False) - interval: Union[float, time] = attr.ib(eq=False) + interval: float | time = attr.ib(eq=False) repeat: bool = attr.ib(eq=False) - job: Optional[asyncio.tasks.Task] = attr.ib(eq=False) - next: Optional[asyncio.TimerHandle] = attr.ib(eq=False) + job: asyncio.tasks.Task | None = attr.ib(eq=False) + next: asyncio.TimerHandle | None = attr.ib(eq=False) class Scheduler(CoreSysAttributes): @@ -37,7 +37,7 @@ class Scheduler(CoreSysAttributes): def register_task( self, coro_callback: Callable[..., Awaitable[None]], - interval: Union[float, time], + interval: float | time, repeat: bool = True, ) -> UUID: """Schedule a coroutine. diff --git a/supervisor/os/data_disk.py b/supervisor/os/data_disk.py index 31e30a574..744298a52 100644 --- a/supervisor/os/data_disk.py +++ b/supervisor/os/data_disk.py @@ -1,7 +1,6 @@ """Home Assistant Operating-System DataDisk.""" import logging from pathlib import Path -from typing import Optional from awesomeversion import AwesomeVersion @@ -29,7 +28,7 @@ class DataDisk(CoreSysAttributes): self.coresys = coresys @property - def disk_used(self) -> Optional[Path]: + def disk_used(self) -> Path | None: """Return Path to used Disk for data.""" return self.sys_dbus.agent.datadisk.current_device diff --git a/supervisor/os/manager.py b/supervisor/os/manager.py index 04aea2395..960d600c0 100644 --- a/supervisor/os/manager.py +++ b/supervisor/os/manager.py @@ -2,7 +2,7 @@ import asyncio import logging from pathlib import Path -from typing import Awaitable, Optional +from typing import Awaitable import aiohttp from awesomeversion import AwesomeVersion, AwesomeVersionException @@ -26,9 +26,9 @@ class OSManager(CoreSysAttributes): self.coresys: CoreSys = coresys self._datadisk: DataDisk = DataDisk(coresys) self._available: bool = False - self._version: Optional[AwesomeVersion] = None - self._board: Optional[str] = None - self._os_name: Optional[str] = None + self._version: AwesomeVersion | None = None + self._board: str | None = None + self._os_name: str | None = None @property def available(self) -> bool: @@ -36,12 +36,12 @@ class OSManager(CoreSysAttributes): return self._available @property - def version(self) -> Optional[AwesomeVersion]: + def version(self) -> AwesomeVersion | None: """Return version of HassOS.""" return self._version @property - def latest_version(self) -> Optional[AwesomeVersion]: + def latest_version(self) -> AwesomeVersion | None: """Return version of HassOS.""" return self.sys_updater.version_hassos @@ -54,12 +54,12 @@ class OSManager(CoreSysAttributes): return False @property - def board(self) -> Optional[str]: + def board(self) -> str | None: """Return board name.""" return self._board @property - def os_name(self) -> Optional[str]: + def os_name(self) -> str | None: """Return OS name.""" return self._os_name @@ -178,7 +178,7 @@ class OSManager(CoreSysAttributes): limit=JobExecutionLimit.ONCE, on_condition=HassOSJobError, ) - async def update(self, version: Optional[AwesomeVersion] = None) -> None: + async def update(self, version: AwesomeVersion | None = None) -> None: """Update HassOS system.""" version = version or self.latest_version diff --git a/supervisor/plugins/audio.py b/supervisor/plugins/audio.py index e21402ead..df6b6d81a 100644 --- a/supervisor/plugins/audio.py +++ b/supervisor/plugins/audio.py @@ -7,7 +7,6 @@ from contextlib import suppress import logging from pathlib import Path, PurePath import shutil -from typing import Optional from awesomeversion import AwesomeVersion import jinja2 @@ -38,8 +37,10 @@ from .validate import SCHEMA_AUDIO_CONFIG _LOGGER: logging.Logger = logging.getLogger(__name__) +# pylint: disable=no-member PULSE_CLIENT_TMPL: Path = Path(__file__).parents[1].joinpath("data/pulse-client.tmpl") ASOUND_TMPL: Path = Path(__file__).parents[1].joinpath("data/asound.tmpl") +# pylint: enable=no-member class PluginAudio(PluginBase): @@ -51,7 +52,7 @@ class PluginAudio(PluginBase): self.slug = "audio" self.coresys: CoreSys = coresys self.instance: DockerAudio = DockerAudio(coresys) - self.client_template: Optional[jinja2.Template] = None + self.client_template: jinja2.Template | None = None @property def path_extern_pulse(self) -> PurePath: @@ -69,7 +70,7 @@ class PluginAudio(PluginBase): return Path(self.sys_config.path_audio, "pulse_audio.json") @property - def latest_version(self) -> Optional[AwesomeVersion]: + def latest_version(self) -> AwesomeVersion | None: """Return latest version of Audio.""" return self.sys_updater.version_audio @@ -117,7 +118,7 @@ class PluginAudio(PluginBase): conditions=PLUGIN_UPDATE_CONDITIONS, on_condition=AudioJobError, ) - async def update(self, version: Optional[str] = None) -> None: + async def update(self, version: str | None = None) -> None: """Update Audio plugin.""" version = version or self.latest_version old_image = self.image diff --git a/supervisor/plugins/base.py b/supervisor/plugins/base.py index c69f138bb..aa1bd0e49 100644 --- a/supervisor/plugins/base.py +++ b/supervisor/plugins/base.py @@ -3,7 +3,7 @@ from abc import ABC, abstractmethod import asyncio from contextlib import suppress import logging -from typing import Awaitable, Optional +from typing import Awaitable from awesomeversion import AwesomeVersion, AwesomeVersionException @@ -26,7 +26,7 @@ class PluginBase(ABC, FileConfiguration, CoreSysAttributes): instance: DockerInterface @property - def version(self) -> Optional[AwesomeVersion]: + def version(self) -> AwesomeVersion | None: """Return current version of the plugin.""" return self._data.get(ATTR_VERSION) @@ -49,7 +49,7 @@ class PluginBase(ABC, FileConfiguration, CoreSysAttributes): @property @abstractmethod - def latest_version(self) -> Optional[AwesomeVersion]: + def latest_version(self) -> AwesomeVersion | None: """Return latest version of the plugin.""" @property @@ -189,7 +189,7 @@ class PluginBase(ABC, FileConfiguration, CoreSysAttributes): """Install system plugin.""" @abstractmethod - async def update(self, version: Optional[str] = None) -> None: + async def update(self, version: str | None = None) -> None: """Update system plugin.""" @abstractmethod diff --git a/supervisor/plugins/cli.py b/supervisor/plugins/cli.py index 2367dc103..cc6f28f88 100644 --- a/supervisor/plugins/cli.py +++ b/supervisor/plugins/cli.py @@ -6,7 +6,7 @@ import asyncio from contextlib import suppress import logging import secrets -from typing import Awaitable, Optional +from typing import Awaitable from awesomeversion import AwesomeVersion @@ -41,7 +41,7 @@ class PluginCli(PluginBase): self.instance: DockerCli = DockerCli(coresys) @property - def latest_version(self) -> Optional[AwesomeVersion]: + def latest_version(self) -> AwesomeVersion | None: """Return version of latest cli.""" return self.sys_updater.version_cli @@ -77,7 +77,7 @@ class PluginCli(PluginBase): conditions=PLUGIN_UPDATE_CONDITIONS, on_condition=CliJobError, ) - async def update(self, version: Optional[AwesomeVersion] = None) -> None: + async def update(self, version: AwesomeVersion | None = None) -> None: """Update local HA cli.""" version = version or self.latest_version old_image = self.image diff --git a/supervisor/plugins/dns.py b/supervisor/plugins/dns.py index 24a947674..9fff92c43 100644 --- a/supervisor/plugins/dns.py +++ b/supervisor/plugins/dns.py @@ -7,7 +7,6 @@ from contextlib import suppress from ipaddress import IPv4Address import logging from pathlib import Path -from typing import Optional import attr from awesomeversion import AwesomeVersion @@ -45,8 +44,10 @@ from .validate import SCHEMA_DNS_CONFIG _LOGGER: logging.Logger = logging.getLogger(__name__) +# pylint: disable=no-member HOSTS_TMPL: Path = Path(__file__).parents[1].joinpath("data/hosts.tmpl") RESOLV_TMPL: Path = Path(__file__).parents[1].joinpath("data/resolv.tmpl") +# pylint: enable=no-member HOST_RESOLV: Path = Path("/etc/resolv.conf") @@ -67,8 +68,8 @@ class PluginDns(PluginBase): self.slug = "dns" self.coresys: CoreSys = coresys self.instance: DockerDNS = DockerDNS(coresys) - self.resolv_template: Optional[jinja2.Template] = None - self.hosts_template: Optional[jinja2.Template] = None + self.resolv_template: jinja2.Template | None = None + self.hosts_template: jinja2.Template | None = None self._hosts: list[HostEntry] = [] self._loop: bool = False @@ -106,7 +107,7 @@ class PluginDns(PluginBase): self._data[ATTR_SERVERS] = value @property - def latest_version(self) -> Optional[AwesomeVersion]: + def latest_version(self) -> AwesomeVersion | None: """Return latest version of CoreDNS.""" return self.sys_updater.version_dns @@ -184,7 +185,7 @@ class PluginDns(PluginBase): conditions=PLUGIN_UPDATE_CONDITIONS, on_condition=CoreDNSJobError, ) - async def update(self, version: Optional[AwesomeVersion] = None) -> None: + async def update(self, version: AwesomeVersion | None = None) -> None: """Update CoreDNS plugin.""" version = version or self.latest_version old_image = self.image @@ -390,7 +391,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]) -> HostEntry | None: """Search a host entry.""" for entry in self._hosts: for name in names: diff --git a/supervisor/plugins/multicast.py b/supervisor/plugins/multicast.py index 652e9dfc4..aa335bd16 100644 --- a/supervisor/plugins/multicast.py +++ b/supervisor/plugins/multicast.py @@ -5,7 +5,6 @@ Code: https://github.com/home-assistant/plugin-multicast import asyncio from contextlib import suppress import logging -from typing import Optional from awesomeversion import AwesomeVersion @@ -44,7 +43,7 @@ class PluginMulticast(PluginBase): self.instance: DockerMulticast = DockerMulticast(coresys) @property - def latest_version(self) -> Optional[AwesomeVersion]: + def latest_version(self) -> AwesomeVersion | None: """Return latest version of Multicast.""" return self.sys_updater.version_multicast @@ -74,7 +73,7 @@ class PluginMulticast(PluginBase): conditions=PLUGIN_UPDATE_CONDITIONS, on_condition=MulticastJobError, ) - async def update(self, version: Optional[AwesomeVersion] = None) -> None: + async def update(self, version: AwesomeVersion | None = None) -> None: """Update Multicast plugin.""" version = version or self.latest_version old_image = self.image diff --git a/supervisor/plugins/observer.py b/supervisor/plugins/observer.py index faa5507b0..171eb3570 100644 --- a/supervisor/plugins/observer.py +++ b/supervisor/plugins/observer.py @@ -6,7 +6,6 @@ import asyncio from contextlib import suppress import logging import secrets -from typing import Optional import aiohttp from awesomeversion import AwesomeVersion @@ -47,7 +46,7 @@ class PluginObserver(PluginBase): self.instance: DockerObserver = DockerObserver(coresys) @property - def latest_version(self) -> Optional[AwesomeVersion]: + def latest_version(self) -> AwesomeVersion | None: """Return version of latest observer.""" return self.sys_updater.version_observer @@ -82,7 +81,7 @@ class PluginObserver(PluginBase): conditions=PLUGIN_UPDATE_CONDITIONS, on_condition=ObserverJobError, ) - async def update(self, version: Optional[AwesomeVersion] = None) -> None: + async def update(self, version: AwesomeVersion | None = None) -> None: """Update local HA observer.""" version = version or self.latest_version old_image = self.image diff --git a/supervisor/resolution/checks/addon_pwned.py b/supervisor/resolution/checks/addon_pwned.py index 32fd2d4ac..fbd48b7b4 100644 --- a/supervisor/resolution/checks/addon_pwned.py +++ b/supervisor/resolution/checks/addon_pwned.py @@ -1,7 +1,6 @@ """Helpers to check core security.""" from datetime import timedelta import logging -from typing import Optional from ...const import AddonState, CoreState from ...coresys import CoreSys @@ -64,7 +63,7 @@ class CheckAddonPwned(CheckBase): pass @Job(conditions=[JobCondition.INTERNET_SYSTEM]) - async def approve_check(self, reference: Optional[str] = None) -> bool: + async def approve_check(self, reference: str | None = None) -> bool: """Approve check if it is affected by issue.""" addon = self.sys_addons.get(reference) diff --git a/supervisor/resolution/checks/base.py b/supervisor/resolution/checks/base.py index 68b6fafbc..203e75a72 100644 --- a/supervisor/resolution/checks/base.py +++ b/supervisor/resolution/checks/base.py @@ -1,7 +1,6 @@ """Baseclass for system checks.""" from abc import ABC, abstractmethod import logging -from typing import Optional from ...const import ATTR_ENABLED, CoreState from ...coresys import CoreSys, CoreSysAttributes @@ -60,7 +59,7 @@ class CheckBase(ABC, CoreSysAttributes): """Run check if not affected by issue.""" @abstractmethod - async def approve_check(self, reference: Optional[str] = None) -> bool: + async def approve_check(self, reference: str | None = None) -> bool: """Approve check if it is affected by issue.""" @property diff --git a/supervisor/resolution/checks/core_security.py b/supervisor/resolution/checks/core_security.py index e6740c6fa..99e5548e0 100644 --- a/supervisor/resolution/checks/core_security.py +++ b/supervisor/resolution/checks/core_security.py @@ -1,7 +1,6 @@ """Helpers to check core security.""" from enum import Enum from pathlib import Path -from typing import Optional from awesomeversion import AwesomeVersion, AwesomeVersionException @@ -42,7 +41,7 @@ class CheckCoreSecurity(CheckBase): except (AwesomeVersionException, OSError): return - async def approve_check(self, reference: Optional[str] = None) -> bool: + async def approve_check(self, reference: str | None = None) -> bool: """Approve check if it is affected by issue.""" try: if self.sys_homeassistant.version >= AwesomeVersion("2021.1.5"): diff --git a/supervisor/resolution/checks/dns_server_failure.py b/supervisor/resolution/checks/dns_server_failure.py index 686d12754..8d559e3f3 100644 --- a/supervisor/resolution/checks/dns_server_failure.py +++ b/supervisor/resolution/checks/dns_server_failure.py @@ -1,7 +1,6 @@ """Helpers to check DNS servers for failure.""" import asyncio from datetime import timedelta -from typing import Optional from aiodns import DNSResolver from aiodns.error import DNSError @@ -43,7 +42,7 @@ class CheckDNSServerFailures(CheckBase): self.sys_capture_exception(results[i]) @Job(conditions=[JobCondition.INTERNET_SYSTEM]) - async def approve_check(self, reference: Optional[str] = None) -> bool: + async def approve_check(self, reference: str | None = None) -> bool: """Approve check if it is affected by issue.""" if reference not in self.dns_servers: return False diff --git a/supervisor/resolution/checks/dns_server_ipv6_error.py b/supervisor/resolution/checks/dns_server_ipv6_error.py index dd3c787cd..641b033e2 100644 --- a/supervisor/resolution/checks/dns_server_ipv6_error.py +++ b/supervisor/resolution/checks/dns_server_ipv6_error.py @@ -1,7 +1,6 @@ """Helpers to check DNS servers for IPv6 errors.""" import asyncio from datetime import timedelta -from typing import Optional from aiodns import DNSResolver from aiodns.error import DNSError @@ -48,7 +47,7 @@ class CheckDNSServerIPv6Errors(CheckBase): self.sys_capture_exception(results[i]) @Job(conditions=[JobCondition.INTERNET_SYSTEM]) - async def approve_check(self, reference: Optional[str] = None) -> bool: + async def approve_check(self, reference: str | None = None) -> bool: """Approve check if it is affected by issue.""" if reference not in self.dns_servers: return False diff --git a/supervisor/resolution/checks/free_space.py b/supervisor/resolution/checks/free_space.py index 387ee14a9..aeb17cca3 100644 --- a/supervisor/resolution/checks/free_space.py +++ b/supervisor/resolution/checks/free_space.py @@ -1,6 +1,4 @@ """Helpers to check and fix issues with free space.""" -from typing import Optional - from ...backups.const import BackupType from ...const import CoreState from ...coresys import CoreSys @@ -50,7 +48,7 @@ class CheckFreeSpace(CheckBase): IssueType.FREE_SPACE, ContextType.SYSTEM, suggestions=suggestions ) - async def approve_check(self, reference: Optional[str] = None) -> bool: + async def approve_check(self, reference: str | None = None) -> bool: """Approve check if it is affected by issue.""" if self.sys_host.info.free_space > MINIMUM_FREE_SPACE_THRESHOLD: return False diff --git a/supervisor/resolution/checks/supervisor_trust.py b/supervisor/resolution/checks/supervisor_trust.py index 427760f0e..f7e204c61 100644 --- a/supervisor/resolution/checks/supervisor_trust.py +++ b/supervisor/resolution/checks/supervisor_trust.py @@ -1,6 +1,5 @@ """Helpers to check supervisor trust.""" import logging -from typing import Optional from ...const import CoreState from ...coresys import CoreSys @@ -35,7 +34,7 @@ class CheckSupervisorTrust(CheckBase): except CodeNotaryError: pass - async def approve_check(self, reference: Optional[str] = None) -> bool: + async def approve_check(self, reference: str | None = None) -> bool: """Approve check if it is affected by issue.""" try: await self.sys_supervisor.check_trust() diff --git a/supervisor/resolution/data.py b/supervisor/resolution/data.py index 84ef9b0d4..b1a5af156 100644 --- a/supervisor/resolution/data.py +++ b/supervisor/resolution/data.py @@ -1,5 +1,4 @@ """Data objects.""" -from typing import Optional from uuid import UUID, uuid4 import attr @@ -13,7 +12,7 @@ class Issue: type: IssueType = attr.ib() context: ContextType = attr.ib() - reference: Optional[str] = attr.ib(default=None) + reference: str | None = attr.ib(default=None) uuid: UUID = attr.ib(factory=lambda: uuid4().hex, eq=False, init=False) @@ -23,5 +22,5 @@ class Suggestion: type: SuggestionType = attr.ib() context: ContextType = attr.ib() - reference: Optional[str] = attr.ib(default=None) + reference: str | None = attr.ib(default=None) uuid: UUID = attr.ib(factory=lambda: uuid4().hex, eq=False, init=False) diff --git a/supervisor/resolution/fixups/base.py b/supervisor/resolution/fixups/base.py index 89ce8f55c..5d8b412a2 100644 --- a/supervisor/resolution/fixups/base.py +++ b/supervisor/resolution/fixups/base.py @@ -1,7 +1,6 @@ """Baseclass for system fixup.""" from abc import ABC, abstractmethod import logging -from typing import Optional from ...coresys import CoreSys, CoreSysAttributes from ...exceptions import ResolutionFixupError @@ -21,7 +20,7 @@ class FixupBase(ABC, CoreSysAttributes): async def __call__(self) -> None: """Execute the evaluation.""" # Get suggestion to fix - fixing_suggestion: Optional[Suggestion] = None + fixing_suggestion: Suggestion | None = None for suggestion in self.sys_resolution.suggestions: if suggestion.type != self.suggestion or suggestion.context != self.context: continue @@ -49,7 +48,7 @@ class FixupBase(ABC, CoreSysAttributes): self.sys_resolution.dismiss_issue(issue) @abstractmethod - async def process_fixup(self, reference: Optional[str] = None) -> None: + async def process_fixup(self, reference: str | None = None) -> None: """Run processing of fixup.""" @property diff --git a/supervisor/resolution/fixups/store_execute_reload.py b/supervisor/resolution/fixups/store_execute_reload.py index 3bb9c4330..a01b91335 100644 --- a/supervisor/resolution/fixups/store_execute_reload.py +++ b/supervisor/resolution/fixups/store_execute_reload.py @@ -1,6 +1,5 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import Optional from ...coresys import CoreSys from ...exceptions import ( @@ -29,7 +28,7 @@ class FixupStoreExecuteReload(FixupBase): conditions=[JobCondition.INTERNET_SYSTEM, JobCondition.FREE_SPACE], on_condition=ResolutionFixupJobError, ) - async def process_fixup(self, reference: Optional[str] = None) -> None: + async def process_fixup(self, reference: str | None = None) -> None: """Initialize the fixup class.""" _LOGGER.info("Reload Store: %s", reference) try: diff --git a/supervisor/resolution/fixups/store_execute_remove.py b/supervisor/resolution/fixups/store_execute_remove.py index 99336f1ce..b83fd3b13 100644 --- a/supervisor/resolution/fixups/store_execute_remove.py +++ b/supervisor/resolution/fixups/store_execute_remove.py @@ -1,6 +1,5 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import Optional from ...coresys import CoreSys from ...exceptions import ResolutionFixupError, StoreError, StoreNotFound @@ -18,7 +17,7 @@ def setup(coresys: CoreSys) -> FixupBase: class FixupStoreExecuteRemove(FixupBase): """Storage class for fixup.""" - async def process_fixup(self, reference: Optional[str] = None) -> None: + async def process_fixup(self, reference: str | None = None) -> None: """Initialize the fixup class.""" _LOGGER.info("Remove invalid Store: %s", reference) try: diff --git a/supervisor/resolution/fixups/store_execute_reset.py b/supervisor/resolution/fixups/store_execute_reset.py index 867d608be..6bc1a1ab3 100644 --- a/supervisor/resolution/fixups/store_execute_reset.py +++ b/supervisor/resolution/fixups/store_execute_reset.py @@ -1,6 +1,5 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import Optional from ...coresys import CoreSys from ...exceptions import ( @@ -30,7 +29,7 @@ class FixupStoreExecuteReset(FixupBase): conditions=[JobCondition.INTERNET_SYSTEM, JobCondition.FREE_SPACE], on_condition=ResolutionFixupJobError, ) - async def process_fixup(self, reference: Optional[str] = None) -> None: + async def process_fixup(self, reference: str | None = None) -> None: """Initialize the fixup class.""" _LOGGER.info("Reset corrupt Store: %s", reference) try: diff --git a/supervisor/resolution/fixups/system_clear_full_backup.py b/supervisor/resolution/fixups/system_clear_full_backup.py index 556056f30..782d6b50e 100644 --- a/supervisor/resolution/fixups/system_clear_full_backup.py +++ b/supervisor/resolution/fixups/system_clear_full_backup.py @@ -1,6 +1,5 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import Optional from ...backups.const import BackupType from ...coresys import CoreSys @@ -18,7 +17,7 @@ def setup(coresys: CoreSys) -> FixupBase: class FixupSystemClearFullBackup(FixupBase): """Storage class for fixup.""" - async def process_fixup(self, reference: Optional[str] = None) -> None: + async def process_fixup(self, reference: str | None = None) -> None: """Initialize the fixup class.""" full_backups = [ x for x in self.sys_backups.list_backups if x.sys_type == BackupType.FULL diff --git a/supervisor/resolution/fixups/system_create_full_backup.py b/supervisor/resolution/fixups/system_create_full_backup.py index 451d0a076..6dc12a46e 100644 --- a/supervisor/resolution/fixups/system_create_full_backup.py +++ b/supervisor/resolution/fixups/system_create_full_backup.py @@ -1,6 +1,5 @@ """Helpers to check and fix issues with free space.""" import logging -from typing import Optional from ...coresys import CoreSys from ..const import ContextType, SuggestionType @@ -17,7 +16,7 @@ def setup(coresys: CoreSys) -> FixupBase: class FixupSystemCreateFullBackup(FixupBase): """Storage class for fixup.""" - async def process_fixup(self, reference: Optional[str] = None) -> None: + async def process_fixup(self, reference: str | None = None) -> None: """Initialize the fixup class.""" _LOGGER.info("Creating a full backup") await self.sys_backups.do_backup_full() diff --git a/supervisor/resolution/fixups/system_execute_integrity.py b/supervisor/resolution/fixups/system_execute_integrity.py index 2cbb61b19..46c282c64 100644 --- a/supervisor/resolution/fixups/system_execute_integrity.py +++ b/supervisor/resolution/fixups/system_execute_integrity.py @@ -1,7 +1,6 @@ """Helpers to check and fix issues with free space.""" from datetime import timedelta import logging -from typing import Optional from ...coresys import CoreSys from ...exceptions import ResolutionFixupError, ResolutionFixupJobError @@ -28,7 +27,7 @@ class FixupSystemExecuteIntegrity(FixupBase): limit=JobExecutionLimit.THROTTLE, throttle_period=timedelta(hours=8), ) - async def process_fixup(self, reference: Optional[str] = None) -> None: + async def process_fixup(self, reference: str | None = None) -> None: """Initialize the fixup class.""" result = await self.sys_security.integrity_check() diff --git a/supervisor/resolution/module.py b/supervisor/resolution/module.py index a078786b7..d361675b9 100644 --- a/supervisor/resolution/module.py +++ b/supervisor/resolution/module.py @@ -1,6 +1,6 @@ """Supervisor resolution center.""" import logging -from typing import Any, Optional +from typing import Any from ..coresys import CoreSys, CoreSysAttributes from ..exceptions import ResolutionError, ResolutionNotFound @@ -142,8 +142,8 @@ class ResolutionManager(FileConfiguration, CoreSysAttributes): self, issue: IssueType, context: ContextType, - reference: Optional[str] = None, - suggestions: Optional[list[SuggestionType]] = None, + reference: str | None = None, + suggestions: list[SuggestionType] | None = None, ) -> None: """Create issues and suggestion.""" self.issues = Issue(issue, context, reference) diff --git a/supervisor/services/__init__.py b/supervisor/services/__init__.py index 7aa29153b..ea8e9d925 100644 --- a/supervisor/services/__init__.py +++ b/supervisor/services/__init__.py @@ -1,6 +1,4 @@ """Handle internal services discovery.""" -from typing import Optional - from ..coresys import CoreSys, CoreSysAttributes from .const import SERVICE_MQTT, SERVICE_MYSQL from .data import ServicesData @@ -25,7 +23,7 @@ class ServiceManager(CoreSysAttributes): """Return a list of services.""" return list(self.services_obj.values()) - def get(self, slug: str) -> Optional[ServiceInterface]: + def get(self, slug: str) -> ServiceInterface | None: """Return service object from slug.""" return self.services_obj.get(slug) diff --git a/supervisor/services/interface.py b/supervisor/services/interface.py index 4bfdd2598..a3c1dab40 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, Optional +from typing import Any import voluptuous as vol @@ -54,7 +54,7 @@ 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) -> dict[str, Any] | None: """Return the requested service data.""" if self.enabled: return self._data diff --git a/supervisor/store/data.py b/supervisor/store/data.py index 41a829fbd..2d58a16cc 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, Awaitable, Optional +from typing import Any, Awaitable import voluptuous as vol from voluptuous.humanize import humanize_error @@ -89,7 +89,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) -> list[Path] | None: """Find add-ons in the path.""" try: # Generate a list without artefact, safe for corruptions diff --git a/supervisor/store/git.py b/supervisor/store/git.py index d549ed327..e989944f1 100644 --- a/supervisor/store/git.py +++ b/supervisor/store/git.py @@ -3,7 +3,6 @@ import asyncio import functools as ft import logging from pathlib import Path -from typing import Optional import git @@ -27,7 +26,7 @@ class GitRepo(CoreSysAttributes): def __init__(self, coresys: CoreSys, path: Path, url: str): """Initialize Git base wrapper.""" self.coresys: CoreSys = coresys - self.repo: Optional[git.Repo] = None + self.repo: git.Repo | None = None self.path: Path = path self.lock: asyncio.Lock = asyncio.Lock() diff --git a/supervisor/store/repository.py b/supervisor/store/repository.py index 48ee352e8..441bfc6ff 100644 --- a/supervisor/store/repository.py +++ b/supervisor/store/repository.py @@ -1,7 +1,6 @@ """Represent a Supervisor repository.""" import logging from pathlib import Path -from typing import Optional import voluptuous as vol @@ -24,7 +23,7 @@ class Repository(CoreSysAttributes): def __init__(self, coresys: CoreSys, repository: str): """Initialize repository object.""" self.coresys: CoreSys = coresys - self.git: Optional[GitRepo] = None + self.git: GitRepo | None = None self.source: str = repository if repository == StoreType.LOCAL: diff --git a/supervisor/supervisor.py b/supervisor/supervisor.py index 0f082a27a..221d97f0c 100644 --- a/supervisor/supervisor.py +++ b/supervisor/supervisor.py @@ -5,7 +5,7 @@ from ipaddress import IPv4Address import logging from pathlib import Path from tempfile import TemporaryDirectory -from typing import Awaitable, Optional +from typing import Awaitable import aiohttp from aiohttp.client_exceptions import ClientError @@ -158,7 +158,7 @@ class Supervisor(CoreSysAttributes): "Can't update AppArmor profile!", _LOGGER.error ) from err - async def update(self, version: Optional[AwesomeVersion] = None) -> None: + async def update(self, version: AwesomeVersion | None = None) -> None: """Update Supervisor version.""" version = version or self.latest_version diff --git a/supervisor/updater.py b/supervisor/updater.py index 771adec8b..c4e8998dc 100644 --- a/supervisor/updater.py +++ b/supervisor/updater.py @@ -4,7 +4,6 @@ from contextlib import suppress from datetime import timedelta import json import logging -from typing import Optional import aiohttp from awesomeversion import AwesomeVersion @@ -60,47 +59,47 @@ class Updater(FileConfiguration, CoreSysAttributes): await self.fetch_data() @property - def version_homeassistant(self) -> Optional[AwesomeVersion]: + def version_homeassistant(self) -> AwesomeVersion | None: """Return latest version of Home Assistant.""" return self._data.get(ATTR_HOMEASSISTANT) @property - def version_supervisor(self) -> Optional[AwesomeVersion]: + def version_supervisor(self) -> AwesomeVersion | None: """Return latest version of Supervisor.""" return self._data.get(ATTR_SUPERVISOR) @property - def version_hassos(self) -> Optional[AwesomeVersion]: + def version_hassos(self) -> AwesomeVersion | None: """Return latest version of HassOS.""" return self._data.get(ATTR_HASSOS) @property - def version_cli(self) -> Optional[AwesomeVersion]: + def version_cli(self) -> AwesomeVersion | None: """Return latest version of CLI.""" return self._data.get(ATTR_CLI) @property - def version_dns(self) -> Optional[AwesomeVersion]: + def version_dns(self) -> AwesomeVersion | None: """Return latest version of DNS.""" return self._data.get(ATTR_DNS) @property - def version_audio(self) -> Optional[AwesomeVersion]: + def version_audio(self) -> AwesomeVersion | None: """Return latest version of Audio.""" return self._data.get(ATTR_AUDIO) @property - def version_observer(self) -> Optional[AwesomeVersion]: + def version_observer(self) -> AwesomeVersion | None: """Return latest version of Observer.""" return self._data.get(ATTR_OBSERVER) @property - def version_multicast(self) -> Optional[AwesomeVersion]: + def version_multicast(self) -> AwesomeVersion | None: """Return latest version of Multicast.""" return self._data.get(ATTR_MULTICAST) @property - def image_homeassistant(self) -> Optional[str]: + def image_homeassistant(self) -> str | None: """Return image of Home Assistant docker.""" if ATTR_HOMEASSISTANT not in self._data[ATTR_IMAGE]: return None @@ -109,7 +108,7 @@ class Updater(FileConfiguration, CoreSysAttributes): ) @property - def image_supervisor(self) -> Optional[str]: + def image_supervisor(self) -> str | None: """Return image of Supervisor docker.""" if ATTR_SUPERVISOR not in self._data[ATTR_IMAGE]: return None @@ -118,28 +117,28 @@ class Updater(FileConfiguration, CoreSysAttributes): ) @property - def image_cli(self) -> Optional[str]: + def image_cli(self) -> str | None: """Return image of CLI docker.""" if ATTR_CLI not in self._data[ATTR_IMAGE]: return None return self._data[ATTR_IMAGE][ATTR_CLI].format(arch=self.sys_arch.supervisor) @property - def image_dns(self) -> Optional[str]: + def image_dns(self) -> str | None: """Return image of DNS docker.""" if ATTR_DNS not in self._data[ATTR_IMAGE]: return None return self._data[ATTR_IMAGE][ATTR_DNS].format(arch=self.sys_arch.supervisor) @property - def image_audio(self) -> Optional[str]: + def image_audio(self) -> str | None: """Return image of Audio docker.""" if ATTR_AUDIO not in self._data[ATTR_IMAGE]: return None return self._data[ATTR_IMAGE][ATTR_AUDIO].format(arch=self.sys_arch.supervisor) @property - def image_observer(self) -> Optional[str]: + def image_observer(self) -> str | None: """Return image of Observer docker.""" if ATTR_OBSERVER not in self._data[ATTR_IMAGE]: return None @@ -148,7 +147,7 @@ class Updater(FileConfiguration, CoreSysAttributes): ) @property - def image_multicast(self) -> Optional[str]: + def image_multicast(self) -> str | None: """Return image of Multicast docker.""" if ATTR_MULTICAST not in self._data[ATTR_IMAGE]: return None @@ -157,7 +156,7 @@ class Updater(FileConfiguration, CoreSysAttributes): ) @property - def ota_url(self) -> Optional[str]: + def ota_url(self) -> str | None: """Return OTA url for OS.""" return self._data.get(ATTR_OTA) diff --git a/supervisor/utils/dt.py b/supervisor/utils/dt.py index 27cffe189..47887f00e 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, Optional +from typing import Any import zoneinfo import ciso8601 @@ -43,7 +43,7 @@ def parse_datetime(dt_str): kws["microsecond"] = kws["microsecond"].ljust(6, "0") tzinfo_str = kws.pop("tzinfo") - tzinfo_val: Optional[tzinfo] = None + tzinfo_val: tzinfo | None = None if tzinfo_str == "Z": tzinfo_val = UTC elif tzinfo_str is not None: @@ -70,7 +70,7 @@ def utc_from_timestamp(timestamp: float) -> datetime: return datetime.utcfromtimestamp(timestamp).replace(tzinfo=UTC) -def get_time_zone(time_zone_str: str) -> Optional[tzinfo]: +def get_time_zone(time_zone_str: str) -> tzinfo | None: """Get time zone from string. Return None if unable to determine.""" try: return zoneinfo.ZoneInfo(time_zone_str) diff --git a/supervisor/validate.py b/supervisor/validate.py index 75579ca09..329a34f3e 100644 --- a/supervisor/validate.py +++ b/supervisor/validate.py @@ -1,7 +1,6 @@ """Validate functions.""" import ipaddress import re -from typing import Optional, Union from awesomeversion import AwesomeVersion import voluptuous as vol @@ -57,8 +56,8 @@ token = vol.Match(r"^[0-9a-f]{32,256}$") def version_tag( - value: Union[str, None, int, float, AwesomeVersion] -) -> Optional[AwesomeVersion]: + value: str | None | int | float | AwesomeVersion, +) -> AwesomeVersion | None: """Validate main version handling.""" if value is None: return None diff --git a/tests/docker/test_monitor.py b/tests/docker/test_monitor.py index ed4dd8c8b..b2ef45786 100644 --- a/tests/docker/test_monitor.py +++ b/tests/docker/test_monitor.py @@ -1,7 +1,7 @@ """Test docker events monitor.""" import asyncio -from typing import Any, Optional +from typing import Any from unittest.mock import MagicMock, PropertyMock, patch from awesomeversion import AwesomeVersion @@ -84,7 +84,7 @@ from supervisor.docker.monitor import DockerContainerStateEvent ], ) async def test_events( - coresys: CoreSys, event: dict[str, Any], expected: Optional[ContainerState] + coresys: CoreSys, event: dict[str, Any], expected: ContainerState | None ): """Test events created from docker events.""" event["Actor"]["Attributes"]["name"] = "some_container" diff --git a/tests/jobs/test_job_decorator.py b/tests/jobs/test_job_decorator.py index 681a0aede..30e423dc0 100644 --- a/tests/jobs/test_job_decorator.py +++ b/tests/jobs/test_job_decorator.py @@ -2,7 +2,6 @@ # pylint: disable=protected-access,import-error import asyncio from datetime import timedelta -from typing import Optional from unittest.mock import PropertyMock, patch import pytest @@ -305,7 +304,7 @@ async def test_execution_limit_throttle_wait( @pytest.mark.parametrize("error", [None, PluginJobError]) async def test_execution_limit_throttle_rate_limit( - coresys: CoreSys, loop: asyncio.BaseEventLoop, error: Optional[JobException] + coresys: CoreSys, loop: asyncio.BaseEventLoop, error: JobException | None ): """Test the throttle wait job execution limit."""