diff --git a/homeassistant/components/androidtv/media_player.py b/homeassistant/components/androidtv/media_player.py index f4be6d6eea5..fb01ffce77f 100644 --- a/homeassistant/components/androidtv/media_player.py +++ b/homeassistant/components/androidtv/media_player.py @@ -140,12 +140,13 @@ async def async_setup_entry( ) +_FuncType = Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]] +_ReturnFuncType = Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]] + + def adb_decorator( override_available: bool = False, -) -> Callable[ - [Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]], - Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]], -]: +) -> Callable[[_FuncType[_ADBDeviceT, _P, _R]], _ReturnFuncType[_ADBDeviceT, _P, _R]]: """Wrap ADB methods and catch exceptions. Allows for overriding the available status of the ADB connection via the @@ -153,8 +154,8 @@ def adb_decorator( """ def _adb_decorator( - func: Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]] - ) -> Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]]: + func: _FuncType[_ADBDeviceT, _P, _R] + ) -> _ReturnFuncType[_ADBDeviceT, _P, _R]: """Wrap the provided ADB method and catch exceptions.""" @functools.wraps(func) diff --git a/homeassistant/components/hassio/addon_manager.py b/homeassistant/components/hassio/addon_manager.py index 88e755e3c7b..22265f49912 100644 --- a/homeassistant/components/hassio/addon_manager.py +++ b/homeassistant/components/hassio/addon_manager.py @@ -31,18 +31,20 @@ _AddonManagerT = TypeVar("_AddonManagerT", bound="AddonManager") _R = TypeVar("_R") _P = ParamSpec("_P") +_FuncType = Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]] +_ReturnFuncType = Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]] + def api_error( error_message: str, ) -> Callable[ - [Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]], - Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]], + [_FuncType[_AddonManagerT, _P, _R]], _ReturnFuncType[_AddonManagerT, _P, _R] ]: """Handle HassioAPIError and raise a specific AddonError.""" def handle_hassio_api_error( - func: Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]] - ) -> Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]]: + func: _FuncType[_AddonManagerT, _P, _R] + ) -> _ReturnFuncType[_AddonManagerT, _P, _R]: """Handle a HassioAPIError.""" @wraps(func) diff --git a/homeassistant/components/heos/media_player.py b/homeassistant/components/heos/media_player.py index 0c7a4bad42e..4184e9f82b7 100644 --- a/homeassistant/components/heos/media_player.py +++ b/homeassistant/components/heos/media_player.py @@ -88,14 +88,14 @@ async def async_setup_entry( async_add_entities(devices, True) -def log_command_error( - command: str, -) -> Callable[[Callable[_P, Awaitable[Any]]], Callable[_P, Coroutine[Any, Any, None]]]: +_FuncType = Callable[_P, Awaitable[Any]] +_ReturnFuncType = Callable[_P, Coroutine[Any, Any, None]] + + +def log_command_error(command: str) -> Callable[[_FuncType[_P]], _ReturnFuncType[_P]]: """Return decorator that logs command failure.""" - def decorator( - func: Callable[_P, Awaitable[Any]] - ) -> Callable[_P, Coroutine[Any, Any, None]]: + def decorator(func: _FuncType[_P]) -> _ReturnFuncType[_P]: @wraps(func) async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None: try: diff --git a/homeassistant/components/openhome/media_player.py b/homeassistant/components/openhome/media_player.py index 33f42ea949d..ef30d37bdcd 100644 --- a/homeassistant/components/openhome/media_player.py +++ b/homeassistant/components/openhome/media_player.py @@ -79,17 +79,22 @@ async def async_setup_platform( ) +_FuncType = Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]] +_ReturnFuncType = Callable[ + Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None] +] + + def catch_request_errors() -> ( Callable[ - [Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]]], - Callable[Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None]], + [_FuncType[_OpenhomeDeviceT, _P, _R]], _ReturnFuncType[_OpenhomeDeviceT, _P, _R] ] ): """Catch asyncio.TimeoutError, aiohttp.ClientError, UpnpError errors.""" def call_wrapper( - func: Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]] - ) -> Callable[Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None]]: + func: _FuncType[_OpenhomeDeviceT, _P, _R] + ) -> _ReturnFuncType[_OpenhomeDeviceT, _P, _R]: """Call wrapper for decorator.""" @functools.wraps(func) diff --git a/homeassistant/components/recorder/util.py b/homeassistant/components/recorder/util.py index 2fc773c5642..90c096e75e2 100644 --- a/homeassistant/components/recorder/util.py +++ b/homeassistant/components/recorder/util.py @@ -582,20 +582,18 @@ def end_incomplete_runs(session: Session, start_time: datetime) -> None: session.add(run) +_FuncType = Callable[Concatenate[_RecorderT, _P], bool] + + def retryable_database_job( description: str, -) -> Callable[ - [Callable[Concatenate[_RecorderT, _P], bool]], - Callable[Concatenate[_RecorderT, _P], bool], -]: +) -> Callable[[_FuncType[_RecorderT, _P]], _FuncType[_RecorderT, _P]]: """Try to execute a database job. The job should return True if it finished, and False if it needs to be rescheduled. """ - def decorator( - job: Callable[Concatenate[_RecorderT, _P], bool] - ) -> Callable[Concatenate[_RecorderT, _P], bool]: + def decorator(job: _FuncType[_RecorderT, _P]) -> _FuncType[_RecorderT, _P]: @functools.wraps(job) def wrapper(instance: _RecorderT, *args: _P.args, **kwargs: _P.kwargs) -> bool: try: diff --git a/homeassistant/components/roku/helpers.py b/homeassistant/components/roku/helpers.py index 22e9ee19b16..60392d89f1d 100644 --- a/homeassistant/components/roku/helpers.py +++ b/homeassistant/components/roku/helpers.py @@ -14,6 +14,9 @@ from .entity import RokuEntity _RokuEntityT = TypeVar("_RokuEntityT", bound=RokuEntity) _P = ParamSpec("_P") +_FuncType = Callable[Concatenate[_RokuEntityT, _P], Awaitable[Any]] +_ReturnFuncType = Callable[Concatenate[_RokuEntityT, _P], Coroutine[Any, Any, None]] + def format_channel_name(channel_number: str, channel_name: str | None = None) -> str: """Format a Roku Channel name.""" @@ -25,15 +28,12 @@ def format_channel_name(channel_number: str, channel_name: str | None = None) -> def roku_exception_handler( ignore_timeout: bool = False, -) -> Callable[ - [Callable[Concatenate[_RokuEntityT, _P], Awaitable[Any]]], - Callable[Concatenate[_RokuEntityT, _P], Coroutine[Any, Any, None]], -]: +) -> Callable[[_FuncType[_RokuEntityT, _P]], _ReturnFuncType[_RokuEntityT, _P]]: """Decorate Roku calls to handle Roku exceptions.""" def decorator( - func: Callable[Concatenate[_RokuEntityT, _P], Awaitable[Any]], - ) -> Callable[Concatenate[_RokuEntityT, _P], Coroutine[Any, Any, None]]: + func: _FuncType[_RokuEntityT, _P] + ) -> _ReturnFuncType[_RokuEntityT, _P]: @wraps(func) async def wrapper( self: _RokuEntityT, *args: _P.args, **kwargs: _P.kwargs diff --git a/homeassistant/components/sonos/helpers.py b/homeassistant/components/sonos/helpers.py index f44775e2f31..5f44b9bae6f 100644 --- a/homeassistant/components/sonos/helpers.py +++ b/homeassistant/components/sonos/helpers.py @@ -31,33 +31,30 @@ _T = TypeVar( _R = TypeVar("_R") _P = ParamSpec("_P") +_FuncType = Callable[Concatenate[_T, _P], _R] +_ReturnFuncType = Callable[Concatenate[_T, _P], _R | None] + @overload def soco_error( errorcodes: None = ..., -) -> Callable[[Callable[Concatenate[_T, _P], _R]], Callable[Concatenate[_T, _P], _R]]: +) -> Callable[[_FuncType[_T, _P, _R]], _FuncType[_T, _P, _R]]: ... @overload def soco_error( errorcodes: list[str], -) -> Callable[ - [Callable[Concatenate[_T, _P], _R]], Callable[Concatenate[_T, _P], _R | None] -]: +) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]: ... def soco_error( errorcodes: list[str] | None = None, -) -> Callable[ - [Callable[Concatenate[_T, _P], _R]], Callable[Concatenate[_T, _P], _R | None] -]: +) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]: """Filter out specified UPnP errors and raise exceptions for service calls.""" - def decorator( - funct: Callable[Concatenate[_T, _P], _R] - ) -> Callable[Concatenate[_T, _P], _R | None]: + def decorator(funct: _FuncType[_T, _P, _R]) -> _ReturnFuncType[_T, _P, _R]: """Decorate functions.""" def wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _R | None: