62 lines
1.5 KiB
Python
62 lines
1.5 KiB
Python
"""
|
|
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
|