Use PEP 695 for decorator typing (1) (#117638)

This commit is contained in:
Marc Mueller 2024-05-17 16:27:02 +02:00 committed by GitHub
parent 067c9e63e9
commit 34bd291615
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 33 additions and 90 deletions

View file

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
import sharp_aquos_rc
import voluptuous as vol
@ -28,9 +28,6 @@ import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
_SharpAquosTVDeviceT = TypeVar("_SharpAquosTVDeviceT", bound="SharpAquosTVDevice")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
DEFAULT_NAME = "Sharp Aquos TV"
@ -85,7 +82,7 @@ def setup_platform(
add_entities([SharpAquosTVDevice(name, remote, power_on_enabled)])
def _retry(
def _retry[_SharpAquosTVDeviceT: SharpAquosTVDevice, **_P](
func: Callable[Concatenate[_SharpAquosTVDeviceT, _P], Any],
) -> Callable[Concatenate[_SharpAquosTVDeviceT, _P], None]:
"""Handle query retries."""

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine
import functools
import logging
from typing import Any, ParamSpec, TypeVar
from typing import Any
from arcam.fmj import ConnectionFailed, SourceCodes
from arcam.fmj.state import State
@ -36,9 +36,6 @@ from .const import (
SIGNAL_CLIENT_STOPPED,
)
_R = TypeVar("_R")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
@ -64,7 +61,7 @@ async def async_setup_entry(
)
def convert_exception(
def convert_exception[**_P, _R](
func: Callable[_P, Coroutine[Any, Any, _R]],
) -> Callable[_P, Coroutine[Any, Any, _R]]:
"""Return decorator to convert a connection error into a home assistant error."""

View file

@ -7,7 +7,7 @@ from datetime import datetime, timedelta
from functools import wraps
import logging
from types import MappingProxyType
from typing import Any, Concatenate, Final, ParamSpec, TypeVar
from typing import Any, Concatenate, Final
from pybravia import (
BraviaAuthError,
@ -35,14 +35,12 @@ from .const import (
SourceType,
)
_BraviaTVCoordinatorT = TypeVar("_BraviaTVCoordinatorT", bound="BraviaTVCoordinator")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
SCAN_INTERVAL: Final = timedelta(seconds=10)
def catch_braviatv_errors(
def catch_braviatv_errors[_BraviaTVCoordinatorT: BraviaTVCoordinator, **_P](
func: Callable[Concatenate[_BraviaTVCoordinatorT, _P], Awaitable[None]],
) -> Callable[Concatenate[_BraviaTVCoordinatorT, _P], Coroutine[Any, Any, None]]:
"""Catch Bravia errors."""

View file

@ -9,7 +9,7 @@ import dataclasses
from functools import wraps
from http import HTTPStatus
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
import aiohttp
from aiohttp import web
@ -116,11 +116,7 @@ def async_setup(hass: HomeAssistant) -> None:
)
_HassViewT = TypeVar("_HassViewT", bound=HomeAssistantView)
_P = ParamSpec("_P")
def _handle_cloud_errors(
def _handle_cloud_errors[_HassViewT: HomeAssistantView, **_P](
handler: Callable[
Concatenate[_HassViewT, web.Request, _P], Awaitable[web.Response]
],

View file

@ -7,7 +7,7 @@ import copy
from functools import wraps
import logging
import time
from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar
from typing import TYPE_CHECKING, Any, Concatenate
from bluepy.btle import BTLEException
import decora
@ -29,10 +29,6 @@ if TYPE_CHECKING:
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
_DecoraLightT = TypeVar("_DecoraLightT", bound="DecoraLight")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
@ -60,7 +56,7 @@ PLATFORM_SCHEMA = vol.Schema(
)
def retry(
def retry[_DecoraLightT: DecoraLight, **_P, _R](
method: Callable[Concatenate[_DecoraLightT, _P], _R],
) -> Callable[Concatenate[_DecoraLightT, _P], _R | None]:
"""Retry bluetooth commands."""

View file

@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable, Coroutine
from datetime import timedelta
from functools import wraps
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
from denonavr import DenonAVR
from denonavr.const import (
@ -100,11 +100,6 @@ TELNET_EVENTS = {
"Z3",
}
_DenonDeviceT = TypeVar("_DenonDeviceT", bound="DenonDevice")
_R = TypeVar("_R")
_P = ParamSpec("_P")
DENON_STATE_MAPPING = {
STATE_ON: MediaPlayerState.ON,
STATE_OFF: MediaPlayerState.OFF,
@ -164,7 +159,7 @@ async def async_setup_entry(
async_add_entities(entities, update_before_add=True)
def async_log_errors(
def async_log_errors[_DenonDeviceT: DenonDevice, **_P, _R](
func: Callable[Concatenate[_DenonDeviceT, _P], Awaitable[_R]],
) -> Callable[Concatenate[_DenonDeviceT, _P], Coroutine[Any, Any, _R | None]]:
"""Log errors occurred when calling a Denon AVR receiver.

View file

@ -7,7 +7,7 @@ from collections.abc import Awaitable, Callable, Coroutine, Sequence
import contextlib
from datetime import datetime, timedelta
import functools
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
from async_upnp_client.client import UpnpService, UpnpStateVariable
from async_upnp_client.const import NotificationSubType
@ -52,11 +52,6 @@ from .data import EventListenAddr, get_domain_data
PARALLEL_UPDATES = 0
_DlnaDmrEntityT = TypeVar("_DlnaDmrEntityT", bound="DlnaDmrEntity")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_TRANSPORT_STATE_TO_MEDIA_PLAYER_STATE = {
TransportState.PLAYING: MediaPlayerState.PLAYING,
TransportState.TRANSITIONING: MediaPlayerState.PLAYING,
@ -68,7 +63,7 @@ _TRANSPORT_STATE_TO_MEDIA_PLAYER_STATE = {
}
def catch_request_errors(
def catch_request_errors[_DlnaDmrEntityT: DlnaDmrEntity, **_P, _R](
func: Callable[Concatenate[_DlnaDmrEntityT, _P], Awaitable[_R]],
) -> Callable[Concatenate[_DlnaDmrEntityT, _P], Coroutine[Any, Any, _R | None]]:
"""Catch UpnpError errors."""

View file

@ -8,7 +8,7 @@ from dataclasses import dataclass
from enum import StrEnum
import functools
from functools import cached_property
from typing import Any, TypeVar, cast
from typing import Any, cast
from async_upnp_client.aiohttp import AiohttpSessionRequester
from async_upnp_client.client import UpnpRequester
@ -43,9 +43,6 @@ from .const import (
STREAMABLE_PROTOCOLS,
)
_DlnaDmsDeviceMethod = TypeVar("_DlnaDmsDeviceMethod", bound="DmsDeviceSource")
_R = TypeVar("_R")
class DlnaDmsData:
"""Storage class for domain global data."""
@ -124,7 +121,7 @@ class ActionError(DlnaDmsDeviceError):
"""Error when calling a UPnP Action on the device."""
def catch_request_errors(
def catch_request_errors[_DlnaDmsDeviceMethod: DmsDeviceSource, _R](
func: Callable[[_DlnaDmsDeviceMethod, str], Coroutine[Any, Any, _R]],
) -> Callable[[_DlnaDmsDeviceMethod, str], Coroutine[Any, Any, _R]]:
"""Catch UpnpError errors."""

View file

@ -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 duotecno.unit import BaseUnit
@ -47,11 +47,7 @@ class DuotecnoEntity(Entity):
return self._unit.is_available()
_T = TypeVar("_T", bound="DuotecnoEntity")
_P = ParamSpec("_P")
def api_call(
def api_call[_T: DuotecnoEntity, **_P](
func: Callable[Concatenate[_T, _P], Awaitable[None]],
) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]:
"""Catch command exceptions."""

View file

@ -4,16 +4,12 @@ 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 . import EvilGeniusEntity
_EvilGeniusEntityT = TypeVar("_EvilGeniusEntityT", bound=EvilGeniusEntity)
_R = TypeVar("_R")
_P = ParamSpec("_P")
def update_when_done(
def update_when_done[_EvilGeniusEntityT: EvilGeniusEntity, **_P, _R](
func: Callable[Concatenate[_EvilGeniusEntityT, _P], Awaitable[_R]],
) -> Callable[Concatenate[_EvilGeniusEntityT, _P], Coroutine[Any, Any, _R]]:
"""Decorate function to trigger update when function is done."""

View file

@ -6,7 +6,7 @@ from collections.abc import Callable, Coroutine, Iterable
from dataclasses import dataclass
from datetime import timedelta
from functools import wraps
from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar
from typing import TYPE_CHECKING, Any, Concatenate
from aioguardian.errors import GuardianError
@ -20,14 +20,10 @@ from .const import LOGGER
if TYPE_CHECKING:
from . import GuardianEntity
_GuardianEntityT = TypeVar("_GuardianEntityT", bound=GuardianEntity)
DEFAULT_UPDATE_INTERVAL = timedelta(seconds=30)
SIGNAL_REBOOT_REQUESTED = "guardian_reboot_requested_{0}"
_P = ParamSpec("_P")
@dataclass
class EntityDomainReplacementStrategy:
@ -64,7 +60,7 @@ def async_finish_entity_domain_replacements(
@callback
def convert_exceptions_to_homeassistant_error(
def convert_exceptions_to_homeassistant_error[_GuardianEntityT: GuardianEntity, **_P](
func: Callable[Concatenate[_GuardianEntityT, _P], Coroutine[Any, Any, Any]],
) -> Callable[Concatenate[_GuardianEntityT, _P], Coroutine[Any, Any, None]]:
"""Decorate to handle exceptions from the Guardian API."""

View file

@ -7,7 +7,7 @@ from collections.abc import Callable, Coroutine
from http import HTTPStatus
import logging
import os
from typing import Any, ParamSpec
from typing import Any
import aiohttp
from yarl import URL
@ -24,8 +24,6 @@ from homeassistant.loader import bind_hass
from .const import ATTR_DISCOVERY, ATTR_MESSAGE, ATTR_RESULT, DOMAIN, X_HASS_SOURCE
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
@ -33,7 +31,7 @@ class HassioAPIError(RuntimeError):
"""Return if a API trow a error."""
def _api_bool(
def _api_bool[**_P](
funct: Callable[_P, Coroutine[Any, Any, dict[str, Any]]],
) -> Callable[_P, Coroutine[Any, Any, bool]]:
"""Return a boolean."""
@ -49,7 +47,7 @@ def _api_bool(
return _wrapper
def api_data(
def api_data[**_P](
funct: Callable[_P, Coroutine[Any, Any, dict[str, Any]]],
) -> Callable[_P, Coroutine[Any, Any, Any]]:
"""Return data of an api."""

View file

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from functools import wraps
import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from typing import Any, Concatenate
from aiohttp.web_exceptions import HTTPException
from apyhiveapi import Auth, Hive
@ -28,9 +28,6 @@ from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, PLATFORM_LOOKUP, PLATFORMS
_HiveEntityT = TypeVar("_HiveEntityT", bound="HiveEntity")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
CONFIG_SCHEMA = vol.Schema(
@ -131,7 +128,7 @@ async def async_remove_config_entry_device(
return True
def refresh_system(
def refresh_system[_HiveEntityT: HiveEntity, **_P](
func: Callable[Concatenate[_HiveEntityT, _P], Awaitable[Any]],
) -> Callable[Concatenate[_HiveEntityT, _P], Coroutine[Any, Any, None]]:
"""Force update all entities after state change."""

View file

@ -6,17 +6,12 @@ from collections.abc import Callable, Coroutine
from functools import wraps
import json
import logging
from typing import Any, Concatenate, ParamSpec, TypeGuard, TypeVar
from typing import Any, Concatenate, TypeGuard
from homeassistant.exceptions import HomeAssistantError
from . import HomematicipGenericEntity
_HomematicipGenericEntityT = TypeVar(
"_HomematicipGenericEntityT", bound=HomematicipGenericEntity
)
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__)
@ -28,7 +23,7 @@ def is_error_response(response: Any) -> TypeGuard[dict[str, Any]]:
return False
def handle_errors(
def handle_errors[_HomematicipGenericEntityT: HomematicipGenericEntity, **_P](
func: Callable[
Concatenate[_HomematicipGenericEntityT, _P], Coroutine[Any, Any, Any]
],

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 homewizard_energy.errors import DisabledError, RequestError
@ -12,11 +12,8 @@ from homeassistant.exceptions import HomeAssistantError
from .const import DOMAIN
from .entity import HomeWizardEntity
_HomeWizardEntityT = TypeVar("_HomeWizardEntityT", bound=HomeWizardEntity)
_P = ParamSpec("_P")
def homewizard_exception_handler(
def homewizard_exception_handler[_HomeWizardEntityT: HomeWizardEntity, **_P](
func: Callable[Concatenate[_HomeWizardEntityT, _P], Coroutine[Any, Any, Any]],
) -> Callable[Concatenate[_HomeWizardEntityT, _P], Coroutine[Any, Any, None]]:
"""Decorate HomeWizard Energy calls to handle HomeWizardEnergy exceptions.

View file

@ -10,7 +10,7 @@ from http import HTTPStatus
from ipaddress import IPv4Address, IPv6Address, ip_address
import logging
from socket import gethostbyaddr, herror
from typing import Any, Concatenate, Final, ParamSpec, TypeVar
from typing import Any, Concatenate, Final
from aiohttp.web import (
AppKey,
@ -32,9 +32,6 @@ from homeassistant.util import dt as dt_util, yaml
from .const import KEY_HASS
from .view import HomeAssistantView
_HassViewT = TypeVar("_HassViewT", bound=HomeAssistantView)
_P = ParamSpec("_P")
_LOGGER: Final = logging.getLogger(__name__)
KEY_BAN_MANAGER = AppKey["IpBanManager"]("ha_banned_ips_manager")
@ -91,7 +88,7 @@ async def ban_middleware(
raise
def log_invalid_auth(
def log_invalid_auth[_HassViewT: HomeAssistantView, **_P](
func: Callable[Concatenate[_HassViewT, Request, _P], Awaitable[Response]],
) -> Callable[Concatenate[_HassViewT, Request, _P], Coroutine[Any, Any, Response]]:
"""Decorate function to handle invalid auth or failed login attempts."""