Add decorator typing [modern_forms] (#107558)

This commit is contained in:
Marc Mueller 2024-01-10 14:14:33 +01:00 committed by GitHub
parent fbbe03c93c
commit e91a159efa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

View File

@ -1,8 +1,10 @@
"""The Modern Forms integration.""" """The Modern Forms integration."""
from __future__ import annotations from __future__ import annotations
from collections.abc import Callable, Coroutine
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any, Concatenate, ParamSpec, TypeVar
from aiomodernforms import ( from aiomodernforms import (
ModernFormsConnectionError, ModernFormsConnectionError,
@ -24,6 +26,11 @@ from homeassistant.helpers.update_coordinator import (
from .const import DOMAIN from .const import DOMAIN
_ModernFormsDeviceEntityT = TypeVar(
"_ModernFormsDeviceEntityT", bound="ModernFormsDeviceEntity"
)
_P = ParamSpec("_P")
SCAN_INTERVAL = timedelta(seconds=5) SCAN_INTERVAL = timedelta(seconds=5)
PLATFORMS = [ PLATFORMS = [
Platform.BINARY_SENSOR, Platform.BINARY_SENSOR,
@ -64,14 +71,18 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
def modernforms_exception_handler(func): def modernforms_exception_handler(
func: Callable[Concatenate[_ModernFormsDeviceEntityT, _P], Any],
) -> Callable[Concatenate[_ModernFormsDeviceEntityT, _P], Coroutine[Any, Any, None]]:
"""Decorate Modern Forms calls to handle Modern Forms exceptions. """Decorate Modern Forms calls to handle Modern Forms exceptions.
A decorator that wraps the passed in function, catches Modern Forms errors, A decorator that wraps the passed in function, catches Modern Forms errors,
and handles the availability of the device in the data coordinator. and handles the availability of the device in the data coordinator.
""" """
async def handler(self, *args, **kwargs): async def handler(
self: _ModernFormsDeviceEntityT, *args: _P.args, **kwargs: _P.kwargs
) -> None:
try: try:
await func(self, *args, **kwargs) await func(self, *args, **kwargs)
self.coordinator.async_update_listeners() self.coordinator.async_update_listeners()