ovdashboard/api/ovdashboard_api/async_helpers.py

"""
Some useful helpers for working in async contexts.
"""

from asyncio import get_running_loop
from functools import partial, wraps
from time import time
from typing import Awaitable, Callable, TypeVar

from async_lru import alru_cache

from .settings import SETTINGS

RT = TypeVar("RT")


def run_in_executor(
    function: Callable[..., RT]
) -> Callable[..., Awaitable[RT]]:
    """
    Decorator to make blocking a function call asyncio compatible.
    https://stackoverflow.com/questions/41063331/how-to-use-asyncio-with-existing-blocking-library/
    https://stackoverflow.com/a/53719009
    """

    @wraps(function)
    async def wrapper(*args, **kwargs) -> RT:
        loop = get_running_loop()
        return await loop.run_in_executor(
            None,
            partial(function, *args, **kwargs),
        )

    return wrapper


def get_ttl_hash() -> int:
    """
    Return the same value within `seconds` time period.
    https://stackoverflow.com/a/55900800
    """

    return round(time() / SETTINGS.cache_time)


def timed_alru_cache(*decorator_args, **decorator_kwargs):
    """
    Decorator which adds an (unused) param `ttl_hash`
    and the `alru_cache` annotation to a function.
    """

    def decorate(f):
        @alru_cache(*decorator_args, **decorator_kwargs)
        @wraps(f)
        async def wrapper(ttl_hash: int, *args, **kwargs):
            del ttl_hash

            return await f(*args, **kwargs)

        return wrapper

    return decorate