evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Use PEP 695 for decorator typing with type aliases (2) (#117663)

Browse source
This commit is contained in:
Marc Mueller 2024-05-18 11:41:46 +02:00 • committed by GitHub
parent d65437e347
commit 907b9c42e5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 41 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

14
homeassistant/components/openhome/media_player.py
View file

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

26
homeassistant/components/recorder/util.py
View file

@ -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):

13
homeassistant/components/roku/helpers.py
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 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."""

19
homeassistant/components/sonos/helpers.py
View file

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

10
homeassistant/components/zha/core/cluster_handlers/__init__.py
View file

@ -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)

14
homeassistant/helpers/singleton.py
View file

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