jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-13 00:07:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Type iaqualink refresh decorator (#70988)

Browse Source
This commit is contained in:
Marc Mueller 2022-04-28 17:01:54 +02:00 committed by GitHub
parent 3db7f945eb
commit 27cf4165fa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 12 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
homeassistant/components/iaqualink/__init__.py
View File

@ -2,8 +2,10 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Awaitable, Callable, Coroutine
from functools import wraps from functools import wraps
import logging import logging
from typing import Any, TypeVar
import aiohttp.client_exceptions import aiohttp.client_exceptions
from iaqualink.client import AqualinkClient from iaqualink.client import AqualinkClient
@ -16,6 +18,7 @@ from iaqualink.device import (
AqualinkToggle, AqualinkToggle,
) )
from iaqualink.exception import AqualinkServiceException from iaqualink.exception import AqualinkServiceException
from typing_extensions import Concatenate, ParamSpec
import voluptuous as vol import voluptuous as vol
from homeassistant import config_entries from homeassistant import config_entries
@ -40,6 +43,9 @@ from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN, UPDATE_INTERVAL from .const import DOMAIN, UPDATE_INTERVAL
_AqualinkEntityT = TypeVar("_AqualinkEntityT", bound="AqualinkEntity")
_P = ParamSpec("_P")
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
ATTR_CONFIG = "config" ATTR_CONFIG = "config"
@ -193,11 +199,15 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return await hass.config_entries.async_unload_platforms(entry, platforms_to_unload) return await hass.config_entries.async_unload_platforms(entry, platforms_to_unload)
def refresh_system(func): def refresh_system(
func: Callable[Concatenate[_AqualinkEntityT, _P], Awaitable[Any]]
) -> Callable[Concatenate[_AqualinkEntityT, _P], Coroutine[Any, Any, None]]:
"""Force update all entities after state change.""" """Force update all entities after state change."""
@wraps(func) @wraps(func)
async def wrapper(self, *args, **kwargs): async def wrapper(
self: _AqualinkEntityT, *args: _P.args, **kwargs: _P.kwargs
) -> None:
"""Call decorated function and send update signal to all entities.""" """Call decorated function and send update signal to all entities."""
await func(self, *args, **kwargs) await func(self, *args, **kwargs)
async_dispatcher_send(self.hass, DOMAIN) async_dispatcher_send(self.hass, DOMAIN)