apifrom.middleware.rate_limit

Rate limiting middleware for APIFromAnything.

This module provides middleware for rate limiting API requests to prevent abuse.

Overview

Classes

Classes

BaseMiddleware(**options):bases: abc.ABC

Base middleware class for APIFromAnything.

This abstract class defines the interface for middleware components. Middleware components can process requests and responses.

apifrom.middleware.rate_limit.options

Options for the middleware.

Type:

dict

Initialize a new BaseMiddleware instance.

param **options:

Options for the middleware.

FixedWindowRateLimiter(limit, window = 60):bases: RateLimiter

Fixed window rate limiter implementation.

This rate limiter uses a fixed time window to limit requests.

Initialize the fixed window rate limiter.

param limit:

Maximum number of requests allowed in the window

param window:

Time window in seconds

apifrom.middleware.rate_limit._get_current_window()

Get the current time window.

returns:

The current time window as an integer

apifrom.middleware.rate_limit.check_limit(key)

Check if a key has exceeded its rate limit.

param key:

The key to check

returns:

A tuple containing (allowed, limit_info)

apifrom.middleware.rate_limit.update(key)

Update the rate limit counter for a key.

param key:

The key to update

InMemoryRateLimitBackend:bases: RateLimitBackend

In-memory implementation of RateLimitBackend.

This backend stores rate limit data in memory.

Initialize a new InMemoryRateLimitBackend instance.

apifrom.middleware.rate_limit._cleanup()

Clean up expired keys.

apifrom.middleware.rate_limit.get(key)

Get rate limit data for a key.

param key:

The key to get data for

returns:

The rate limit data, or None if not found

apifrom.middleware.rate_limit.increment(key, amount=1, ttl=None)

Increment rate limit counter for a key.

param key:

The key to increment

param amount:

The amount to increment by

param ttl:

Time to live in seconds

returns:

The new counter value

apifrom.middleware.rate_limit.reset(key)

Reset rate limit data for a key.

param key:

The key to reset

apifrom.middleware.rate_limit.set(key, value, ttl=None)

Set rate limit data for a key.

param key:

The key to set data for

param value:

The data to set

param ttl:

Time to live in seconds

JSONResponse(content = None, status_code = 200, headers = None):bases: Response

JSON response for APIFromAnything.

This class represents an HTTP response with JSON content.

Initialize a new JSONResponse instance.

param content:

The response content.

param status_code:

The HTTP status code.

param headers:

HTTP headers.

class apifrom.middleware.rate_limit.RateLimit[source]

Decorator for controlling rate limiting on specific endpoints.

apifrom.middleware.rate_limit.exempt(func)
:staticmethod:

Exempt an endpoint from rate limiting.

param func:

The function to decorate

returns:

The decorated function

apifrom.middleware.rate_limit.limit(limit, window=60, key_func=None)
:staticmethod:

Apply a rate limit to an endpoint.

param limit:

Maximum number of requests allowed in the window

param window:

Time window in seconds

param key_func:

Function to extract the rate limit key from a request

returns:

A decorator function

class apifrom.middleware.rate_limit.RateLimitBackend[source]

Base class for rate limit backends.

Rate limit backends are responsible for storing and retrieving rate limit data.

apifrom.middleware.rate_limit.get(key)
:abstractmethod:

Get rate limit data for a key.

param key:

The key to get data for

returns:

The rate limit data

apifrom.middleware.rate_limit.increment(key, amount=1, ttl=None)
:abstractmethod:

Increment rate limit counter for a key.

param key:

The key to increment

param amount:

The amount to increment by

param ttl:

Time to live in seconds

returns:

The new counter value

apifrom.middleware.rate_limit.reset(key)
:abstractmethod:

Reset rate limit data for a key.

param key:

The key to reset

apifrom.middleware.rate_limit.set(key, value, ttl=None)
:abstractmethod:

Set rate limit data for a key.

param key:

The key to set data for

param value:

The data to set

param ttl:

Time to live in seconds

RateLimitMiddleware(limiter, key_func = None, exclude_routes = None, headers_enabled = True):bases: apifrom.middleware.base.BaseMiddleware

Middleware for rate limiting API requests.

Initialize the rate limit middleware.

param limiter:

The rate limiter to use

param key_func:

Function to extract the rate limit key from a request

param exclude_routes:

Routes to exclude from rate limiting

param headers_enabled:

Whether to include rate limit headers in responses

apifrom.middleware.rate_limit.__call__(scope, receive, send)
:async:

ASGI callable.

param scope:

The ASGI scope.

param receive:

The ASGI receive function.

param send:

The ASGI send function.

apifrom.middleware.rate_limit._add_rate_limit_headers(response, limit_info)

Add rate limit headers to a response.

param response:

The response object

param limit_info:

Rate limit information

apifrom.middleware.rate_limit._default_key_func(request)

Default function to extract the rate limit key from a request.

param request:

The request object

returns:

The rate limit key

apifrom.middleware.rate_limit._should_limit(request)

Determine if a request should be rate limited.

param request:

The request object

returns:

True if the request should be rate limited, False otherwise

apifrom.middleware.rate_limit.process_request(request)
:async:

Process a request through the rate limit middleware.

param request:

The request object

returns:

The request object

apifrom.middleware.rate_limit.process_response(response)
:async:

Process a response through the rate limit middleware.

param response:

The response object

returns:

The response object

class apifrom.middleware.rate_limit.RateLimiter[source]

Base rate limiter interface.

apifrom.middleware.rate_limit.check_limit(key)
:abstractmethod:

Check if a key has exceeded its rate limit.

param key:

The key to check

returns:

A tuple containing (allowed, limit_info)

apifrom.middleware.rate_limit.update(key)
:abstractmethod:

Update the rate limit counter for a key.

param key:

The key to update

class apifrom.middleware.rate_limit.Request(request=None, path_params=None, method=None, path=None, query_params=None, headers=None, body=None, client_ip=None)[source]
Parameters:

  • request (Optional[starlette.requests.Request])

  • path_params (Optional[dict[Any, Any]])

  • method (Optional[str])

  • path (Optional[str])

  • query_params (Optional[dict[Any, Any]])

  • headers (Optional[dict[Any, Any]])

  • body (Optional[Union[str, bytes]])

  • client_ip (Optional[str])

Request class for APIFromAnything.

This class wraps a Starlette request and provides methods for accessing request data in a convenient way.

apifrom.middleware.rate_limit._request

The underlying Starlette request.

apifrom.middleware.rate_limit.path_params

Path parameters extracted from the URL.

apifrom.middleware.rate_limit.query_params

Query parameters extracted from the URL.

apifrom.middleware.rate_limit.headers

HTTP headers.

apifrom.middleware.rate_limit.method

HTTP method.

apifrom.middleware.rate_limit.path

Request path.

apifrom.middleware.rate_limit._body

Cached request body.

Initialize a new Request instance.

param request:

The underlying Starlette request.

param path_params:

Path parameters extracted from the URL.

param method:

The HTTP method.

param path:

The request path.

param query_params:

Query parameters.

param headers:

HTTP headers.

param body:

Request body.

param client_ip:

Client IP address.

class apifrom.middleware.rate_limit.Response(content=None, status_code=200, headers=None, content_type='application/json')[source]
Parameters:

  • content (Any)

  • status_code (int)

  • headers (Optional[Dict[str, str]])

  • content_type (str)

Response class for APIFromAnything.

This class represents an HTTP response and provides methods for setting response data, status code, and headers.

apifrom.middleware.rate_limit.content

The response content.

apifrom.middleware.rate_limit.status_code

The HTTP status code.

apifrom.middleware.rate_limit.headers

HTTP headers.

apifrom.middleware.rate_limit.content_type

The content type of the response.

Initialize a new Response instance.

param content:

The response content.

param status_code:

The HTTP status code.

param headers:

HTTP headers.

param content_type:

The content type of the response.

SlidingWindowRateLimiter(limit, window = 60):bases: RateLimiter

Sliding window rate limiter implementation.

This rate limiter uses a sliding time window to limit requests.

Initialize the sliding window rate limiter.

param limit:

Maximum number of requests allowed in the window

param window:

Time window in seconds

apifrom.middleware.rate_limit._clean_old_requests(key)

Remove expired timestamps for a key.

param key:

The key to clean

apifrom.middleware.rate_limit.check_limit(key)

Check if a key has exceeded its rate limit.

param key:

The key to check

returns:

A tuple containing (allowed, limit_info)

apifrom.middleware.rate_limit.update(key)

Update the rate limit counter for a key.

param key:

The key to update

TokenBucketRateLimiter(rate, capacity):bases: RateLimiter

Token bucket rate limiter implementation.

This rate limiter uses a token bucket algorithm to limit requests.

Initialize the token bucket rate limiter.

param rate:

Token refill rate per second

param capacity:

Maximum number of tokens in the bucket

apifrom.middleware.rate_limit._refill(key)

Refill tokens for a key.

param key:

The key to refill

apifrom.middleware.rate_limit.check_limit(key)

Check if a key has exceeded its rate limit.

param key:

The key to check

returns:

A tuple containing (allowed, limit_info)

apifrom.middleware.rate_limit.update(key)

Update the rate limit counter for a key.

param key:

The key to update