Documentation for miscellaneous support functions

sankaku.utils.ratelimit(*, rps=None, rpm=None)

Limit the number of requests.

Parameters:

Name	Type	Description	Default
rps	Optional[int]	Request per second	None
rpm	Optional[int]	Requests per minute	None

Source code in sankaku/utils.py

def ratelimit(
        *,
        rps: Optional[int] = None,
        rpm: Optional[int] = None
) -> Callable[[Callable[_P, Awaitable[_T]]], Callable[_P, Awaitable[_T]]]:
    """Limit the number of requests.

    Args:
        rps: Request per second
        rpm: Requests per minute
    """
    if all(locals().values()):
        raise RateLimitError
    elif not any(locals().values()):
        raise TypeError("At least one argument must be specified.")

    sleep_time: float = (1 / rps) if rps else (60 / rpm)  # type: ignore

    def wrapper(func: Callable[_P, Awaitable[_T]]) -> Callable[_P, Awaitable[_T]]:
        @wraps(func)
        async def inner(*args: _P.args, **kwargs: _P.kwargs) -> _T:
            await asyncio.sleep(sleep_time)
            return await func(*args, **kwargs)

        return inner

    return wrapper

---

sankaku.utils.convert_ts_to_datetime(ts)

Convert timestamp in datetime dict into datetime class.

Source code in sankaku/utils.py

def convert_ts_to_datetime(ts: Timestamp) -> Optional[datetime]:
    """Convert timestamp in datetime dict into datetime class."""
    if ts.get("s") is None:
        return None
    return datetime.utcfromtimestamp(ts["s"]).astimezone()  # type: ignore