From ac044c8ffa52bd92232d92d37da13f8fb728321c Mon Sep 17 00:00:00 2001 From: Marc Mueller <30130371+cdce8p@users.noreply.github.com> Date: Thu, 28 Apr 2022 17:04:19 +0200 Subject: [PATCH] Type openhome error decorator (#70990) --- .../components/openhome/media_player.py | 21 ++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/homeassistant/components/openhome/media_player.py b/homeassistant/components/openhome/media_player.py index 95d9ca57992..fa9cce1cfb6 100644 --- a/homeassistant/components/openhome/media_player.py +++ b/homeassistant/components/openhome/media_player.py @@ -2,12 +2,15 @@ from __future__ import annotations import asyncio +from collections.abc import Awaitable, Callable, Coroutine import functools import logging +from typing import Any, TypeVar import aiohttp from async_upnp_client.client import UpnpError from openhomedevice.device import Device +from typing_extensions import Concatenate, ParamSpec import voluptuous as vol from homeassistant.components import media_source @@ -27,6 +30,10 @@ from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from .const import ATTR_PIN_INDEX, DATA_OPENHOME, SERVICE_INVOKE_PIN +_OpenhomeDeviceT = TypeVar("_OpenhomeDeviceT", bound="OpenhomeDevice") +_R = TypeVar("_R") +_P = ParamSpec("_P") + SUPPORT_OPENHOME = ( MediaPlayerEntityFeature.SELECT_SOURCE | MediaPlayerEntityFeature.TURN_OFF @@ -74,19 +81,27 @@ async def async_setup_platform( ) -def catch_request_errors(): +def catch_request_errors() -> Callable[ + [Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]]], + Callable[Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None]], +]: """Catch asyncio.TimeoutError, aiohttp.ClientError, UpnpError errors.""" - def call_wrapper(func): + def call_wrapper( + func: Callable[Concatenate[_OpenhomeDeviceT, _P], Awaitable[_R]] + ) -> Callable[Concatenate[_OpenhomeDeviceT, _P], Coroutine[Any, Any, _R | None]]: """Call wrapper for decorator.""" @functools.wraps(func) - async def wrapper(self, *args, **kwargs): + async def wrapper( + self: _OpenhomeDeviceT, *args: _P.args, **kwargs: _P.kwargs + ) -> _R | None: """Catch asyncio.TimeoutError, aiohttp.ClientError, UpnpError errors.""" try: return await func(self, *args, **kwargs) except (asyncio.TimeoutError, aiohttp.ClientError, UpnpError): _LOGGER.error("Error during call %s", func.__name__) + return None return wrapper