Use PEP 695 for decorator typing (3) (#117640)

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)

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.

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.

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."""

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."""

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."""

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."""

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."""

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.

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."""

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],

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."""

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,

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)