limitor

Main module for rate limiting functionality.

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

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

Parameters:

Name	Type	Description	Default
capacity	int	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	A decorator that applies the rate limit to the function

Source code in limitor/__init__.py

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

    Args:
        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(LeakyBucketConfig(capacity=capacity, seconds=seconds), max_concurrent=max_concurrent)

    def decorator(func):
        async def wrapper(*args, **kwargs):
            async with bucket:
                return await func(*args, **kwargs)

        return wrapper

    return decorator

rate_limit(capacity=10, seconds=1, bucket_cls=SyncLeakyBucket)

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

Parameters:

Name	Type	Description	Default
capacity	int	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	A decorator that applies the rate limit to the function

Source code in limitor/__init__.py

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

    Args:
        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(LeakyBucketConfig(capacity=capacity, seconds=seconds))

    def decorator(func):
        def wrapper(*args, **kwargs):
            with bucket:
                return func(*args, **kwargs)

        return wrapper

    return decorator