limitor

Main module for rate limiting functionality.

async_rate_limit(_func=None, *, capacity=10, seconds=1, max_concurrent=None, bucket_cls=AsyncLeakyBucket)

async_rate_limit(
    _func: None = None,
    *,
    capacity: float = 10,
    seconds: float = 1,
    max_concurrent: int | None = None,
    bucket_cls: type[AsyncRateLimit] = AsyncLeakyBucket
) -> Callable[
    [Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]
]

async_rate_limit(
    _func: Callable[P, Awaitable[R]],
    *,
    capacity: float = 10,
    seconds: float = 1,
    max_concurrent: int | None = None,
    bucket_cls: type[AsyncRateLimit] = AsyncLeakyBucket
) -> Callable[P, Awaitable[R]]

Decorator to apply an asynchronous leaky bucket rate limit to a function.

Parameters:

Name	Type	Description	Default
_func	Callable[P, Awaitable[R]] | None	Function to apply the rate limit to the function	None
capacity	float	Maximum number of requests allowed in the bucket, defaults to 10	10
seconds	float	Time period in seconds for the bucket to refill, defaults to 1	1
max_concurrent	int | None	Maximum number of concurrent requests allowed, defaults to None (no limit)	None
bucket_cls	type[AsyncRateLimit]	Bucket class, defaults to AsyncLeakyBucket	AsyncLeakyBucket

Returns:

Type	Description
Callable[P, Awaitable[R]] | Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]	A decorator that applies the rate limit to the function

Source code in limitor/__init__.py

def async_rate_limit[**P, R](
    _func: Callable[P, Awaitable[R]] | None = None,
    *,
    capacity: float = 10,
    seconds: float = 1,
    max_concurrent: int | None = None,
    bucket_cls: type[AsyncRateLimit] = AsyncLeakyBucket,
) -> Callable[P, Awaitable[R]] | Callable[[Callable[P, Awaitable[R]]], Callable[P, Awaitable[R]]]:
    """Decorator to apply an asynchronous leaky bucket rate limit to a function.

    Args:
        _func: Function to apply the rate limit to the function
        capacity: Maximum number of requests allowed in the bucket, defaults to 10
        seconds: Time period in seconds for the bucket to refill, defaults to 1
        max_concurrent: Maximum number of concurrent requests allowed, defaults to None (no limit)
        bucket_cls: Bucket class, defaults to AsyncLeakyBucket

    Returns:
        A decorator that applies the rate limit to the function
    """
    bucket = bucket_cls(BucketConfig(capacity=capacity, seconds=seconds), max_concurrent=max_concurrent)

    def decorator(func: Callable[P, Awaitable[R]]) -> Callable[P, Awaitable[R]]:
        @wraps(func)
        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            async with bucket:
                return await func(*args, **kwargs)

        return wrapper

    if _func is None:
        return decorator
    return decorator(_func)

rate_limit(_func=None, *, capacity=10, seconds=1, bucket_cls=SyncLeakyBucket)

rate_limit(
    _func: None = None,
    *,
    capacity: float = 10,
    seconds: float = 1,
    bucket_cls: type[SyncRateLimit] = SyncLeakyBucket
) -> Callable[[Callable[P, R]], Callable[P, R]]

rate_limit(
    _func: Callable[P, R],
    *,
    capacity: float = 10,
    seconds: float = 1,
    bucket_cls: type[SyncRateLimit] = SyncLeakyBucket
) -> Callable[P, R]

Decorator to apply a synchronous leaky bucket rate limit to a function.

Parameters:

Name	Type	Description	Default
_func	Callable[P, R] | None	Function to apply the rate limit to the function	None
capacity	float	Maximum number of requests allowed in the bucket, defaults to 10	10
seconds	float	Time period in seconds for the bucket to refill, defaults to 1	1
bucket_cls	type[SyncRateLimit]	Bucket class, defaults to SyncLeakyBucket	SyncLeakyBucket

Returns:

Type	Description
Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]	A decorator that applies the rate limit to the function

Source code in limitor/__init__.py

def rate_limit[**P, R](
    _func: Callable[P, R] | None = None,
    *,
    capacity: float = 10,
    seconds: float = 1,
    bucket_cls: type[SyncRateLimit] = SyncLeakyBucket,
) -> Callable[P, R] | Callable[[Callable[P, R]], Callable[P, R]]:
    """Decorator to apply a synchronous leaky bucket rate limit to a function.

    Args:
        _func: Function to apply the rate limit to the function
        capacity: Maximum number of requests allowed in the bucket, defaults to 10
        seconds: Time period in seconds for the bucket to refill, defaults to 1
        bucket_cls: Bucket class, defaults to SyncLeakyBucket

    Returns:
        A decorator that applies the rate limit to the function
    """
    bucket = bucket_cls(BucketConfig(capacity=capacity, seconds=seconds))

    def decorator(func: Callable[P, R]) -> Callable[P, R]:
        @wraps(func)
        def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            with bucket:
                return func(*args, **kwargs)

        return wrapper

    if _func is None:
        return decorator
    return decorator(_func)