From 907b9c42e5aaee6d1cd35e45d9674b94a9ed2ef2 Mon Sep 17 00:00:00 2001
From: Marc Mueller <30130371+cdce8p@users.noreply.github.com>
Date: Sat, 18 May 2024 11:41:46 +0200
Subject: [PATCH] Use PEP 695 for decorator typing with type aliases (2)
 (#117663)

---
 .../components/openhome/media_player.py       | 14 ++++------
 homeassistant/components/recorder/util.py     | 26 ++++++++-----------
 homeassistant/components/roku/helpers.py      | 13 +++++-----
 homeassistant/components/sonos/helpers.py     | 19 ++++++--------
 .../zha/core/cluster_handlers/__init__.py     | 10 +++----
 homeassistant/helpers/singleton.py            | 14 +++++-----
 6 files changed, 41 insertions(+), 55 deletions(-)

diff --git a/homeassistant/components/openhome/media_player.py b/homeassistant/components/openhome/media_player.py
index 12e5ed992c2..c9143c977ce 100644
--- a/homeassistant/components/openhome/media_player.py
+++ b/homeassistant/components/openhome/media_player.py
@@ -5,7 +5,7 @@ from __future__ import annotations
 from collections.abc import Awaitable, Callable, Coroutine
 import functools
 import logging
-from typing import Any, Concatenate, ParamSpec, TypeVar
+from typing import Any, Concatenate
 
 import aiohttp
 from async_upnp_client.client import UpnpError
@@ -28,10 +28,6 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from .const import ATTR_PIN_INDEX, DOMAIN, SERVICE_INVOKE_PIN
 
-_OpenhomeDeviceT = TypeVar("_OpenhomeDeviceT", bound="OpenhomeDevice")
-_R = TypeVar("_R")
-_P = ParamSpec("_P")
-
 SUPPORT_OPENHOME = (
     MediaPlayerEntityFeature.SELECT_SOURCE
     | MediaPlayerEntityFeature.TURN_OFF
@@ -65,13 +61,13 @@ async def async_setup_entry(
     )
 
 
-_FuncType = Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]]
-_ReturnFuncType = Callable[
-    Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None]
+type _FuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], Awaitable[_R]]
+type _ReturnFuncType[_T, **_P, _R] = Callable[
+    Concatenate[_T, _P], Coroutine[Any, Any, _R | None]
 ]
 
 
-def catch_request_errors() -> (
+def catch_request_errors[_OpenhomeDeviceT: OpenhomeDevice, **_P, _R]() -> (
     Callable[
         [_FuncType[_OpenhomeDeviceT, _P, _R]], _ReturnFuncType[_OpenhomeDeviceT, _P, _R]
     ]
diff --git a/homeassistant/components/recorder/util.py b/homeassistant/components/recorder/util.py
index bb5446debc1..fe781f6841d 100644
--- a/homeassistant/components/recorder/util.py
+++ b/homeassistant/components/recorder/util.py
@@ -12,7 +12,7 @@ from itertools import islice
 import logging
 import os
 import time
-from typing import TYPE_CHECKING, Any, Concatenate, NoReturn, ParamSpec, TypeVar
+from typing import TYPE_CHECKING, Any, Concatenate, NoReturn
 
 from awesomeversion import (
     AwesomeVersion,
@@ -61,9 +61,6 @@ if TYPE_CHECKING:
 
     from . import Recorder
 
-_RecorderT = TypeVar("_RecorderT", bound="Recorder")
-_P = ParamSpec("_P")
-
 _LOGGER = logging.getLogger(__name__)
 
 RETRIES = 3
@@ -628,18 +625,20 @@ def _is_retryable_error(instance: Recorder, err: OperationalError) -> bool:
     )
 
 
-_FuncType = Callable[Concatenate[_RecorderT, _P], bool]
+type _FuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], _R]
 
 
-def retryable_database_job(
+def retryable_database_job[_RecorderT: Recorder, **_P](
     description: str,
-) -> Callable[[_FuncType[_RecorderT, _P]], _FuncType[_RecorderT, _P]]:
+) -> Callable[[_FuncType[_RecorderT, _P, bool]], _FuncType[_RecorderT, _P, bool]]:
     """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: _FuncType[_RecorderT, _P]) -> _FuncType[_RecorderT, _P]:
+    def decorator(
+        job: _FuncType[_RecorderT, _P, bool],
+    ) -> _FuncType[_RecorderT, _P, bool]:
         @functools.wraps(job)
         def wrapper(instance: _RecorderT, *args: _P.args, **kwargs: _P.kwargs) -> bool:
             try:
@@ -664,12 +663,9 @@ def retryable_database_job(
     return decorator
 
 
-_WrappedFuncType = Callable[Concatenate[_RecorderT, _P], None]
-
-
-def database_job_retry_wrapper(
+def database_job_retry_wrapper[_RecorderT: Recorder, **_P](
     description: str, attempts: int = 5
-) -> Callable[[_WrappedFuncType[_RecorderT, _P]], _WrappedFuncType[_RecorderT, _P]]:
+) -> Callable[[_FuncType[_RecorderT, _P, None]], _FuncType[_RecorderT, _P, None]]:
     """Try to execute a database job multiple times.
 
     This wrapper handles InnoDB deadlocks and lock timeouts.
@@ -679,8 +675,8 @@ def database_job_retry_wrapper(
     """
 
     def decorator(
-        job: _WrappedFuncType[_RecorderT, _P],
-    ) -> _WrappedFuncType[_RecorderT, _P]:
+        job: _FuncType[_RecorderT, _P, None],
+    ) -> _FuncType[_RecorderT, _P, None]:
         @functools.wraps(job)
         def wrapper(instance: _RecorderT, *args: _P.args, **kwargs: _P.kwargs) -> None:
             for attempt in range(attempts):
diff --git a/homeassistant/components/roku/helpers.py b/homeassistant/components/roku/helpers.py
index fc68e82c2d8..ad8bee63b6f 100644
--- a/homeassistant/components/roku/helpers.py
+++ b/homeassistant/components/roku/helpers.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 rokuecp import RokuConnectionError, RokuConnectionTimeoutError, RokuError
 
@@ -12,11 +12,10 @@ from homeassistant.exceptions import HomeAssistantError
 
 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]]
+type _FuncType[_T, **_P] = Callable[Concatenate[_T, _P], Awaitable[Any]]
+type _ReturnFuncType[_T, **_P] = Callable[
+    Concatenate[_T, _P], Coroutine[Any, Any, None]
+]
 
 
 def format_channel_name(channel_number: str, channel_name: str | None = None) -> str:
@@ -27,7 +26,7 @@ def format_channel_name(channel_number: str, channel_name: str | None = None) ->
     return channel_number
 
 
-def roku_exception_handler(
+def roku_exception_handler[_RokuEntityT: RokuEntity, **_P](
     ignore_timeout: bool = False,
 ) -> Callable[[_FuncType[_RokuEntityT, _P]], _ReturnFuncType[_RokuEntityT, _P]]:
     """Decorate Roku calls to handle Roku exceptions."""
diff --git a/homeassistant/components/sonos/helpers.py b/homeassistant/components/sonos/helpers.py
index 31becc1f032..8ced5a87b28 100644
--- a/homeassistant/components/sonos/helpers.py
+++ b/homeassistant/components/sonos/helpers.py
@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from collections.abc import Callable
 import logging
-from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar, overload
+from typing import TYPE_CHECKING, Any, Concatenate, overload
 
 from requests.exceptions import Timeout
 from soco import SoCo
@@ -26,29 +26,26 @@ UID_POSTFIX = "01400"
 
 _LOGGER = logging.getLogger(__name__)
 
-_T = TypeVar(
-    "_T", bound="SonosSpeaker | SonosMedia | SonosEntity | SonosHouseholdCoordinator"
+type _SonosEntitiesType = (
+    SonosSpeaker | SonosMedia | SonosEntity | SonosHouseholdCoordinator
 )
-_R = TypeVar("_R")
-_P = ParamSpec("_P")
-
-_FuncType = Callable[Concatenate[_T, _P], _R]
-_ReturnFuncType = Callable[Concatenate[_T, _P], _R | None]
+type _FuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], _R]
+type _ReturnFuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], _R | None]
 
 
 @overload
-def soco_error(
+def soco_error[_T: _SonosEntitiesType, **_P, _R](
     errorcodes: None = ...,
 ) -> Callable[[_FuncType[_T, _P, _R]], _FuncType[_T, _P, _R]]: ...
 
 
 @overload
-def soco_error(
+def soco_error[_T: _SonosEntitiesType, **_P, _R](
     errorcodes: list[str],
 ) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]: ...
 
 
-def soco_error(
+def soco_error[_T: _SonosEntitiesType, **_P, _R](
     errorcodes: list[str] | None = None,
 ) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]:
     """Filter out specified UPnP errors and raise exceptions for service calls."""
diff --git a/homeassistant/components/zha/core/cluster_handlers/__init__.py b/homeassistant/components/zha/core/cluster_handlers/__init__.py
index 7425a408745..8833d5c116f 100644
--- a/homeassistant/components/zha/core/cluster_handlers/__init__.py
+++ b/homeassistant/components/zha/core/cluster_handlers/__init__.py
@@ -7,7 +7,7 @@ import contextlib
 from enum import Enum
 import functools
 import logging
-from typing import TYPE_CHECKING, Any, ParamSpec, TypedDict
+from typing import TYPE_CHECKING, Any, TypedDict
 
 import zigpy.exceptions
 import zigpy.util
@@ -51,10 +51,8 @@ _LOGGER = logging.getLogger(__name__)
 RETRYABLE_REQUEST_DECORATOR = zigpy.util.retryable_request(tries=3)
 UNPROXIED_CLUSTER_METHODS = {"general_command"}
 
-
-_P = ParamSpec("_P")
-_FuncType = Callable[_P, Awaitable[Any]]
-_ReturnFuncType = Callable[_P, Coroutine[Any, Any, Any]]
+type _FuncType[**_P] = Callable[_P, Awaitable[Any]]
+type _ReturnFuncType[**_P] = Callable[_P, Coroutine[Any, Any, Any]]
 
 
 @contextlib.contextmanager
@@ -75,7 +73,7 @@ def wrap_zigpy_exceptions() -> Iterator[None]:
         raise HomeAssistantError(message) from exc
 
 
-def retry_request(func: _FuncType[_P]) -> _ReturnFuncType[_P]:
+def retry_request[**_P](func: _FuncType[_P]) -> _ReturnFuncType[_P]:
     """Send a request with retries and wrap expected zigpy exceptions."""
 
     @functools.wraps(func)
diff --git a/homeassistant/helpers/singleton.py b/homeassistant/helpers/singleton.py
index d11a4cc627c..20e4ee82162 100644
--- a/homeassistant/helpers/singleton.py
+++ b/homeassistant/helpers/singleton.py
@@ -5,26 +5,26 @@ from __future__ import annotations
 import asyncio
 from collections.abc import Callable
 import functools
-from typing import Any, TypeVar, cast, overload
+from typing import Any, cast, overload
 
 from homeassistant.core import HomeAssistant
 from homeassistant.loader import bind_hass
 from homeassistant.util.hass_dict import HassKey
 
-_T = TypeVar("_T")
-
-_FuncType = Callable[[HomeAssistant], _T]
+type _FuncType[_T] = Callable[[HomeAssistant], _T]
 
 
 @overload
-def singleton(data_key: HassKey[_T]) -> Callable[[_FuncType[_T]], _FuncType[_T]]: ...
+def singleton[_T](
+    data_key: HassKey[_T],
+) -> Callable[[_FuncType[_T]], _FuncType[_T]]: ...
 
 
 @overload
-def singleton(data_key: str) -> Callable[[_FuncType[_T]], _FuncType[_T]]: ...
+def singleton[_T](data_key: str) -> Callable[[_FuncType[_T]], _FuncType[_T]]: ...
 
 
-def singleton(data_key: Any) -> Callable[[_FuncType[_T]], _FuncType[_T]]:
+def singleton[_T](data_key: Any) -> Callable[[_FuncType[_T]], _FuncType[_T]]:
     """Decorate a function that should be called once per instance.
 
     Result will be cached and simultaneous calls will be handled.