From 25d1ca747b81c02eb546c40551f7acb64f222f0d Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Fri, 17 May 2024 16:27:32 +0200 Subject: [PATCH] Use PEP 695 for decorator typing (3) (#117640) --- .../components/synology_dsm/coordinator.py | 14 +++++--------- homeassistant/components/technove/helpers.py | 7 ++----- homeassistant/components/toon/helpers.py | 7 ++----- homeassistant/components/tplink/entity.py | 7 ++----- homeassistant/components/velbus/entity.py | 8 ++------ .../components/vlc_telnet/media_player.py | 7 ++----- homeassistant/components/wallbox/coordinator.py | 7 ++----- homeassistant/components/webostv/media_player.py | 8 ++------ homeassistant/components/wled/helpers.py | 7 ++----- homeassistant/components/yeelight/light.py | 8 ++------ homeassistant/components/zwave_js/api.py | 6 ++---- homeassistant/helpers/event.py | 5 ++--- homeassistant/util/loop.py | 8 ++------ tests/conftest.py | 8 ++------ 14 files changed, 31 insertions(+), 76 deletions(-) diff --git a/homeassistant/components/synology_dsm/coordinator.py b/homeassistant/components/synology_dsm/coordinator.py index 52a3e1de1eb..bce59d2546e 100644 --- a/homeassistant/components/synology_dsm/coordinator.py +++ b/homeassistant/components/synology_dsm/coordinator.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Coroutine from datetime import timedelta import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate, TypeVar from synology_dsm.api.surveillance_station.camera import SynoCamera from synology_dsm.exceptions import ( @@ -31,16 +31,12 @@ _LOGGER = logging.getLogger(__name__) _DataT = TypeVar("_DataT") -_T = TypeVar("_T", bound="SynologyDSMUpdateCoordinator") -_P = ParamSpec("_P") - - -def async_re_login_on_expired( - func: Callable[Concatenate[_T, _P], Awaitable[_DataT]], -) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, _DataT]]: +def async_re_login_on_expired[_T: SynologyDSMUpdateCoordinator[Any], **_P, _R]( + func: Callable[Concatenate[_T, _P], Awaitable[_R]], +) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, _R]]: """Define a wrapper to re-login when expired.""" - async def _async_wrap(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _DataT: + async def _async_wrap(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _R: for attempts in range(2): try: return await func(self, *args, **kwargs) diff --git a/homeassistant/components/technove/helpers.py b/homeassistant/components/technove/helpers.py index 4d8bda38a25..a4aebf5f1fe 100644 --- a/homeassistant/components/technove/helpers.py +++ b/homeassistant/components/technove/helpers.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from technove import TechnoVEConnectionError, TechnoVEError @@ -11,11 +11,8 @@ from homeassistant.exceptions import HomeAssistantError from .entity import TechnoVEEntity -_TechnoVEEntityT = TypeVar("_TechnoVEEntityT", bound=TechnoVEEntity) -_P = ParamSpec("_P") - -def technove_exception_handler( +def technove_exception_handler[_TechnoVEEntityT: TechnoVEEntity, **_P]( func: Callable[Concatenate[_TechnoVEEntityT, _P], Coroutine[Any, Any, Any]], ) -> Callable[Concatenate[_TechnoVEEntityT, _P], Coroutine[Any, Any, None]]: """Decorate TechnoVE calls to handle TechnoVE exceptions. diff --git a/homeassistant/components/toon/helpers.py b/homeassistant/components/toon/helpers.py index cd4e55fd050..0dd740544df 100644 --- a/homeassistant/components/toon/helpers.py +++ b/homeassistant/components/toon/helpers.py @@ -4,19 +4,16 @@ from __future__ import annotations from collections.abc import Callable, Coroutine import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from toonapi import ToonConnectionError, ToonError from .models import ToonEntity -_ToonEntityT = TypeVar("_ToonEntityT", bound=ToonEntity) -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) -def toon_exception_handler( +def toon_exception_handler[_ToonEntityT: ToonEntity, **_P]( func: Callable[Concatenate[_ToonEntityT, _P], Coroutine[Any, Any, None]], ) -> Callable[Concatenate[_ToonEntityT, _P], Coroutine[Any, Any, None]]: """Decorate Toon calls to handle Toon exceptions. diff --git a/homeassistant/components/tplink/entity.py b/homeassistant/components/tplink/entity.py index 23766e69257..52b226a1c57 100644 --- a/homeassistant/components/tplink/entity.py +++ b/homeassistant/components/tplink/entity.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Coroutine -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from kasa import ( AuthenticationException, @@ -20,11 +20,8 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN from .coordinator import TPLinkDataUpdateCoordinator -_T = TypeVar("_T", bound="CoordinatedTPLinkEntity") -_P = ParamSpec("_P") - -def async_refresh_after( +def async_refresh_after[_T: CoordinatedTPLinkEntity, **_P]( func: Callable[Concatenate[_T, _P], Awaitable[None]], ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]: """Define a wrapper to raise HA errors and refresh after.""" diff --git a/homeassistant/components/velbus/entity.py b/homeassistant/components/velbus/entity.py index 202666e6123..65f8a1d8d31 100644 --- a/homeassistant/components/velbus/entity.py +++ b/homeassistant/components/velbus/entity.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Coroutine from functools import wraps -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from velbusaio.channels import Channel as VelbusChannel @@ -44,11 +44,7 @@ class VelbusEntity(Entity): self.async_write_ha_state() -_T = TypeVar("_T", bound="VelbusEntity") -_P = ParamSpec("_P") - - -def api_call( +def api_call[_T: VelbusEntity, **_P]( func: Callable[Concatenate[_T, _P], Awaitable[None]], ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]: """Catch command exceptions.""" diff --git a/homeassistant/components/vlc_telnet/media_player.py b/homeassistant/components/vlc_telnet/media_player.py index 6245f0e45e6..42bf42de97e 100644 --- a/homeassistant/components/vlc_telnet/media_player.py +++ b/homeassistant/components/vlc_telnet/media_player.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections.abc import Awaitable, Callable, Coroutine from functools import wraps -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from aiovlc.client import Client from aiovlc.exceptions import AuthError, CommandError, ConnectError @@ -30,9 +30,6 @@ from .const import DEFAULT_NAME, DOMAIN, LOGGER MAX_VOLUME = 500 -_VlcDeviceT = TypeVar("_VlcDeviceT", bound="VlcDevice") -_P = ParamSpec("_P") - async def async_setup_entry( hass: HomeAssistant, entry: VlcConfigEntry, async_add_entities: AddEntitiesCallback @@ -46,7 +43,7 @@ async def async_setup_entry( async_add_entities([VlcDevice(entry, vlc, name, available)], True) -def catch_vlc_errors( +def catch_vlc_errors[_VlcDeviceT: VlcDevice, **_P]( func: Callable[Concatenate[_VlcDeviceT, _P], Awaitable[None]], ) -> Callable[Concatenate[_VlcDeviceT, _P], Coroutine[Any, Any, None]]: """Catch VLC errors.""" diff --git a/homeassistant/components/wallbox/coordinator.py b/homeassistant/components/wallbox/coordinator.py index bf7c6d1f654..e24ccd28440 100644 --- a/homeassistant/components/wallbox/coordinator.py +++ b/homeassistant/components/wallbox/coordinator.py @@ -6,7 +6,7 @@ from collections.abc import Callable from datetime import timedelta from http import HTTPStatus import logging -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate import requests from wallbox import Wallbox @@ -64,11 +64,8 @@ CHARGER_STATUS: dict[int, ChargerStatus] = { 210: ChargerStatus.LOCKED_CAR_CONNECTED, } -_WallboxCoordinatorT = TypeVar("_WallboxCoordinatorT", bound="WallboxCoordinator") -_P = ParamSpec("_P") - -def _require_authentication( +def _require_authentication[_WallboxCoordinatorT: WallboxCoordinator, **_P]( func: Callable[Concatenate[_WallboxCoordinatorT, _P], Any], ) -> Callable[Concatenate[_WallboxCoordinatorT, _P], Any]: """Authenticate with decorator using Wallbox API.""" diff --git a/homeassistant/components/webostv/media_player.py b/homeassistant/components/webostv/media_player.py index 34ff8aafca2..6aef47515db 100644 --- a/homeassistant/components/webostv/media_player.py +++ b/homeassistant/components/webostv/media_player.py @@ -9,7 +9,7 @@ from datetime import timedelta from functools import wraps from http import HTTPStatus import logging -from typing import Any, Concatenate, ParamSpec, TypeVar, cast +from typing import Any, Concatenate, cast from aiowebostv import WebOsClient, WebOsTvPairError @@ -79,11 +79,7 @@ async def async_setup_entry( async_add_entities([LgWebOSMediaPlayerEntity(entry, client)]) -_T = TypeVar("_T", bound="LgWebOSMediaPlayerEntity") -_P = ParamSpec("_P") - - -def cmd( +def cmd[_T: LgWebOSMediaPlayerEntity, **_P]( func: Callable[Concatenate[_T, _P], Awaitable[None]], ) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]: """Catch command exceptions.""" diff --git a/homeassistant/components/wled/helpers.py b/homeassistant/components/wled/helpers.py index 1358a3c05f1..0dd29fdc2a3 100644 --- a/homeassistant/components/wled/helpers.py +++ b/homeassistant/components/wled/helpers.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate from wled import WLEDConnectionError, WLEDError @@ -11,11 +11,8 @@ from homeassistant.exceptions import HomeAssistantError from .entity import WLEDEntity -_WLEDEntityT = TypeVar("_WLEDEntityT", bound=WLEDEntity) -_P = ParamSpec("_P") - -def wled_exception_handler( +def wled_exception_handler[_WLEDEntityT: WLEDEntity, **_P]( func: Callable[Concatenate[_WLEDEntityT, _P], Coroutine[Any, Any, Any]], ) -> Callable[Concatenate[_WLEDEntityT, _P], Coroutine[Any, Any, None]]: """Decorate WLED calls to handle WLED exceptions. diff --git a/homeassistant/components/yeelight/light.py b/homeassistant/components/yeelight/light.py index ede652dd037..1d514c131d2 100644 --- a/homeassistant/components/yeelight/light.py +++ b/homeassistant/components/yeelight/light.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine import logging import math -from typing import Any, Concatenate, ParamSpec, TypeVar +from typing import Any, Concatenate import voluptuous as vol import yeelight @@ -67,10 +67,6 @@ from .const import ( from .device import YeelightDevice from .entity import YeelightEntity -_YeelightBaseLightT = TypeVar("_YeelightBaseLightT", bound="YeelightBaseLight") -_R = TypeVar("_R") -_P = ParamSpec("_P") - _LOGGER = logging.getLogger(__name__) ATTR_MINUTES = "minutes" @@ -243,7 +239,7 @@ def _parse_custom_effects(effects_config) -> dict[str, dict[str, Any]]: return effects -def _async_cmd( +def _async_cmd[_YeelightBaseLightT: YeelightBaseLight, **_P, _R]( func: Callable[Concatenate[_YeelightBaseLightT, _P], Coroutine[Any, Any, _R]], ) -> Callable[Concatenate[_YeelightBaseLightT, _P], Coroutine[Any, Any, _R | None]]: """Define a wrapper to catch exceptions from the bulb.""" diff --git a/homeassistant/components/zwave_js/api.py b/homeassistant/components/zwave_js/api.py index ca03cd643c9..997a9b6dad0 100644 --- a/homeassistant/components/zwave_js/api.py +++ b/homeassistant/components/zwave_js/api.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Callable, Coroutine import dataclasses from functools import partial, wraps -from typing import Any, Concatenate, Literal, ParamSpec, cast +from typing import Any, Concatenate, Literal, cast from aiohttp import web, web_exceptions, web_request import voluptuous as vol @@ -84,8 +84,6 @@ from .helpers import ( get_device_id, ) -_P = ParamSpec("_P") - DATA_UNSUBSCRIBE = "unsubs" # general API constants @@ -362,7 +360,7 @@ def async_get_node( return async_get_node_func -def async_handle_failed_command( +def async_handle_failed_command[**_P]( orig_func: Callable[ Concatenate[HomeAssistant, ActiveConnection, dict[str, Any], _P], Coroutine[Any, Any, None], diff --git a/homeassistant/helpers/event.py b/homeassistant/helpers/event.py index c54af93d320..9739f8fbaa6 100644 --- a/homeassistant/helpers/event.py +++ b/homeassistant/helpers/event.py @@ -12,7 +12,7 @@ from functools import partial, wraps import logging from random import randint import time -from typing import TYPE_CHECKING, Any, Concatenate, Generic, ParamSpec, TypeVar +from typing import TYPE_CHECKING, Any, Concatenate, Generic, TypeVar from homeassistant.const import ( EVENT_CORE_CONFIG_UPDATE, @@ -93,7 +93,6 @@ RANDOM_MICROSECOND_MIN = 50000 RANDOM_MICROSECOND_MAX = 500000 _TypedDictT = TypeVar("_TypedDictT", bound=Mapping[str, Any]) -_P = ParamSpec("_P") @dataclass(slots=True, frozen=True) @@ -168,7 +167,7 @@ class TrackTemplateResult: result: Any -def threaded_listener_factory( +def threaded_listener_factory[**_P]( async_factory: Callable[Concatenate[HomeAssistant, _P], Any], ) -> Callable[Concatenate[HomeAssistant, _P], CALLBACK_TYPE]: """Convert an async event helper to a threaded one.""" diff --git a/homeassistant/util/loop.py b/homeassistant/util/loop.py index 071eb42149b..accb63198ba 100644 --- a/homeassistant/util/loop.py +++ b/homeassistant/util/loop.py @@ -8,7 +8,7 @@ import functools import linecache import logging import threading -from typing import Any, ParamSpec, TypeVar +from typing import Any from homeassistant.core import HomeAssistant, async_get_hass from homeassistant.exceptions import HomeAssistantError @@ -22,10 +22,6 @@ from homeassistant.loader import async_suggest_report_issue _LOGGER = logging.getLogger(__name__) -_R = TypeVar("_R") -_P = ParamSpec("_P") - - def _get_line_from_cache(filename: str, lineno: int) -> str: """Get line from cache or read from file.""" return (linecache.getline(filename, lineno) or "?").strip() @@ -114,7 +110,7 @@ def raise_for_blocking_call( ) -def protect_loop( +def protect_loop[**_P, _R]( func: Callable[_P, _R], loop_thread_id: int, strict: bool = True, diff --git a/tests/conftest.py b/tests/conftest.py index 3d4d55e696c..4de97bd5094 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -14,7 +14,7 @@ import reprlib import sqlite3 import ssl import threading -from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar, cast +from typing import TYPE_CHECKING, Any, cast from unittest.mock import AsyncMock, MagicMock, Mock, patch from aiohttp import client @@ -204,11 +204,7 @@ class HAFakeDatetime(freezegun.api.FakeDatetime): # type: ignore[name-defined] return ha_datetime_to_fakedatetime(result) -_R = TypeVar("_R") -_P = ParamSpec("_P") - - -def check_real(func: Callable[_P, Coroutine[Any, Any, _R]]): +def check_real[**_P, _R](func: Callable[_P, Coroutine[Any, Any, _R]]): """Force a function to require a keyword _test_real to be passed in.""" @functools.wraps(func)