fastapi-traffic/docs/api/decorator.rst

Decorator API
=============

The ``@rate_limit`` decorator is the primary way to add rate limiting to your
FastAPI endpoints.

rate_limit
----------

.. py:function:: rate_limit(limit, window_size=60.0, *, algorithm=Algorithm.SLIDING_WINDOW_COUNTER, key_prefix="ratelimit", key_extractor=default_key_extractor, burst_size=None, include_headers=True, error_message="Rate limit exceeded", status_code=429, skip_on_error=False, cost=1, exempt_when=None, on_blocked=None)

   Apply rate limiting to a FastAPI endpoint.

   :param limit: Maximum number of requests allowed in the window.
   :type limit: int
   :param window_size: Time window in seconds. Defaults to 60.
   :type window_size: float
   :param algorithm: Rate limiting algorithm to use.
   :type algorithm: Algorithm
   :param key_prefix: Prefix for the rate limit key.
   :type key_prefix: str
   :param key_extractor: Function to extract client identifier from request.
   :type key_extractor: Callable[[Request], str]
   :param burst_size: Maximum burst size for token bucket/leaky bucket algorithms.
   :type burst_size: int | None
   :param include_headers: Whether to include rate limit headers in response.
   :type include_headers: bool
   :param error_message: Error message when rate limit is exceeded.
   :type error_message: str
   :param status_code: HTTP status code when rate limit is exceeded.
   :type status_code: int
   :param skip_on_error: Skip rate limiting if backend errors occur.
   :type skip_on_error: bool
   :param cost: Cost of each request (default 1).
   :type cost: int
   :param exempt_when: Function to determine if request should be exempt.
   :type exempt_when: Callable[[Request], bool] | None
   :param on_blocked: Callback when a request is blocked.
   :type on_blocked: Callable[[Request, Any], Any] | None
   :returns: Decorated function with rate limiting applied.
   :rtype: Callable

   **Basic usage:**

   .. code-block:: python

      from fastapi import FastAPI, Request
      from fastapi_traffic import rate_limit

      app = FastAPI()

      @app.get("/api/data")
      @rate_limit(100, 60)  # 100 requests per minute
      async def get_data(request: Request):
          return {"data": "here"}

   **With algorithm:**

   .. code-block:: python

      from fastapi_traffic import rate_limit, Algorithm

      @app.get("/api/burst")
      @rate_limit(100, 60, algorithm=Algorithm.TOKEN_BUCKET, burst_size=20)
      async def burst_endpoint(request: Request):
          return {"status": "ok"}

   **With custom key extractor:**

   .. code-block:: python

      def get_api_key(request: Request) -> str:
          return request.headers.get("X-API-Key", "anonymous")

      @app.get("/api/data")
      @rate_limit(1000, 3600, key_extractor=get_api_key)
      async def api_endpoint(request: Request):
          return {"data": "here"}

   **With exemption:**

   .. code-block:: python

      def is_admin(request: Request) -> bool:
          return getattr(request.state, "is_admin", False)

      @app.get("/api/admin")
      @rate_limit(100, 60, exempt_when=is_admin)
      async def admin_endpoint(request: Request):
          return {"admin": "data"}

RateLimitDependency
-------------------

.. py:class:: RateLimitDependency(limit, window_size=60.0, *, algorithm=Algorithm.SLIDING_WINDOW_COUNTER, key_prefix="ratelimit", key_extractor=default_key_extractor, burst_size=None, error_message="Rate limit exceeded", status_code=429, skip_on_error=False, cost=1, exempt_when=None)
   :no-index:

   FastAPI dependency for rate limiting. Returns rate limit info that can be
   used in your endpoint. See :doc:`dependency` for full documentation.

   :param limit: Maximum number of requests allowed in the window.
   :type limit: int
   :param window_size: Time window in seconds.
   :type window_size: float

   **Usage:**

   .. code-block:: python

      from fastapi import FastAPI, Depends, Request
      from fastapi_traffic.core.decorator import RateLimitDependency

      app = FastAPI()
      rate_dep = RateLimitDependency(limit=100, window_size=60)

      @app.get("/api/data")
      async def get_data(request: Request, rate_info=Depends(rate_dep)):
          return {
              "data": "here",
              "remaining_requests": rate_info.remaining,
              "reset_at": rate_info.reset_at,
          }

   The dependency returns a ``RateLimitInfo`` object with:

   - ``limit``: The configured limit
   - ``remaining``: Remaining requests in the current window
   - ``reset_at``: Unix timestamp when the window resets
   - ``retry_after``: Seconds until retry (if rate limited)

create_rate_limit_response
--------------------------

.. py:function:: create_rate_limit_response(exc, *, include_headers=True)

   Create a standard rate limit response from a RateLimitExceeded exception.

   :param exc: The RateLimitExceeded exception.
   :type exc: RateLimitExceeded
   :param include_headers: Whether to include rate limit headers.
   :type include_headers: bool
   :returns: A JSONResponse with rate limit information.
   :rtype: Response

   **Usage:**

   .. code-block:: python

      from fastapi_traffic import RateLimitExceeded
      from fastapi_traffic.core.decorator import create_rate_limit_response

      @app.exception_handler(RateLimitExceeded)
      async def handler(request: Request, exc: RateLimitExceeded):
          return create_rate_limit_response(exc)