diff --git a/homeassistant/components/dlna_dmr/media_player.py b/homeassistant/components/dlna_dmr/media_player.py index 1b64f592460..fa20c4f6708 100644 --- a/homeassistant/components/dlna_dmr/media_player.py +++ b/homeassistant/components/dlna_dmr/media_player.py @@ -2,17 +2,18 @@ from __future__ import annotations import asyncio -from collections.abc import Sequence +from collections.abc import Awaitable, Callable, Sequence import contextlib from datetime import datetime, timedelta import functools -from typing import TYPE_CHECKING, Any, Callable, TypeVar, cast +from typing import TYPE_CHECKING, Any, TypeVar from async_upnp_client import UpnpService, UpnpStateVariable from async_upnp_client.const import NotificationSubType from async_upnp_client.exceptions import UpnpError, UpnpResponseError from async_upnp_client.profiles.dlna import DmrDevice, PlayMode, TransportState from async_upnp_client.utils import async_get_local_ip +from typing_extensions import Concatenate, ParamSpec from homeassistant import config_entries from homeassistant.components import ssdp @@ -65,27 +66,32 @@ from .data import EventListenAddr, get_domain_data PARALLEL_UPDATES = 0 -Func = TypeVar("Func", bound=Callable[..., Any]) +_T = TypeVar("_T", bound="DlnaDmrEntity") +_R = TypeVar("_R") +_P = ParamSpec("_P") -def catch_request_errors(func: Func) -> Func: +def catch_request_errors( + func: Callable[Concatenate[_T, _P], Awaitable[_R]] # type: ignore[misc] +) -> Callable[Concatenate[_T, _P], Awaitable[_R | None]]: # type: ignore[misc] """Catch UpnpError errors.""" @functools.wraps(func) - async def wrapper(self: "DlnaDmrEntity", *args: Any, **kwargs: Any) -> Any: + async def wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _R | None: """Catch UpnpError errors and check availability before and after request.""" if not self.available: _LOGGER.warning( "Device disappeared when trying to call service %s", func.__name__ ) - return + return None try: - return await func(self, *args, **kwargs) + return await func(self, *args, **kwargs) # type: ignore[no-any-return] # mypy can't yet infer 'func' except UpnpError as err: self.check_available = True _LOGGER.error("Error during call %s: %r", func.__name__, err) + return None - return cast(Func, wrapper) + return wrapper async def async_setup_entry( diff --git a/homeassistant/components/evil_genius_labs/util.py b/homeassistant/components/evil_genius_labs/util.py index 42088f69797..dd9ce97326e 100644 --- a/homeassistant/components/evil_genius_labs/util.py +++ b/homeassistant/components/evil_genius_labs/util.py @@ -1,21 +1,29 @@ """Utilities for Evil Genius Labs.""" -from collections.abc import Callable +from __future__ import annotations + +from collections.abc import Awaitable, Callable from functools import wraps -from typing import Any, TypeVar, cast +from typing import TypeVar + +from typing_extensions import Concatenate, ParamSpec from . import EvilGeniusEntity -CallableT = TypeVar("CallableT", bound=Callable) +_T = TypeVar("_T", bound=EvilGeniusEntity) +_R = TypeVar("_R") +_P = ParamSpec("_P") -def update_when_done(func: CallableT) -> CallableT: +def update_when_done( + func: Callable[Concatenate[_T, _P], Awaitable[_R]] # type: ignore[misc] +) -> Callable[Concatenate[_T, _P], Awaitable[_R]]: # type: ignore[misc] """Decorate function to trigger update when function is done.""" @wraps(func) - async def wrapper(self: EvilGeniusEntity, *args: Any, **kwargs: Any) -> Any: + async def wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _R: """Wrap function.""" result = await func(self, *args, **kwargs) await self.coordinator.async_request_refresh() - return result + return result # type: ignore[no-any-return] # mypy can't yet infer 'func' - return cast(CallableT, wrapper) + return wrapper diff --git a/homeassistant/components/sonos/helpers.py b/homeassistant/components/sonos/helpers.py index 74897a618ea..625b54b941e 100644 --- a/homeassistant/components/sonos/helpers.py +++ b/homeassistant/components/sonos/helpers.py @@ -1,10 +1,12 @@ """Helper methods for common tasks.""" from __future__ import annotations +from collections.abc import Callable import logging -from typing import TYPE_CHECKING, Any, Callable, TypeVar, cast +from typing import TYPE_CHECKING, TypeVar from soco.exceptions import SoCoException, SoCoUPnPException +from typing_extensions import Concatenate, ParamSpec from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.dispatcher import dispatcher_send @@ -19,20 +21,26 @@ if TYPE_CHECKING: UID_PREFIX = "RINCON_" UID_POSTFIX = "01400" -WrapFuncType = TypeVar("WrapFuncType", bound=Callable[..., Any]) - _LOGGER = logging.getLogger(__name__) +_T = TypeVar("_T", "SonosSpeaker", "SonosEntity") +_R = TypeVar("_R") +_P = ParamSpec("_P") + def soco_error( errorcodes: list[str] | None = None, raise_on_err: bool = True -) -> Callable: +) -> Callable[ # type: ignore[misc] + [Callable[Concatenate[_T, _P], _R]], Callable[Concatenate[_T, _P], _R | None] +]: """Filter out specified UPnP errors and raise exceptions for service calls.""" - def decorator(funct: WrapFuncType) -> WrapFuncType: + def decorator( + funct: Callable[Concatenate[_T, _P], _R] # type: ignore[misc] + ) -> Callable[Concatenate[_T, _P], _R | None]: # type: ignore[misc] """Decorate functions.""" - def wrapper(self: SonosSpeaker | SonosEntity, *args: Any, **kwargs: Any) -> Any: + def wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _R | None: """Wrap for all soco UPnP exception.""" try: result = funct(self, *args, **kwargs) @@ -65,7 +73,7 @@ def soco_error( ) return result - return cast(WrapFuncType, wrapper) + return wrapper return decorator diff --git a/homeassistant/components/tplink/entity.py b/homeassistant/components/tplink/entity.py index 3038e93ac25..0bfecaafa3d 100644 --- a/homeassistant/components/tplink/entity.py +++ b/homeassistant/components/tplink/entity.py @@ -1,9 +1,11 @@ """Common code for tplink.""" from __future__ import annotations -from typing import Any, Callable, TypeVar, cast +from collections.abc import Awaitable, Callable +from typing import TypeVar from kasa import SmartDevice +from typing_extensions import Concatenate, ParamSpec from homeassistant.helpers import device_registry as dr from homeassistant.helpers.entity import DeviceInfo @@ -12,19 +14,20 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN from .coordinator import TPLinkDataUpdateCoordinator -WrapFuncType = TypeVar("WrapFuncType", bound=Callable[..., Any]) +_T = TypeVar("_T", bound="CoordinatedTPLinkEntity") +_P = ParamSpec("_P") -def async_refresh_after(func: WrapFuncType) -> WrapFuncType: +def async_refresh_after( + func: Callable[Concatenate[_T, _P], Awaitable[None]] # type: ignore[misc] +) -> Callable[Concatenate[_T, _P], Awaitable[None]]: # type: ignore[misc] """Define a wrapper to refresh after.""" - async def _async_wrap( - self: CoordinatedTPLinkEntity, *args: Any, **kwargs: Any - ) -> None: + async def _async_wrap(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> None: await func(self, *args, **kwargs) await self.coordinator.async_request_refresh_without_children() - return cast(WrapFuncType, _async_wrap) + return _async_wrap class CoordinatedTPLinkEntity(CoordinatorEntity): diff --git a/homeassistant/components/vlc_telnet/media_player.py b/homeassistant/components/vlc_telnet/media_player.py index 1aa612ade25..6b53a2c3470 100644 --- a/homeassistant/components/vlc_telnet/media_player.py +++ b/homeassistant/components/vlc_telnet/media_player.py @@ -1,12 +1,14 @@ """Provide functionality to interact with the vlc telnet interface.""" from __future__ import annotations +from collections.abc import Awaitable, Callable from datetime import datetime from functools import wraps -from typing import Any, Callable, TypeVar, cast +from typing import Any, TypeVar from aiovlc.client import Client from aiovlc.exceptions import AuthError, CommandError, ConnectError +from typing_extensions import Concatenate, ParamSpec from homeassistant.components.media_player import MediaPlayerEntity from homeassistant.components.media_player.const import ( @@ -49,7 +51,9 @@ SUPPORT_VLC = ( | SUPPORT_VOLUME_SET ) -Func = TypeVar("Func", bound=Callable[..., Any]) +_T = TypeVar("_T", bound="VlcDevice") +_R = TypeVar("_R") +_P = ParamSpec("_P") async def async_setup_entry( @@ -64,11 +68,13 @@ async def async_setup_entry( async_add_entities([VlcDevice(entry, vlc, name, available)], True) -def catch_vlc_errors(func: Func) -> Func: +def catch_vlc_errors( + func: Callable[Concatenate[_T, _P], Awaitable[None]] # type: ignore[misc] +) -> Callable[Concatenate[_T, _P], Awaitable[None]]: # type: ignore[misc] """Catch VLC errors.""" @wraps(func) - async def wrapper(self: VlcDevice, *args: Any, **kwargs: Any) -> Any: + async def wrapper(self: VlcDevice, *args: _P.args, **kwargs: _P.kwargs) -> None: """Catch VLC errors and modify availability.""" try: await func(self, *args, **kwargs) @@ -80,7 +86,7 @@ def catch_vlc_errors(func: Func) -> Func: LOGGER.error("Connection error: %s", err) self._available = False - return cast(Func, wrapper) + return wrapper class VlcDevice(MediaPlayerEntity): diff --git a/homeassistant/components/zwave_js/addon.py b/homeassistant/components/zwave_js/addon.py index 38fdee9a051..1ea71b260a6 100644 --- a/homeassistant/components/zwave_js/addon.py +++ b/homeassistant/components/zwave_js/addon.py @@ -2,10 +2,13 @@ from __future__ import annotations import asyncio +from collections.abc import Awaitable, Callable from dataclasses import dataclass from enum import Enum from functools import partial -from typing import Any, Callable, TypeVar, cast +from typing import Any, TypeVar + +from typing_extensions import ParamSpec from homeassistant.components.hassio import ( async_create_backup, @@ -35,7 +38,8 @@ from .const import ( LOGGER, ) -F = TypeVar("F", bound=Callable[..., Any]) # pylint: disable=invalid-name +_R = TypeVar("_R") +_P = ParamSpec("_P") DATA_ADDON_MANAGER = f"{DOMAIN}_addon_manager" @@ -47,13 +51,17 @@ def get_addon_manager(hass: HomeAssistant) -> AddonManager: return AddonManager(hass) -def api_error(error_message: str) -> Callable[[F], F]: +def api_error( + error_message: str, +) -> Callable[[Callable[_P, Awaitable[_R]]], Callable[_P, Awaitable[_R]]]: """Handle HassioAPIError and raise a specific AddonError.""" - def handle_hassio_api_error(func: F) -> F: + def handle_hassio_api_error( + func: Callable[_P, Awaitable[_R]] + ) -> Callable[_P, Awaitable[_R]]: """Handle a HassioAPIError.""" - async def wrapper(*args, **kwargs): # type: ignore + async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> _R: """Wrap an add-on manager method.""" try: return_value = await func(*args, **kwargs) @@ -62,7 +70,7 @@ def api_error(error_message: str) -> Callable[[F], F]: return return_value - return cast(F, wrapper) + return wrapper return handle_hassio_api_error diff --git a/homeassistant/package_constraints.txt b/homeassistant/package_constraints.txt index 034bac2e7ca..ae9efb17c1c 100644 --- a/homeassistant/package_constraints.txt +++ b/homeassistant/package_constraints.txt @@ -29,6 +29,7 @@ pyyaml==6.0 requests==2.26.0 scapy==2.4.5 sqlalchemy==1.4.27 +typing-extensions>=3.10.0.2,<5.0 voluptuous-serialize==2.5.0 voluptuous==0.12.2 yarl==1.6.3 diff --git a/requirements.txt b/requirements.txt index 4c6af849ce8..09f702a13fe 100644 --- a/requirements.txt +++ b/requirements.txt @@ -20,6 +20,7 @@ pip>=8.0.3,<20.3 python-slugify==4.0.1 pyyaml==6.0 requests==2.26.0 +typing-extensions>=3.10.0.2,<5.0 voluptuous==0.12.2 voluptuous-serialize==2.5.0 yarl==1.6.3 diff --git a/setup.py b/setup.py index 270f5c58f58..407edb5ef7e 100755 --- a/setup.py +++ b/setup.py @@ -52,6 +52,7 @@ REQUIRES = [ "python-slugify==4.0.1", "pyyaml==6.0", "requests==2.26.0", + "typing-extensions>=3.10.0.2,<5.0", "voluptuous==0.12.2", "voluptuous-serialize==2.5.0", "yarl==1.6.3",