Use PEP 695 for decorator typing (2) (#117639)

This commit is contained in:
Marc Mueller 2024-05-17 16:31:01 +02:00 committed by GitHub
parent 25d1ca747b
commit c41962455e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 39 additions and 85 deletions

View file

@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable, Coroutine
from datetime import datetime
from functools import wraps
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
import httpx
from iaqualink.client import AqualinkClient
@ -39,9 +39,6 @@ from homeassistant.helpers.event import async_track_time_interval
from .const import DOMAIN, UPDATE_INTERVAL
_AqualinkEntityT = TypeVar("_AqualinkEntityT", bound="AqualinkEntity")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
ATTR_CONFIG = "config"
@ -182,7 +179,7 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return await hass.config_entries.async_unload_platforms(entry, platforms_to_unload)
def refresh_system(
def refresh_system[_AqualinkEntityT: AqualinkEntity, **_P](
func: Callable[Concatenate[_AqualinkEntityT, _P], Awaitable[Any]],
) -> Callable[Concatenate[_AqualinkEntityT, _P], Coroutine[Any, Any, None]]:
"""Force update all entities after state change."""

View file

@ -7,7 +7,7 @@ from datetime import timedelta
from functools import wraps
import logging
import re
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
from jsonrpc_base.jsonrpc import ProtocolError, TransportError
from pykodi import CannotConnectError
@ -71,9 +71,6 @@ from .const import (
EVENT_TURN_ON,
)
_KodiEntityT = TypeVar("_KodiEntityT", bound="KodiEntity")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
EVENT_KODI_CALL_METHOD_RESULT = "kodi_call_method_result"
@ -231,7 +228,7 @@ async def async_setup_entry(
async_add_entities([entity])
def cmd(
def cmd[_KodiEntityT: KodiEntity, **_P](
func: Callable[Concatenate[_KodiEntityT, _P], Awaitable[Any]],
) -> Callable[Concatenate[_KodiEntityT, _P], Coroutine[Any, Any, None]]:
"""Catch command exceptions."""

View file

@ -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 demetriek import LaMetricConnectionError, LaMetricError
@ -15,11 +15,8 @@ from .const import DOMAIN
from .coordinator import LaMetricDataUpdateCoordinator
from .entity import LaMetricEntity
_LaMetricEntityT = TypeVar("_LaMetricEntityT", bound=LaMetricEntity)
_P = ParamSpec("_P")
def lametric_exception_handler(
def lametric_exception_handler[_LaMetricEntityT: LaMetricEntity, **_P](
func: Callable[Concatenate[_LaMetricEntityT, _P], Coroutine[Any, Any, Any]],
) -> Callable[Concatenate[_LaMetricEntityT, _P], Coroutine[Any, Any, None]]:
"""Decorate LaMetric calls to handle LaMetric exceptions.

View file

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar, cast
from typing import Any, Concatenate, cast
from limitlessled import Color
from limitlessled.bridge import Bridge
@ -40,9 +40,6 @@ from homeassistant.helpers.restore_state import RestoreEntity
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util.color import color_hs_to_RGB, color_temperature_mired_to_kelvin
_LimitlessLEDGroupT = TypeVar("_LimitlessLEDGroupT", bound="LimitlessLEDGroup")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
CONF_BRIDGES = "bridges"
@ -176,7 +173,7 @@ def setup_platform(
add_entities(lights)
def state(
def state[_LimitlessLEDGroupT: LimitlessLEDGroup, **_P](
new_state: bool,
) -> Callable[
[Callable[Concatenate[_LimitlessLEDGroupT, int, Pipeline, _P], Any]],

View file

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine
from functools import wraps
from typing import Any, Concatenate, ParamSpec
from typing import Any, Concatenate
from matter_server.client.models.node import MatterNode
from matter_server.common.errors import MatterError
@ -18,8 +18,6 @@ from homeassistant.core import HomeAssistant, callback
from .adapter import MatterAdapter
from .helpers import MissingNode, get_matter, node_from_ha_device_id
_P = ParamSpec("_P")
ID = "id"
TYPE = "type"
DEVICE_ID = "device_id"
@ -93,7 +91,7 @@ def async_get_matter_adapter(
return _get_matter
def async_handle_failed_command(
def async_handle_failed_command[**_P](
func: Callable[
Concatenate[HomeAssistant, ActiveConnection, dict[str, Any], _P],
Coroutine[Any, Any, None],

View file

@ -4,7 +4,7 @@ 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 aiomodernforms import ModernFormsConnectionError, ModernFormsError
@ -17,11 +17,6 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import ModernFormsDataUpdateCoordinator
_ModernFormsDeviceEntityT = TypeVar(
"_ModernFormsDeviceEntityT", bound="ModernFormsDeviceEntity"
)
_P = ParamSpec("_P")
PLATFORMS = [
Platform.BINARY_SENSOR,
Platform.FAN,
@ -61,7 +56,10 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok
def modernforms_exception_handler(
def modernforms_exception_handler[
_ModernFormsDeviceEntityT: ModernFormsDeviceEntity,
**_P,
](
func: Callable[Concatenate[_ModernFormsDeviceEntityT, _P], Any],
) -> Callable[Concatenate[_ModernFormsDeviceEntityT, _P], Coroutine[Any, Any, None]]:
"""Decorate Modern Forms calls to handle Modern Forms exceptions.

View file

@ -7,7 +7,7 @@ import dataclasses
from functools import wraps
import logging
import random
from typing import Any, Concatenate, ParamSpec, TypeVar, cast
from typing import Any, Concatenate, cast
import python_otbr_api
from python_otbr_api import PENDING_DATASET_DELAY_TIMER, tlv_parser
@ -27,9 +27,6 @@ from homeassistant.helpers import issue_registry as ir
from .const import DOMAIN
_R = TypeVar("_R")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
INFO_URL_SKY_CONNECT = (
@ -61,7 +58,7 @@ def generate_random_pan_id() -> int:
return random.randint(0, 0xFFFE)
def _handle_otbr_error(
def _handle_otbr_error[**_P, _R](
func: Callable[Concatenate[OTBRData, _P], Coroutine[Any, Any, _R]],
) -> Callable[Concatenate[OTBRData, _P], Coroutine[Any, Any, _R]]:
"""Handle OTBR errors."""

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable
from functools import wraps
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar, cast
from typing import Any, Concatenate, cast
import plexapi.exceptions
import requests.exceptions
@ -46,14 +46,10 @@ from .helpers import get_plex_data, get_plex_server
from .media_browser import browse_media
from .services import process_plex_payload
_PlexMediaPlayerT = TypeVar("_PlexMediaPlayerT", bound="PlexMediaPlayer")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
def needs_session(
def needs_session[_PlexMediaPlayerT: PlexMediaPlayer, **_P, _R](
func: Callable[Concatenate[_PlexMediaPlayerT, _P], _R],
) -> Callable[Concatenate[_PlexMediaPlayerT, _P], _R | None]:
"""Ensure session is available for certain attributes."""

View file

@ -1,7 +1,7 @@
"""Utilities for Plugwise."""
from collections.abc import Awaitable, Callable, Coroutine
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
from plugwise.exceptions import PlugwiseException
@ -9,12 +9,8 @@ from homeassistant.exceptions import HomeAssistantError
from .entity import PlugwiseEntity
_PlugwiseEntityT = TypeVar("_PlugwiseEntityT", bound=PlugwiseEntity)
_R = TypeVar("_R")
_P = ParamSpec("_P")
def plugwise_command(
def plugwise_command[_PlugwiseEntityT: PlugwiseEntity, **_P, _R](
func: Callable[Concatenate[_PlugwiseEntityT, _P], Awaitable[_R]],
) -> Callable[Concatenate[_PlugwiseEntityT, _P], Coroutine[Any, Any, _R]]:
"""Decorate Plugwise calls that send commands/make changes to the device.

View file

@ -6,7 +6,7 @@ import asyncio
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
from regenmaschine.errors import RainMachineError
import voluptuous as vol
@ -110,11 +110,7 @@ VEGETATION_MAP = {
}
_T = TypeVar("_T", bound="RainMachineBaseSwitch")
_P = ParamSpec("_P")
def raise_on_request_error(
def raise_on_request_error[_T: RainMachineBaseSwitch, **_P](
func: Callable[Concatenate[_T, _P], Awaitable[None]],
) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]:
"""Define a decorator to raise on a request error."""

View file

@ -8,7 +8,7 @@ from dataclasses import dataclass
from datetime import datetime, timedelta
from functools import wraps
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar, cast
from typing import Any, Concatenate, cast
from renault_api.exceptions import RenaultException
from renault_api.kamereon import models
@ -22,13 +22,11 @@ from .const import DOMAIN
from .coordinator import RenaultDataUpdateCoordinator
LOGGER = logging.getLogger(__name__)
_T = TypeVar("_T")
_P = ParamSpec("_P")
def with_error_wrapping(
func: Callable[Concatenate[RenaultVehicleProxy, _P], Awaitable[_T]],
) -> Callable[Concatenate[RenaultVehicleProxy, _P], Coroutine[Any, Any, _T]]:
def with_error_wrapping[**_P, _R](
func: Callable[Concatenate[RenaultVehicleProxy, _P], Awaitable[_R]],
) -> Callable[Concatenate[RenaultVehicleProxy, _P], Coroutine[Any, Any, _R]]:
"""Catch Renault errors."""
@wraps(func)
@ -36,7 +34,7 @@ def with_error_wrapping(
self: RenaultVehicleProxy,
*args: _P.args,
**kwargs: _P.kwargs,
) -> _T:
) -> _R:
"""Catch RenaultException errors and raise HomeAssistantError."""
try:
return await func(self, *args, **kwargs)

View file

@ -1,7 +1,7 @@
"""Base class for Ring entity."""
from collections.abc import Callable
from typing import Any, Concatenate, Generic, ParamSpec, cast
from typing import Any, Concatenate, Generic, cast
from ring_doorbell import (
AuthenticationError,
@ -26,12 +26,9 @@ _RingCoordinatorT = TypeVar(
"_RingCoordinatorT",
bound=(RingDataCoordinator | RingNotificationsCoordinator),
)
_RingBaseEntityT = TypeVar("_RingBaseEntityT", bound="RingBaseEntity[Any, Any]")
_R = TypeVar("_R")
_P = ParamSpec("_P")
def exception_wrap(
def exception_wrap[_RingBaseEntityT: RingBaseEntity[Any, Any], **_P, _R](
func: Callable[Concatenate[_RingBaseEntityT, _P], _R],
) -> Callable[Concatenate[_RingBaseEntityT, _P], _R]:
"""Define a wrapper to catch exceptions and raise HomeAssistant errors."""

View file

@ -4,7 +4,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine
from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar
from typing import TYPE_CHECKING, Any, Concatenate
from pysensibo.model import MotionSensor, SensiboDevice
@ -15,11 +15,8 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, LOGGER, SENSIBO_ERRORS, TIMEOUT
from .coordinator import SensiboDataUpdateCoordinator
_T = TypeVar("_T", bound="SensiboDeviceBaseEntity")
_P = ParamSpec("_P")
def async_handle_api_call(
def async_handle_api_call[_T: SensiboDeviceBaseEntity, **_P](
function: Callable[Concatenate[_T, _P], Coroutine[Any, Any, Any]],
) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, Any]]:
"""Decorate api calls."""

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from dataclasses import dataclass
from functools import wraps
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
from sfrbox_api.bridge import SFRBox
from sfrbox_api.exceptions import SFRBoxError
@ -26,13 +26,10 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN
from .models import DomainData
_T = TypeVar("_T")
_P = ParamSpec("_P")
def with_error_wrapping(
func: Callable[Concatenate[SFRBoxButton, _P], Awaitable[_T]],
) -> Callable[Concatenate[SFRBoxButton, _P], Coroutine[Any, Any, _T]]:
def with_error_wrapping[**_P, _R](
func: Callable[Concatenate[SFRBoxButton, _P], Awaitable[_R]],
) -> Callable[Concatenate[SFRBoxButton, _P], Coroutine[Any, Any, _R]]:
"""Catch SFR errors."""
@wraps(func)
@ -40,7 +37,7 @@ def with_error_wrapping(
self: SFRBoxButton,
*args: _P.args,
**kwargs: _P.kwargs,
) -> _T:
) -> _R:
"""Catch SFRBoxError errors and raise HomeAssistantError."""
try:
return await func(self, *args, **kwargs)

View file

@ -6,7 +6,7 @@ from asyncio import run_coroutine_threadsafe
from collections.abc import Callable
from datetime import timedelta
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
import requests
from spotipy import SpotifyException
@ -35,10 +35,6 @@ from .browse_media import async_browse_media_internal
from .const import DOMAIN, MEDIA_PLAYER_PREFIX, PLAYABLE_MEDIA_TYPES, SPOTIFY_SCOPES
from .util import fetch_image_url
_SpotifyMediaPlayerT = TypeVar("_SpotifyMediaPlayerT", bound="SpotifyMediaPlayer")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL = timedelta(seconds=30)
@ -86,7 +82,7 @@ async def async_setup_entry(
async_add_entities([spotify], True)
def spotify_exception_handler(
def spotify_exception_handler[_SpotifyMediaPlayerT: SpotifyMediaPlayer, **_P, _R](
func: Callable[Concatenate[_SpotifyMediaPlayerT, _P], _R],
) -> Callable[Concatenate[_SpotifyMediaPlayerT, _P], _R | None]:
"""Decorate Spotify calls to handle Spotify exception.