Type openhome error decorator (#70990)

This commit is contained in:
Marc Mueller 2022-04-28 17:04:19 +02:00 committed by GitHub
parent 4a574eb701
commit ac044c8ffa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -2,12 +2,15 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Awaitable, Callable, Coroutine
import functools import functools
import logging import logging
from typing import Any, TypeVar
import aiohttp import aiohttp
from async_upnp_client.client import UpnpError from async_upnp_client.client import UpnpError
from openhomedevice.device import Device from openhomedevice.device import Device
from typing_extensions import Concatenate, ParamSpec
import voluptuous as vol import voluptuous as vol
from homeassistant.components import media_source 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 from .const import ATTR_PIN_INDEX, DATA_OPENHOME, SERVICE_INVOKE_PIN
_OpenhomeDeviceT = TypeVar("_OpenhomeDeviceT", bound="OpenhomeDevice")
_R = TypeVar("_R")
_P = ParamSpec("_P")
SUPPORT_OPENHOME = ( SUPPORT_OPENHOME = (
MediaPlayerEntityFeature.SELECT_SOURCE MediaPlayerEntityFeature.SELECT_SOURCE
| MediaPlayerEntityFeature.TURN_OFF | 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.""" """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.""" """Call wrapper for decorator."""
@functools.wraps(func) @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.""" """Catch asyncio.TimeoutError, aiohttp.ClientError, UpnpError errors."""
try: try:
return await func(self, *args, **kwargs) return await func(self, *args, **kwargs)
except (asyncio.TimeoutError, aiohttp.ClientError, UpnpError): except (asyncio.TimeoutError, aiohttp.ClientError, UpnpError):
_LOGGER.error("Error during call %s", func.__name__) _LOGGER.error("Error during call %s", func.__name__)
return None
return wrapper return wrapper