Add decorator typing [homematicip_cloud] (#107555)

pull/107866/head
Marc Mueller 2024-01-12 11:32:03 +01:00 committed by GitHub
parent bec88e5e51
commit 827a1b1f48
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 18 additions and 4 deletions

View File

@ -1,17 +1,25 @@
"""Helper functions for Homematicip Cloud Integration.""" """Helper functions for Homematicip Cloud Integration."""
from __future__ import annotations
from collections.abc import Callable, Coroutine
from functools import wraps from functools import wraps
import json import json
import logging import logging
from typing import Any, Concatenate, ParamSpec, TypeGuard, TypeVar
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from . import HomematicipGenericEntity from . import HomematicipGenericEntity
_HomematicipGenericEntityT = TypeVar(
"_HomematicipGenericEntityT", bound=HomematicipGenericEntity
)
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
def is_error_response(response) -> bool: def is_error_response(response: Any) -> TypeGuard[dict[str, Any]]:
"""Response from async call contains errors or not.""" """Response from async call contains errors or not."""
if isinstance(response, dict): if isinstance(response, dict):
return response.get("errorCode") not in ("", None) return response.get("errorCode") not in ("", None)
@ -19,13 +27,19 @@ def is_error_response(response) -> bool:
return False return False
def handle_errors(func): def handle_errors(
func: Callable[
Concatenate[_HomematicipGenericEntityT, _P], Coroutine[Any, Any, Any]
],
) -> Callable[Concatenate[_HomematicipGenericEntityT, _P], Coroutine[Any, Any, Any]]:
"""Handle async errors.""" """Handle async errors."""
@wraps(func) @wraps(func)
async def inner(self: HomematicipGenericEntity) -> None: async def inner(
self: _HomematicipGenericEntityT, *args: _P.args, **kwargs: _P.kwargs
) -> None:
"""Handle errors from async call.""" """Handle errors from async call."""
result = await func(self) result = await func(self, *args, **kwargs)
if is_error_response(result): if is_error_response(result):
_LOGGER.error( _LOGGER.error(
"Error while execute function %s: %s", "Error while execute function %s: %s",