apifrom.utils.rate_limit
========================




.. py:currentmodule:: apifrom.utils.rate_limit

Overview
--------

**Classes**

* :py:class:`FixedWindowRateLimiter`
* :py:class:`RateLimiter`
* :py:class:`TokenBucketRateLimiter`






Classes
-------

.. py:class:: 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




.. :: _windows
   :annotation: Dict[str, Dict[int, int]]























.. method:: 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




.. py:class:: RateLimiter(rate, period, burst = None, key_func = None, error_message = None)

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


.. :: _tokens
   :annotation: Dict[str, List[float]]




.. :: burst




.. :: error_message




.. :: key_func




.. :: period




.. :: rate







.. method:: 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



.. method:: get_remaining(request)

   Get the number of remaining requests allowed.

      :param request: The request to check

      :returns: Number of remaining requests allowed
      :rtype: int



.. method:: 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



.. method:: 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




.. py:class:: 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


.. :: _last_refill
   :annotation: Dict[str, float]




.. :: _tokens























.. method:: 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