Type kodi error decorator (#70989)

pull/71013/head
Marc Mueller 2022-04-28 17:03:27 +02:00 committed by GitHub
parent 27cf4165fa
commit 4a574eb701
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 10 additions and 3 deletions

View File

@ -1,16 +1,18 @@
"""Support for interfacing with the XBMC/Kodi JSON-RPC API.""" """Support for interfacing with the XBMC/Kodi JSON-RPC API."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from datetime import timedelta from datetime import timedelta
from functools import wraps from functools import wraps
import logging import logging
import re import re
from typing import Any from typing import Any, TypeVar
import urllib.parse import urllib.parse
import jsonrpc_base import jsonrpc_base
from jsonrpc_base.jsonrpc import ProtocolError, TransportError from jsonrpc_base.jsonrpc import ProtocolError, TransportError
from pykodi import CannotConnectError from pykodi import CannotConnectError
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
@ -86,6 +88,9 @@ from .const import (
EVENT_TURN_ON, EVENT_TURN_ON,
) )
_KodiEntityT = TypeVar("_KodiEntityT", bound="KodiEntity")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
EVENT_KODI_CALL_METHOD_RESULT = "kodi_call_method_result" EVENT_KODI_CALL_METHOD_RESULT = "kodi_call_method_result"
@ -243,11 +248,13 @@ async def async_setup_entry(
async_add_entities([entity]) async_add_entities([entity])
def cmd(func): def cmd(
func: Callable[Concatenate[_KodiEntityT, _P], Awaitable[Any]]
) -> Callable[Concatenate[_KodiEntityT, _P], Coroutine[Any, Any, None]]:
"""Catch command exceptions.""" """Catch command exceptions."""
@wraps(func) @wraps(func)
async def wrapper(obj, *args, **kwargs): async def wrapper(obj: _KodiEntityT, *args: _P.args, **kwargs: _P.kwargs) -> None:
"""Wrap all command methods.""" """Wrap all command methods."""
try: try:
await func(obj, *args, **kwargs) await func(obj, *args, **kwargs)