mirror of
https://github.com/home-assistant/core
synced 2024-10-01 05:30:36 +02:00
56 lines
1.5 KiB
Python
56 lines
1.5 KiB
Python
"""Helper to help coordinating calls."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import functools
|
|
from typing import Callable, TypeVar, cast
|
|
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.loader import bind_hass
|
|
|
|
T = TypeVar("T")
|
|
|
|
FUNC = Callable[[HomeAssistant], T]
|
|
|
|
|
|
def singleton(data_key: str) -> Callable[[FUNC], FUNC]:
|
|
"""Decorate a function that should be called once per instance.
|
|
|
|
Result will be cached and simultaneous calls will be handled.
|
|
"""
|
|
|
|
def wrapper(func: FUNC) -> FUNC:
|
|
"""Wrap a function with caching logic."""
|
|
if not asyncio.iscoroutinefunction(func):
|
|
|
|
@bind_hass
|
|
@functools.wraps(func)
|
|
def wrapped(hass: HomeAssistant) -> T:
|
|
if data_key not in hass.data:
|
|
hass.data[data_key] = func(hass)
|
|
return cast(T, hass.data[data_key])
|
|
|
|
return wrapped
|
|
|
|
@bind_hass
|
|
@functools.wraps(func)
|
|
async def async_wrapped(hass: HomeAssistant) -> T:
|
|
if data_key not in hass.data:
|
|
evt = hass.data[data_key] = asyncio.Event()
|
|
result = await func(hass)
|
|
hass.data[data_key] = result
|
|
evt.set()
|
|
return cast(T, result)
|
|
|
|
obj_or_evt = hass.data[data_key]
|
|
|
|
if isinstance(obj_or_evt, asyncio.Event):
|
|
await obj_or_evt.wait()
|
|
return cast(T, hass.data[data_key])
|
|
|
|
return cast(T, obj_or_evt)
|
|
|
|
return async_wrapped
|
|
|
|
return wrapper
|