apifrom.middleware.rate_limit ============================= Rate limiting middleware for APIFromAnything. This module provides middleware for rate limiting API requests to prevent abuse. .. py:currentmodule:: apifrom.middleware.rate_limit Overview -------- **Classes** * :py:class:`BaseMiddleware` * :py:class:`FixedWindowRateLimiter` * :py:class:`InMemoryRateLimitBackend` * :py:class:`JSONResponse` * :py:class:`RateLimit` * :py:class:`RateLimitBackend` * :py:class:`RateLimitMiddleware` * :py:class:`RateLimiter` * :py:class:`Request` * :py:class:`Response` * :py:class:`SlidingWindowRateLimiter` * :py:class:`TokenBucketRateLimiter` Classes ------- .. py:class:: 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. .. attribute:: options Options for the middleware. :type: dict Initialize a new BaseMiddleware instance. :param \*\*options: Options for the middleware. .. py:class:: 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 .. :: counters :annotation: Dict[str, Dict[int, int]] .. :: limit .. :: window .. method:: _get_current_window() Get the current time window. :returns: The current time window as an integer .. method:: 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) .. method:: update(key) Update the rate limit counter for a key. :param key: The key to update .. py:class:: InMemoryRateLimitBackend:bases: RateLimitBackend In-memory implementation of RateLimitBackend. This backend stores rate limit data in memory. Initialize a new InMemoryRateLimitBackend instance. .. :: _data .. :: _expiry .. method:: _cleanup() Clean up expired keys. .. method:: 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 .. method:: 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 .. method:: reset(key) Reset rate limit data for a key. :param key: The key to reset .. method:: 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 .. py:class:: 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. .. py:class:: RateLimit Decorator for controlling rate limiting on specific endpoints. .. method:: exempt(func) :staticmethod: Exempt an endpoint from rate limiting. :param func: The function to decorate :returns: The decorated function .. method:: 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 .. py:class:: RateLimitBackend Base class for rate limit backends. Rate limit backends are responsible for storing and retrieving rate limit data. .. method:: get(key) :abstractmethod: Get rate limit data for a key. :param key: The key to get data for :returns: The rate limit data .. method:: 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 .. method:: reset(key) :abstractmethod: Reset rate limit data for a key. :param key: The key to reset .. method:: 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 .. py:class:: 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 .. :: exclude_routes .. :: headers_enabled .. :: key_func .. :: limiter .. method:: __call__(scope, receive, send) :async: ASGI callable. :param scope: The ASGI scope. :param receive: The ASGI receive function. :param send: The ASGI send function. .. method:: _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 .. method:: _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 .. method:: _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 .. method:: process_request(request) :async: Process a request through the rate limit middleware. :param request: The request object :returns: The request object .. method:: process_response(response) :async: Process a response through the rate limit middleware. :param response: The response object :returns: The response object .. py:class:: RateLimiter Base rate limiter interface. .. method:: 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) .. method:: update(key) :abstractmethod: Update the rate limit counter for a key. :param key: The key to update .. py:class:: Request(request = None, path_params = None, method = None, path = None, query_params = None, headers = None, body = None, client_ip = None) Request class for APIFromAnything. This class wraps a Starlette request and provides methods for accessing request data in a convenient way. .. attribute:: _request The underlying Starlette request. .. attribute:: path_params Path parameters extracted from the URL. .. attribute:: query_params Query parameters extracted from the URL. .. attribute:: headers HTTP headers. .. attribute:: method HTTP method. .. attribute:: path Request path. .. attribute:: _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. .. py:class:: Response(content = None, status_code = 200, headers = None, content_type = 'application/json') Response class for APIFromAnything. This class represents an HTTP response and provides methods for setting response data, status code, and headers. .. attribute:: content The response content. .. attribute:: status_code The HTTP status code. .. attribute:: headers HTTP headers. .. attribute:: 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. .. py:class:: 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 .. :: limit .. :: requests :annotation: Dict[str, collections.deque] .. :: window .. method:: _clean_old_requests(key) Remove expired timestamps for a key. :param key: The key to clean .. method:: 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) .. method:: update(key) Update the rate limit counter for a key. :param key: The key to update .. py:class:: 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 .. :: buckets :annotation: Dict[str, Dict[str, float]] .. :: capacity .. :: rate .. method:: _refill(key) Refill tokens for a key. :param key: The key to refill .. method:: 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) .. method:: update(key) Update the rate limit counter for a key. :param key: The key to update