Use PEP 695 for decorator typing with type aliases (1) (#117662)

pull/114056/head
Marc Mueller 2024-05-18 11:42:39 +02:00 committed by GitHub
parent 907b9c42e5
commit 34ea781031
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 48 additions and 56 deletions

View File

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine from collections.abc import Awaitable, Callable, Coroutine
import functools import functools
import logging import logging
from typing import Any, Concatenate, ParamSpec, TypeVar from typing import Any, Concatenate
from androidtv.exceptions import LockNotAcquiredException from androidtv.exceptions import LockNotAcquiredException
@ -34,15 +34,13 @@ PREFIX_FIRETV = "Fire TV"
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
_ADBDeviceT = TypeVar("_ADBDeviceT", bound="AndroidTVEntity") type _FuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], Awaitable[_R]]
_R = TypeVar("_R") type _ReturnFuncType[_T, **_P, _R] = Callable[
_P = ParamSpec("_P") Concatenate[_T, _P], Coroutine[Any, Any, _R | None]
]
_FuncType = Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]
_ReturnFuncType = Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]]
def adb_decorator( def adb_decorator[_ADBDeviceT: AndroidTVEntity, **_P, _R](
override_available: bool = False, override_available: bool = False,
) -> Callable[[_FuncType[_ADBDeviceT, _P, _R]], _ReturnFuncType[_ADBDeviceT, _P, _R]]: ) -> Callable[[_FuncType[_ADBDeviceT, _P, _R]], _ReturnFuncType[_ADBDeviceT, _P, _R]]:
"""Wrap ADB methods and catch exceptions. """Wrap ADB methods and catch exceptions.

View File

@ -7,7 +7,7 @@ from collections import namedtuple
from collections.abc import Awaitable, Callable, Coroutine from collections.abc import Awaitable, Callable, Coroutine
import functools import functools
import logging import logging
from typing import Any, TypeVar, cast from typing import Any, cast
from aioasuswrt.asuswrt import AsusWrt as AsusWrtLegacy from aioasuswrt.asuswrt import AsusWrt as AsusWrtLegacy
from aiohttp import ClientSession from aiohttp import ClientSession
@ -56,15 +56,11 @@ WrtDevice = namedtuple("WrtDevice", ["ip", "name", "connected_to"])
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
type _FuncType[_T] = Callable[[_T], Awaitable[list[Any] | tuple[Any] | dict[str, Any]]]
_AsusWrtBridgeT = TypeVar("_AsusWrtBridgeT", bound="AsusWrtBridge") type _ReturnFuncType[_T] = Callable[[_T], Coroutine[Any, Any, dict[str, Any]]]
_FuncType = Callable[
[_AsusWrtBridgeT], Awaitable[list[Any] | tuple[Any] | dict[str, Any]]
]
_ReturnFuncType = Callable[[_AsusWrtBridgeT], Coroutine[Any, Any, dict[str, Any]]]
def handle_errors_and_zip( def handle_errors_and_zip[_AsusWrtBridgeT: AsusWrtBridge](
exceptions: type[Exception] | tuple[type[Exception], ...], keys: list[str] | None exceptions: type[Exception] | tuple[type[Exception], ...], keys: list[str] | None
) -> Callable[[_FuncType[_AsusWrtBridgeT]], _ReturnFuncType[_AsusWrtBridgeT]]: ) -> Callable[[_FuncType[_AsusWrtBridgeT]], _ReturnFuncType[_AsusWrtBridgeT]]:
"""Run library methods and zip results or manage exceptions.""" """Run library methods and zip results or manage exceptions."""

View File

@ -8,7 +8,7 @@ from datetime import datetime
from functools import wraps from functools import wraps
import json import json
import logging import logging
from typing import TYPE_CHECKING, Any, Concatenate, ParamSpec, TypeVar from typing import TYPE_CHECKING, Any, Concatenate
import pychromecast import pychromecast
from pychromecast.controllers.homeassistant import HomeAssistantController from pychromecast.controllers.homeassistant import HomeAssistantController
@ -85,18 +85,12 @@ APP_IDS_UNRELIABLE_MEDIA_INFO = ("Netflix",)
CAST_SPLASH = "https://www.home-assistant.io/images/cast/splash.png" CAST_SPLASH = "https://www.home-assistant.io/images/cast/splash.png"
type _FuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], _R]
_CastDeviceT = TypeVar("_CastDeviceT", bound="CastDevice")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_FuncType = Callable[Concatenate[_CastDeviceT, _P], _R]
_ReturnFuncType = Callable[Concatenate[_CastDeviceT, _P], _R]
def api_error( def api_error[_CastDeviceT: CastDevice, **_P, _R](
func: _FuncType[_CastDeviceT, _P, _R], func: _FuncType[_CastDeviceT, _P, _R],
) -> _ReturnFuncType[_CastDeviceT, _P, _R]: ) -> _FuncType[_CastDeviceT, _P, _R]:
"""Handle PyChromecastError and reraise a HomeAssistantError.""" """Handle PyChromecastError and reraise a HomeAssistantError."""
@wraps(func) @wraps(func)

View File

@ -8,7 +8,7 @@ from dataclasses import dataclass
from enum import Enum from enum import Enum
from functools import partial, wraps from functools import partial, wraps
import logging import logging
from typing import Any, Concatenate, ParamSpec, TypeVar from typing import Any, Concatenate
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
@ -28,15 +28,13 @@ from .handler import (
async_update_addon, async_update_addon,
) )
_AddonManagerT = TypeVar("_AddonManagerT", bound="AddonManager") type _FuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], Awaitable[_R]]
_R = TypeVar("_R") type _ReturnFuncType[_T, **_P, _R] = Callable[
_P = ParamSpec("_P") Concatenate[_T, _P], Coroutine[Any, Any, _R]
]
_FuncType = Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]
_ReturnFuncType = Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]]
def api_error( def api_error[_AddonManagerT: AddonManager, **_P, _R](
error_message: str, error_message: str,
) -> Callable[ ) -> Callable[
[_FuncType[_AddonManagerT, _P, _R]], _ReturnFuncType[_AddonManagerT, _P, _R] [_FuncType[_AddonManagerT, _P, _R]], _ReturnFuncType[_AddonManagerT, _P, _R]

View File

@ -6,7 +6,7 @@ from collections.abc import Awaitable, Callable, Coroutine
from functools import reduce, wraps from functools import reduce, wraps
import logging import logging
from operator import ior from operator import ior
from typing import Any, ParamSpec from typing import Any
from pyheos import HeosError, const as heos_const from pyheos import HeosError, const as heos_const
@ -41,8 +41,6 @@ from .const import (
SIGNAL_HEOS_UPDATED, SIGNAL_HEOS_UPDATED,
) )
_P = ParamSpec("_P")
BASE_SUPPORTED_FEATURES = ( BASE_SUPPORTED_FEATURES = (
MediaPlayerEntityFeature.VOLUME_MUTE MediaPlayerEntityFeature.VOLUME_MUTE
| MediaPlayerEntityFeature.VOLUME_SET | MediaPlayerEntityFeature.VOLUME_SET
@ -90,11 +88,13 @@ async def async_setup_entry(
async_add_entities(devices, True) async_add_entities(devices, True)
_FuncType = Callable[_P, Awaitable[Any]] type _FuncType[**_P] = Callable[_P, Awaitable[Any]]
_ReturnFuncType = Callable[_P, Coroutine[Any, Any, None]] type _ReturnFuncType[**_P] = Callable[_P, Coroutine[Any, Any, None]]
def log_command_error(command: str) -> Callable[[_FuncType[_P]], _ReturnFuncType[_P]]: def log_command_error[**_P](
command: str,
) -> Callable[[_FuncType[_P]], _ReturnFuncType[_P]]:
"""Return decorator that logs command failure.""" """Return decorator that logs command failure."""
def decorator(func: _FuncType[_P]) -> _ReturnFuncType[_P]: def decorator(func: _FuncType[_P]) -> _ReturnFuncType[_P]:

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine from collections.abc import Callable, Coroutine
from functools import wraps from functools import wraps
from typing import Any, Concatenate, ParamSpec, TypeVar, overload from typing import Any, Concatenate, overload
from aiohttp.web import Request, Response, StreamResponse from aiohttp.web import Request, Response, StreamResponse
@ -13,16 +13,18 @@ from homeassistant.exceptions import Unauthorized
from .view import HomeAssistantView from .view import HomeAssistantView
_HomeAssistantViewT = TypeVar("_HomeAssistantViewT", bound=HomeAssistantView) type _ResponseType = Response | StreamResponse
_ResponseT = TypeVar("_ResponseT", bound=Response | StreamResponse) type _FuncType[_T, **_P, _R] = Callable[
_P = ParamSpec("_P") Concatenate[_T, Request, _P], Coroutine[Any, Any, _R]
_FuncType = Callable[
Concatenate[_HomeAssistantViewT, Request, _P], Coroutine[Any, Any, _ResponseT]
] ]
@overload @overload
def require_admin( def require_admin[
_HomeAssistantViewT: HomeAssistantView,
**_P,
_ResponseT: _ResponseType,
](
_func: None = None, _func: None = None,
*, *,
error: Unauthorized | None = None, error: Unauthorized | None = None,
@ -33,12 +35,20 @@ def require_admin(
@overload @overload
def require_admin( def require_admin[
_HomeAssistantViewT: HomeAssistantView,
**_P,
_ResponseT: _ResponseType,
](
_func: _FuncType[_HomeAssistantViewT, _P, _ResponseT], _func: _FuncType[_HomeAssistantViewT, _P, _ResponseT],
) -> _FuncType[_HomeAssistantViewT, _P, _ResponseT]: ... ) -> _FuncType[_HomeAssistantViewT, _P, _ResponseT]: ...
def require_admin( def require_admin[
_HomeAssistantViewT: HomeAssistantView,
**_P,
_ResponseT: _ResponseType,
](
_func: _FuncType[_HomeAssistantViewT, _P, _ResponseT] | None = None, _func: _FuncType[_HomeAssistantViewT, _P, _ResponseT] | None = None,
*, *,
error: Unauthorized | None = None, error: Unauthorized | None = None,

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable, Mapping from collections.abc import Callable, Mapping
import logging import logging
from typing import Any, Concatenate, ParamSpec, TypeVar from typing import Any, Concatenate
from pizone import Controller, Zone from pizone import Controller, Zone
import voluptuous as vol import voluptuous as vol
@ -48,11 +48,7 @@ from .const import (
IZONE, IZONE,
) )
_DeviceT = TypeVar("_DeviceT", bound="ControllerDevice | ZoneDevice") type _FuncType[_T, **_P, _R] = Callable[Concatenate[_T, _P], _R]
_T = TypeVar("_T")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_FuncType = Callable[Concatenate[_T, _P], _R]
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@ -119,7 +115,7 @@ async def async_setup_entry(
) )
def _return_on_connection_error( def _return_on_connection_error[_DeviceT: ControllerDevice | ZoneDevice, **_P, _R, _T](
ret: _T = None, # type: ignore[assignment] ret: _T = None, # type: ignore[assignment]
) -> Callable[[_FuncType[_DeviceT, _P, _R]], _FuncType[_DeviceT, _P, _R | _T]]: ) -> Callable[[_FuncType[_DeviceT, _P, _R]], _FuncType[_DeviceT, _P, _R | _T]]:
def wrap(func: _FuncType[_DeviceT, _P, _R]) -> _FuncType[_DeviceT, _P, _R | _T]: def wrap(func: _FuncType[_DeviceT, _P, _R]) -> _FuncType[_DeviceT, _P, _R | _T]: