hass-core/homeassistant/helpers/singleton.py

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

68 lines
1.9 KiB
Python
Raw Permalink Normal View History

"""Helper to help coordinating calls."""
2021-03-17 18:34:19 +01:00
from __future__ import annotations
import asyncio
from collections.abc import Callable
import functools
from typing import Any, cast, overload
from homeassistant.core import HomeAssistant
from homeassistant.loader import bind_hass
2024-05-07 10:53:13 +02:00
from homeassistant.util.hass_dict import HassKey
type _FuncType[_T] = Callable[[HomeAssistant], _T]
2024-05-07 10:53:13 +02:00
@overload
def singleton[_T](
data_key: HassKey[_T],
) -> Callable[[_FuncType[_T]], _FuncType[_T]]: ...
2024-05-07 10:53:13 +02:00
@overload
def singleton[_T](data_key: str) -> Callable[[_FuncType[_T]], _FuncType[_T]]: ...
2024-05-07 10:53:13 +02:00
def singleton[_T](data_key: Any) -> Callable[[_FuncType[_T]], _FuncType[_T]]:
"""Decorate a function that should be called once per instance.
Result will be cached and simultaneous calls will be handled.
"""
def wrapper(func: _FuncType[_T]) -> _FuncType[_T]:
"""Wrap a function with caching logic."""
if not asyncio.iscoroutinefunction(func):
@functools.lru_cache(maxsize=1)
@bind_hass
@functools.wraps(func)
2022-03-17 19:09:55 +01:00
def wrapped(hass: HomeAssistant) -> _T:
if data_key not in hass.data:
hass.data[data_key] = func(hass)
2022-03-17 19:09:55 +01:00
return cast(_T, hass.data[data_key])
return wrapped
@bind_hass
@functools.wraps(func)
async def async_wrapped(hass: HomeAssistant) -> Any:
if data_key not in hass.data:
evt = hass.data[data_key] = asyncio.Event()
2023-06-21 16:12:51 +02:00
result = await func(hass)
hass.data[data_key] = result
evt.set()
2022-03-17 19:09:55 +01:00
return cast(_T, result)
obj_or_evt = hass.data[data_key]
if isinstance(obj_or_evt, asyncio.Event):
await obj_or_evt.wait()
2022-03-17 19:09:55 +01:00
return cast(_T, hass.data[data_key])
2022-03-17 19:09:55 +01:00
return cast(_T, obj_or_evt)
return async_wrapped # type: ignore[return-value]
return wrapper