Add ParamSpec TypeAliases [mypy 1.0] (#87597)

pull/80614/head^2
Marc Mueller 2023-02-07 15:56:26 +01:00 committed by GitHub
parent 1b23429f41
commit b478b4fa16
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 46 additions and 43 deletions

View File

@ -140,12 +140,13 @@ async def async_setup_entry(
)
_FuncType = Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]
_ReturnFuncType = Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]]
def adb_decorator(
override_available: bool = False,
) -> Callable[
[Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]],
Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]],
]:
) -> Callable[[_FuncType[_ADBDeviceT, _P, _R]], _ReturnFuncType[_ADBDeviceT, _P, _R]]:
"""Wrap ADB methods and catch exceptions.
Allows for overriding the available status of the ADB connection via the
@ -153,8 +154,8 @@ def adb_decorator(
"""
def _adb_decorator(
func: Callable[Concatenate[_ADBDeviceT, _P], Awaitable[_R]]
) -> Callable[Concatenate[_ADBDeviceT, _P], Coroutine[Any, Any, _R | None]]:
func: _FuncType[_ADBDeviceT, _P, _R]
) -> _ReturnFuncType[_ADBDeviceT, _P, _R]:
"""Wrap the provided ADB method and catch exceptions."""
@functools.wraps(func)

View File

@ -31,18 +31,20 @@ _AddonManagerT = TypeVar("_AddonManagerT", bound="AddonManager")
_R = TypeVar("_R")
_P = ParamSpec("_P")
_FuncType = Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]
_ReturnFuncType = Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]]
def api_error(
error_message: str,
) -> Callable[
[Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]],
Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]],
[_FuncType[_AddonManagerT, _P, _R]], _ReturnFuncType[_AddonManagerT, _P, _R]
]:
"""Handle HassioAPIError and raise a specific AddonError."""
def handle_hassio_api_error(
func: Callable[Concatenate[_AddonManagerT, _P], Awaitable[_R]]
) -> Callable[Concatenate[_AddonManagerT, _P], Coroutine[Any, Any, _R]]:
func: _FuncType[_AddonManagerT, _P, _R]
) -> _ReturnFuncType[_AddonManagerT, _P, _R]:
"""Handle a HassioAPIError."""
@wraps(func)

View File

@ -88,14 +88,14 @@ async def async_setup_entry(
async_add_entities(devices, True)
def log_command_error(
command: str,
) -> Callable[[Callable[_P, Awaitable[Any]]], Callable[_P, Coroutine[Any, Any, None]]]:
_FuncType = Callable[_P, Awaitable[Any]]
_ReturnFuncType = Callable[_P, Coroutine[Any, Any, None]]
def log_command_error(command: str) -> Callable[[_FuncType[_P]], _ReturnFuncType[_P]]:
"""Return decorator that logs command failure."""
def decorator(
func: Callable[_P, Awaitable[Any]]
) -> Callable[_P, Coroutine[Any, Any, None]]:
def decorator(func: _FuncType[_P]) -> _ReturnFuncType[_P]:
@wraps(func)
async def wrapper(*args: _P.args, **kwargs: _P.kwargs) -> None:
try:

View File

@ -79,17 +79,22 @@ async def async_setup_platform(
)
_FuncType = Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]]
_ReturnFuncType = Callable[
Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None]
]
def catch_request_errors() -> (
Callable[
[Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]]],
Callable[Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None]],
[_FuncType[_OpenhomeDeviceT, _P, _R]], _ReturnFuncType[_OpenhomeDeviceT, _P, _R]
]
):
"""Catch asyncio.TimeoutError, aiohttp.ClientError, UpnpError errors."""
def call_wrapper(
func: Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]]
) -> Callable[Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None]]:
func: _FuncType[_OpenhomeDeviceT, _P, _R]
) -> _ReturnFuncType[_OpenhomeDeviceT, _P, _R]:
"""Call wrapper for decorator."""
@functools.wraps(func)

View File

@ -582,20 +582,18 @@ def end_incomplete_runs(session: Session, start_time: datetime) -> None:
session.add(run)
_FuncType = Callable[Concatenate[_RecorderT, _P], bool]
def retryable_database_job(
description: str,
) -> Callable[
[Callable[Concatenate[_RecorderT, _P], bool]],
Callable[Concatenate[_RecorderT, _P], bool],
]:
) -> Callable[[_FuncType[_RecorderT, _P]], _FuncType[_RecorderT, _P]]:
"""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: Callable[Concatenate[_RecorderT, _P], bool]
) -> Callable[Concatenate[_RecorderT, _P], bool]:
def decorator(job: _FuncType[_RecorderT, _P]) -> _FuncType[_RecorderT, _P]:
@functools.wraps(job)
def wrapper(instance: _RecorderT, *args: _P.args, **kwargs: _P.kwargs) -> bool:
try:

View File

@ -14,6 +14,9 @@ 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]]
def format_channel_name(channel_number: str, channel_name: str | None = None) -> str:
"""Format a Roku Channel name."""
@ -25,15 +28,12 @@ def format_channel_name(channel_number: str, channel_name: str | None = None) ->
def roku_exception_handler(
ignore_timeout: bool = False,
) -> Callable[
[Callable[Concatenate[_RokuEntityT, _P], Awaitable[Any]]],
Callable[Concatenate[_RokuEntityT, _P], Coroutine[Any, Any, None]],
]:
) -> Callable[[_FuncType[_RokuEntityT, _P]], _ReturnFuncType[_RokuEntityT, _P]]:
"""Decorate Roku calls to handle Roku exceptions."""
def decorator(
func: Callable[Concatenate[_RokuEntityT, _P], Awaitable[Any]],
) -> Callable[Concatenate[_RokuEntityT, _P], Coroutine[Any, Any, None]]:
func: _FuncType[_RokuEntityT, _P]
) -> _ReturnFuncType[_RokuEntityT, _P]:
@wraps(func)
async def wrapper(
self: _RokuEntityT, *args: _P.args, **kwargs: _P.kwargs

View File

@ -31,33 +31,30 @@ _T = TypeVar(
_R = TypeVar("_R")
_P = ParamSpec("_P")
_FuncType = Callable[Concatenate[_T, _P], _R]
_ReturnFuncType = Callable[Concatenate[_T, _P], _R | None]
@overload
def soco_error(
errorcodes: None = ...,
) -> Callable[[Callable[Concatenate[_T, _P], _R]], Callable[Concatenate[_T, _P], _R]]:
) -> Callable[[_FuncType[_T, _P, _R]], _FuncType[_T, _P, _R]]:
...
@overload
def soco_error(
errorcodes: list[str],
) -> Callable[
[Callable[Concatenate[_T, _P], _R]], Callable[Concatenate[_T, _P], _R | None]
]:
) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]:
...
def soco_error(
errorcodes: list[str] | None = None,
) -> Callable[
[Callable[Concatenate[_T, _P], _R]], Callable[Concatenate[_T, _P], _R | None]
]:
) -> Callable[[_FuncType[_T, _P, _R]], _ReturnFuncType[_T, _P, _R]]:
"""Filter out specified UPnP errors and raise exceptions for service calls."""
def decorator(
funct: Callable[Concatenate[_T, _P], _R]
) -> Callable[Concatenate[_T, _P], _R | None]:
def decorator(funct: _FuncType[_T, _P, _R]) -> _ReturnFuncType[_T, _P, _R]:
"""Decorate functions."""
def wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> _R | None: