54 lines
1.9 KiB
Python
54 lines
1.9 KiB
Python
"""Utils for Comelit."""
|
|
|
|
from collections.abc import Awaitable, Callable, Coroutine
|
|
from functools import wraps
|
|
from typing import Any, Concatenate
|
|
|
|
from aiocomelit.exceptions import CannotAuthenticate, CannotConnect, CannotRetrieveData
|
|
from aiohttp import ClientSession, CookieJar
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
from homeassistant.helpers import aiohttp_client
|
|
|
|
from .const import DOMAIN
|
|
from .entity import ComelitBridgeBaseEntity
|
|
|
|
|
|
async def async_client_session(hass: HomeAssistant) -> ClientSession:
|
|
"""Return a new aiohttp session."""
|
|
return aiohttp_client.async_create_clientsession(
|
|
hass, verify_ssl=False, cookie_jar=CookieJar(unsafe=True)
|
|
)
|
|
|
|
|
|
def bridge_api_call[_T: ComelitBridgeBaseEntity, **_P](
|
|
func: Callable[Concatenate[_T, _P], Awaitable[None]],
|
|
) -> Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]]:
|
|
"""Catch Bridge API call exceptions."""
|
|
|
|
@wraps(func)
|
|
async def cmd_wrapper(self: _T, *args: _P.args, **kwargs: _P.kwargs) -> None:
|
|
"""Wrap all command methods."""
|
|
try:
|
|
await func(self, *args, **kwargs)
|
|
except CannotConnect as err:
|
|
self.coordinator.last_update_success = False
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="cannot_connect",
|
|
translation_placeholders={"error": repr(err)},
|
|
) from err
|
|
except CannotRetrieveData as err:
|
|
self.coordinator.last_update_success = False
|
|
raise HomeAssistantError(
|
|
translation_domain=DOMAIN,
|
|
translation_key="cannot_retrieve_data",
|
|
translation_placeholders={"error": repr(err)},
|
|
) from err
|
|
except CannotAuthenticate:
|
|
self.coordinator.last_update_success = False
|
|
self.coordinator.config_entry.async_start_reauth(self.hass)
|
|
|
|
return cmd_wrapper
|