apifrom.utils.rate_limit

Overview

Classes

• FixedWindowRateLimiter

• RateLimiter

• TokenBucketRateLimiter

Classes

FixedWindowRateLimiter(rate, period, key_func = None, error_message = None):bases: RateLimiter

A fixed window implementation of rate limiting.

Initialize a fixed window rate limiter.

param rate:

Maximum number of requests allowed in the window

param period:

Window size in seconds

param key_func:

Function to extract a key from the request

param error_message:

Custom error message when rate limit is exceeded

apifrom.utils.rate_limit.is_rate_limited(request)

:async:

Check if a request exceeds the rate limit using a fixed window algorithm.

param request:

The request to check

returns:

True if the request is rate limited, False otherwise

rtype:

bool

class apifrom.utils.rate_limit.RateLimiter(rate, period, burst=None, key_func=None, error_message=None)

Parameters:

• rate (int)

• period (int)

• burst (Optional[int])

• key_func (Optional[Callable[[Any], str]])

• error_message (Optional[str])

A rate limiter implementation that limits the number of requests in a specified time period.

Initialize a rate limiter.

param rate:

Maximum number of requests allowed in the time period

param period:

Time period in seconds

param burst:

Optional burst limit (maximum number of consecutive requests)

param key_func:

Optional function to extract a key from the request

param error_message:

Optional custom error message when rate limit is exceeded

apifrom.utils.rate_limit.acquire(request)

:async:

Try to acquire a token for the request.

param request:

The request to acquire a token for

returns:

True if a token was acquired, False if rate limited

rtype:

bool

apifrom.utils.rate_limit.get_remaining(request)

Get the number of remaining requests allowed.

param request:

The request to check

returns:

Number of remaining requests allowed

rtype:

int

apifrom.utils.rate_limit.get_reset_time(request)

Get the time until the rate limit resets.

param request:

The request to check

returns:

Time in seconds until the rate limit resets

rtype:

float

apifrom.utils.rate_limit.is_rate_limited(request)

:async:

Check if a request exceeds the rate limit.

param request:

The request to check

returns:

True if the request is rate limited, False otherwise

rtype:

bool

TokenBucketRateLimiter(rate, period, burst = None, key_func = None, error_message = None):bases: RateLimiter

A token bucket implementation of rate limiting.

Initialize a token bucket rate limiter.

param rate:

Rate at which tokens are added to the bucket (tokens/second)

param period:

How often to add tokens (in seconds)

param burst:

Maximum bucket size (defaults to rate)

param key_func:

Function to extract a key from the request

param error_message:

Custom error message when rate limit is exceeded

apifrom.utils.rate_limit.is_rate_limited(request)

:async:

Check if a request exceeds the rate limit using the token bucket algorithm.

param request:

The request to check

returns:

True if the request is rate limited, False otherwise

rtype:

bool