Rate Limiting

Application Kit provides rate limiting with support for per-project, per-endpoint limits and Redis-backed overrides at both the project and organization level.

Overview

Rate limiting protects your endpoints from abuse by limiting the number of requests a project can make within a time window. Features include:

Per-project, per-endpoint rate limits with configurable time windows
Element-based counting (e.g., limit by matrix elements, not just requests)
Rate limit overrides stored in Redis/Valkey — per-project and per-organization
Monitor mode — count violations without blocking, with Datadog span tagging
Programmatic API for runtime-determined limits (apply_rate_limit)
IETF RateLimit-Policy and RateLimit structured headers (draft-ietf-httpapi-ratelimit-headers-10)

flowchart LR
    Request([Request]) --> Auth[Authentication]
    Auth --> RL{Rate Limit<br/>Check}
    RL -->|Under limit| Handler[Request Handler]
    RL -->|Over limit| Mode{Mode?}
    Mode -->|on| Reject[429 Too Many Requests]
    Mode -->|monitor| Tag[Tag Datadog Span] --> Handler
    Handler --> Response([Response + Headers])
    Reject --> Headers([Response + Headers])

Authentication must come first

Rate limiting requires authentication to identify the project. Ensure authentication is applied before rate limiting in your decorator/dependency chain.

Usage

FastAPIDjangoDjango Ninja (legacy)

Use ProjectRateLimiter or PathRateLimiter as dependencies:

"""FastAPI rate limiting example."""

from fastapi import APIRouter, Depends

from application_kit.fastapi.ratelimit import PathRateLimiter, ProjectRateLimiter
from application_kit.fastapi.security import AuthenticateKey

# Use standard APIRouter - rate limit docs are auto-injected when included in app
router = APIRouter()


# Rate limit per project (requires authentication)
@router.get(
    "/datasets/{dataset_id}/search",
    dependencies=[
        Depends(AuthenticateKey()),  # Authentication first
        Depends(ProjectRateLimiter(max_requests=100, expiry=60)),  # Then rate limit
    ],
)
async def search_dataset(dataset_id: int) -> dict[str, str]:
    """Search within a dataset.

    Rate limit info is automatically added to OpenAPI when using get_fastapi_app().
    """
    return {"status": "ok"}


# Rate limit by path only (no authentication required)
@router.get(
    "/public/endpoint",
    dependencies=[Depends(PathRateLimiter(max_requests=100, expiry=60))],
)
async def public_endpoint() -> dict[str, str]:
    """Public endpoint with path-based rate limiting."""
    return {"status": "ok"}

Use the @rate_limit decorator after authentication:

"""Django rate limiting example."""

from django.http import HttpRequest, JsonResponse

from application_kit.django.decorators import authenticate_key
from application_kit.django.ratelimit import rate_limit


@authenticate_key()  # Authentication first (outermost)
@rate_limit(max_requests=100, expiry=60)  # Then rate limit
def my_view(request: HttpRequest) -> JsonResponse:
    return JsonResponse({"status": "ok"})


@authenticate_key()
@rate_limit(max_requests=100, expiry=60)
async def async_view(request: HttpRequest) -> JsonResponse:
    return JsonResponse({"status": "ok"})

Use the Django @rate_limit decorator with Shinobi:

"""Django Ninja rate limiting example."""

from django.http import HttpRequest

from application_kit.django.decorators import authenticate_key
from application_kit.django.ratelimit import rate_limit
from application_kit.shinobi.api import WoosmapApi
from application_kit.shinobi.authentication import PublicKeyAuth
from application_kit.shinobi.decorators import chain, terminate

api = WoosmapApi(description="My API")


@api.get("/search", auth=[PublicKeyAuth()])
@chain(authenticate_key(), rate_limit(max_requests=100, expiry=60))
@terminate()
def search(request: HttpRequest) -> dict[str, str]:
    return {"status": "ok"}

Choosing a Rate Limiter

ProjectRateLimiter (Recommended)

Use ProjectRateLimiter for most endpoints. It creates rate limit keys based on:

Endpoint path pattern
Organization ID
Project ID

This ensures each project has its own rate limit counter, and supports per-project overrides.

PathRateLimiter (Limited Use Cases)

Use PathRateLimiter only when you don't have access to organization/project IDs. This applies to:

Webhook endpoints where the authentication key doesn't provide project context (e.g., dataset reimport webhooks)
Public endpoints that don't require authentication

PathRateLimiter limitations

PathRateLimiter creates a rate limit counter per actual request path. For example, /users/123 and /users/456 have separate counters. This means:

Each unique URL path gets its own counter
No per-project isolation
Rate limit overrides are not supported
Monitor mode is not supported (treated as disabled)

Only use this when ProjectRateLimiter isn't possible.

Parameters

These apply to ProjectRateLimiter, PathRateLimiter (FastAPI), and the @rate_limit decorator (Django):

Parameter	Type	Default	Description
`max_requests`	`int`	required	Maximum requests allowed in the time window
`expiry`	`int`	`1`	Time window in seconds
`policy_name`	`str`	`"requests"`	IETF policy name for structured headers
`endpoint_name`	`str`	auto-detected	Identifier for the endpoint (Django only)
`redis_dependency`	`str`	`"ratelimit"`	Name of the Redis dependency (Django only)

For the programmatic API (apply_rate_limit), see Programmatic Rate Limiting.

Response Headers

All rate-limited responses include IETF structured headers per draft-ietf-httpapi-ratelimit-headers-10:

Header	Description
`RateLimit-Policy`	Declared policies with quota and window — e.g. `"requests";q=100;w=60`
`RateLimit`	Current status with remaining quota and time — e.g. `"requests";r=82;t=45`

Each rate limiter appends its policy as a comma-separated list member, so endpoints with multiple limiters (e.g. request + element counting) show all policies in a single header:

RateLimit-Policy: "requests";q=100;w=60, "elements";q=5000;w=60
RateLimit: "requests";r=82;t=45, "elements";r=4200;t=30

Policy names are auto-derived from the key suffix ("" → "requests", "/elements" → "elements"), or can be set explicitly via the policy_name parameter.

When a 429 is returned, the response carries all accumulated policy headers from prior limiters — not just the one that triggered the rejection.

Legacy Headers

Legacy RateLimit-Limit, RateLimit-Remaining, and RateLimit-Reset headers are automatically injected alongside the IETF structured headers for backward compatibility:

FastAPI: get_fastapi_app() includes LegacyRateLimitHeaderMiddleware by default. Pass use_x_ratelimit_prefix=True to emit X-RateLimit-* headers instead.
Django: ApplicationKitApiMiddleware injects legacy headers via process_response. Set RATELIMIT_USE_X_PREFIX = True in Django settings to emit X-RateLimit-* headers instead.

The legacy headers are derived from the last policy in the structured header, matching the old behavior where each limiter overwrote previous values.

Pending removal

LegacyRateLimitHeaderMiddleware and the Django process_response legacy header injection are pending removal. Migrate consumers to the IETF structured headers (RateLimit-Policy, RateLimit).

Rate Limit Exceeded

When the rate limit is exceeded, a 429 Too Many Requests response is returned with RateLimit-Policy and RateLimit headers.

FastAPIDjangoDjango Ninja (legacy)

FastAPI returns an HTTPException with status code 429 and the message "Rate limit exceeded. Try again later."

When using ApplicationKitApiMiddleware, the RateLimitExceeded exception is automatically handled and returns a JSON response:

{"detail": "Rate limit exceeded. Try again later."}

When using WoosmapApi, the RateLimitExceeded exception is automatically handled and returns a JSON response:

{"detail": "Rate limit exceeded. Try again later."}

Rate Limit Overrides

Override default rate limits without code changes. Overrides are stored in Redis and checked atomically with rate limiting. Two levels are supported:

Project overrides — per-project, per-endpoint limits
Organization overrides — per-org, per-endpoint limits (fallback when no project override exists)

The priority chain is: project override → org override → defaults.

Only works with ProjectRateLimiter

Overrides only apply to endpoints using ProjectRateLimiter. Custom rate limiters (like IP-based limiters) do not support overrides.

Internal use only

The override management functions are provided for use by dedicated provisioning services. Individual applications should not manage overrides directly—this responsibility belongs to a centralized service that handles rate limit configuration across all projects.

Project Override Functions

"""Rate limit override management examples."""

import redis.asyncio

from application_kit.fastapi.ratelimit import (
    clear_project_rate_limit_overrides,
    delete_rate_limit_override,
    get_rate_limit_override,
    list_rate_limit_overrides,
    set_rate_limit_override,
)


async def grant_higher_limit(redis: "redis.asyncio.Redis[bytes]", organization_id: int, project_id: int) -> None:
    await set_rate_limit_override(
        redis,
        organization_id=organization_id,
        project_id=project_id,
        endpoint_pattern="/datasets/{dataset_id}/search",  # FastAPI format
        max_requests=1000,
        expiry=60,
        override_ttl=86400,  # Optional: auto-expire after 24 hours
    )


async def manage_overrides(redis: "redis.asyncio.Redis[bytes]") -> None:
    # Get a specific override
    override = await get_rate_limit_override(redis, organization_id=1, project_id=2, endpoint_pattern="/api/search")
    if override:
        print(f"Limit: {override.max_requests}, Window: {override.expiry}s")

    # Delete a specific override
    deleted = await delete_rate_limit_override(redis, organization_id=1, project_id=2, endpoint_pattern="/api/search")
    print(f"Deleted: {deleted}")

    # List all overrides for a project
    overrides = await list_rate_limit_overrides(redis, organization_id=1, project_id=2)
    for endpoint, ovr in overrides:
        print(f"{endpoint}: {ovr.max_requests} req/{ovr.expiry}s")

    # Clear all overrides for a project
    count = await clear_project_rate_limit_overrides(redis, organization_id=1, project_id=2)
    print(f"Cleared {count} overrides")

Organization Override Functions

Org overrides set limits for all projects in an organization at once. Each project still gets its own counter — the org override only defines what the limit is, not a shared bucket. For example, if an org override sets 100 req/min, project A and project B each get 100 req/min independently.

"""Organization-level rate limit override management examples."""

import redis.asyncio

from application_kit.fastapi.ratelimit import (
    clear_org_rate_limit_overrides,
    delete_org_rate_limit_override,
    get_org_rate_limit_override,
    list_org_rate_limit_overrides,
    set_org_rate_limit_override,
)


async def grant_org_higher_limit(redis: "redis.asyncio.Redis[bytes]", organization_id: int) -> None:
    """Set a rate limit override for all projects in an organization."""
    await set_org_rate_limit_override(
        redis,
        organization_id=organization_id,
        endpoint_pattern="/datasets/{dataset_id}/search",  # FastAPI format
        max_requests=500,
        expiry=60,
        override_ttl=86400,  # Optional: auto-expire after 24 hours
    )


async def manage_org_overrides(redis: "redis.asyncio.Redis[bytes]") -> None:
    # Get a specific org override
    override = await get_org_rate_limit_override(redis, organization_id=1, endpoint_pattern="/api/search")
    if override:
        print(f"Limit: {override.max_requests}, Window: {override.expiry}s")

    # Delete a specific org override
    deleted = await delete_org_rate_limit_override(redis, organization_id=1, endpoint_pattern="/api/search")
    print(f"Deleted: {deleted}")

    # List all overrides for an organization
    overrides = await list_org_rate_limit_overrides(redis, organization_id=1)
    for endpoint, ovr in overrides:
        print(f"{endpoint}: {ovr.max_requests} req/{ovr.expiry}s")

    # Clear all overrides for an organization
    count = await clear_org_rate_limit_overrides(redis, organization_id=1)
    print(f"Cleared {count} overrides")

Note

The endpoint_pattern format depends on your framework:

FastAPI: /datasets/{dataset_id}/search
Django: /api/v1/datasets/<int:dataset_id>/search

Temporary Overrides

Use the override_ttl parameter to create overrides that automatically expire. This works for both project and org overrides:

"""Temporary rate limit override examples."""

import redis.asyncio

from application_kit.ratelimit_overrides import set_org_rate_limit_override, set_rate_limit_override


async def temporary_project_override(redis: "redis.asyncio.Redis[bytes]") -> None:
    await set_rate_limit_override(
        redis,
        organization_id=1,
        project_id=2,
        endpoint_pattern="/api/v1/search",
        max_requests=1000,
        expiry=60,
        override_ttl=3600,  # Override expires after 1 hour
    )


async def temporary_org_override(redis: "redis.asyncio.Redis[bytes]") -> None:
    await set_org_rate_limit_override(
        redis,
        organization_id=1,
        endpoint_pattern="/api/v1/search",
        max_requests=500,
        expiry=60,
        override_ttl=86400,  # Override expires after 24 hours
    )

Use cases for temporary overrides:

Promotions: Increase limits during marketing events
Trial periods: Grant premium rate limits temporarily
Incident response: Temporarily raise limits while investigating issues

Programmatic Rate Limiting (FastAPI)

Use apply_rate_limit when rate limit parameters aren't known at import time — e.g., limits that vary by request body, or element-based counting where the increment depends on the payload size.

"""FastAPI programmatic rate limiting example."""

from typing import Annotated

import redis.asyncio
from fastapi import APIRouter, Depends
from starlette.requests import Request
from starlette.responses import Response

from application_kit.fastapi.ratelimit import apply_element_rate_limit, apply_rate_limit
from application_kit.fastapi.redis import RedisDependency
from application_kit.fastapi.security import AuthenticateKey

router = APIRouter(dependencies=[Depends(AuthenticateKey())])


@router.post("/geocode/batch")
async def batch_geocode(
    request: Request,
    response: Response,
    ratelimit_redis: Annotated["redis.asyncio.Redis[bytes]", Depends(RedisDependency("ratelimit"))],
) -> dict[str, str]:
    addresses = await request.json()

    # Standard rate limit: 100 requests per minute
    await apply_rate_limit(request, response, max_requests=100, expiry=60, redis_client=ratelimit_redis)

    # Element-based rate limit: 1000 addresses per minute (separate counter)
    await apply_element_rate_limit(
        request, response, max_requests=1000, expiry=60, increment_amount=len(addresses), redis_client=ratelimit_redis
    )

    return {"status": "ok"}

Element-Based Counting

apply_element_rate_limit is a convenience wrapper for counting elements instead of requests. For example, a geocoding batch endpoint might allow 1000 addresses per minute regardless of how many requests that takes.

The Lua script checks current + increment_amount <= max_requests before incrementing, so a request that would cross the boundary is blocked — no overshoot. For example, at 95/100 with increment_amount=10, the request is rejected because 95 + 10 > 100.

It uses key_suffix="/elements" by default to keep element counters separate from request counters on the same endpoint. You can use a custom key_suffix on apply_rate_limit to create additional independent counters (e.g., key_suffix="/heavy" for expensive operations).

Parameters

Parameter	Type	Default	Description
`request`	`Request`	required	The incoming FastAPI request
`response`	`Response`	required	The FastAPI response (headers are set on this)
`max_requests`	`int`	required	Maximum count allowed in the time window
`expiry`	`int`	required	Time window in seconds
`redis_client`	`Redis`	required	Async Redis/Valkey connection
`increment_amount`	`int`	`1`	Amount to increment per call (>1 for element-based)
`key_suffix`	`str`	`""`	Suffix for separate counters on the same endpoint
`policy_name`	`str \\| None`	`None`	IETF policy name. When `None`, auto-derived from `key_suffix`

Custom Rate Limiters (FastAPI)

For advanced use cases (e.g., rate limiting by IP instead of project), subclass RateLimiter and implement make_key():

"""Custom rate limiter example — rate limit by client IP address."""

from fastapi import APIRouter, Depends
from starlette.requests import Request

from application_kit.authenticator.utils import extract_user_ip_from_headers
from application_kit.fastapi.ratelimit import RateLimiter, is_rate_limit_disabled
from application_kit.ratelimit import make_path_rate_limit_key


class IpRateLimiter(RateLimiter):
    """Rate limit by client IP address instead of project."""

    def make_key(self, request: Request) -> str | None:
        if is_rate_limit_disabled():
            return None
        ip = extract_user_ip_from_headers(request.headers)
        if ip is None:
            return None
        return make_path_rate_limit_key(f"{request.url.path}:{ip}")


router = APIRouter()


@router.get(
    "/public-endpoint",
    dependencies=[Depends(IpRateLimiter(max_requests=50, expiry=1))],
)
async def public_endpoint() -> dict[str, str]:
    return {"status": "ok"}

Return None to skip rate limiting

If make_key() returns None, rate limiting is skipped for that request. This is useful for disabling rate limits in development or for certain request types.

Overrides not supported for custom rate limiters

Rate limit overrides (via set_rate_limit_override()) only work with ProjectRateLimiter. Custom rate limiters like IpRateLimiter cannot be overridden because they don't use project-based keys.

Automatic Endpoint Detection

FastAPIDjango

ProjectRateLimiter uses request.scope['route'].path automatically:

# Route: /datasets/{dataset_id}/search
# Rate limit key: ratelimit:/datasets/{dataset_id}/search:{org_id}:{project_id}

When endpoint_name is not provided, the decorator extracts the route pattern from Django's resolver_match:

"""Django automatic endpoint detection example."""

from django.http import HttpRequest, JsonResponse

from application_kit.django.decorators import authenticate_key
from application_kit.django.ratelimit import rate_limit

# URL pattern: path('api/v1/datasets/<int:dataset_id>/search', search_view)


@authenticate_key()
@rate_limit(max_requests=100, expiry=60)  # Auto-detects: "/api/v1/datasets/<int:dataset_id>/search"
def search_view(request: HttpRequest, dataset_id: int) -> JsonResponse:
    return JsonResponse({"results": []})

Implementation Details

Rate limiting uses Redis with Lua scripts for atomic operations:

Atomic counting: Increment and check happen in a single Redis operation
Override lookup: Override configuration is fetched and applied atomically
TTL management: Rate limit windows are automatically managed via Redis TTL

flowchart TD
    Request([Request]) --> A
    subgraph LuaScript ["Lua Script (Atomic)"]
        A[Check Project Override] --> B{Project override<br/>exists?}
        B -->|Yes| C[Use project override limits]
        B -->|No| D{Org override<br/>exists?}
        D -->|Yes| C2[Use org override limits]
        D -->|No| E2[Use default limits]
        C --> F[Get current count]
        C2 --> F
        E2 --> F
        F --> G{Count >= Limit?}
        G -->|No| H[Increment counter]
        G -->|Yes| I[Over limit]
        H --> J{First request?}
        J -->|Yes| K[Set TTL on key]
        J -->|No| L[Return count]
        K --> L
    end
    subgraph Responses [" "]
        Response429([429 + Headers])
        Response200([200 + Headers])
    end
    I --> Response429
    L --> Response200

Redis key format:

Rate limit counter: ratelimit:{endpoint}:{organization_id}:{project_id}
Project override: ratelimit_override:{endpoint}:{organization_id}:{project_id}
Org override: ratelimit_org_override:{endpoint}:{organization_id}

Configuration

Rate Limit Mode

Rate limiting behavior is controlled by the RATE_LIMIT_MODE environment variable:

Mode	Django	FastAPI (`ProjectRateLimiter`)	FastAPI (`PathRateLimiter`)	Description
`on`	Yes	Yes	Yes	Rate limiting is enabled and enforced (default)
`off`	Yes	Yes	Yes	Rate limiting is completely disabled
`monitor`	Yes	Yes	Disabled	Requests are counted but not blocked; tags Datadog spans

Set the mode via environment variable:

export RATE_LIMIT_MODE=on  # or "off" or "monitor"

Monitor mode support

Monitor mode is supported by Django and FastAPI's ProjectRateLimiter. The PathRateLimiter and custom rate limiters using the base RateLimiter class do not support monitor mode — when set to monitor, rate limiting is disabled (same as off). To support monitor mode in custom rate limiters, override __call__ and check get_rate_limit_mode() directly.

Redis Dependency

Add the ratelimit Redis dependency to your application.json:

{
  "dependencies": {
    "databases": [
      {"name": "ratelimit", "type": "redis"}
    ]
  }
}

OpenAPI Documentation

FastAPIDjango

When using get_fastapi_app(), rate limit metadata is automatically added to the OpenAPI schema when you include routers:

Machine-readable: x-ratelimit-limit and x-ratelimit-window-seconds fields
Human-readable: "Ratelimit: X/Ys" appended to the endpoint description

from fastapi import APIRouter, Depends
from application_kit.fastapi.fastapi import get_fastapi_app
from application_kit.fastapi.ratelimit import ProjectRateLimiter

app = get_fastapi_app("myapi")
router = APIRouter()

@router.get("/search", dependencies=[Depends(ProjectRateLimiter(max_requests=100, expiry=60))])
async def search():
    """Search for items."""  # Becomes "Search for items.\n\n**Ratelimit:** 100/60s"
    return {"status": "ok"}

app.include_router(router)  # Rate limit docs auto-injected here

The generated OpenAPI will include:

{
  "paths": {
    "/search": {
      "get": {
        "description": "Search for items.\n\n**Ratelimit:** 100/60s",
        "x-ratelimit-limit": 100,
        "x-ratelimit-window-seconds": 60
      }
    }
  }
}

Django does not automatically generate OpenAPI schemas. Document rate limits in your API documentation manually.

API Reference

FastAPIDjango

application_kit.fastapi.ratelimit

set_rate_limit_override `module-attribute`

set_rate_limit_override = _warn(set_rate_limit_override)

get_rate_limit_override `module-attribute`

get_rate_limit_override = _warn(get_rate_limit_override)

delete_rate_limit_override `module-attribute`

delete_rate_limit_override = _warn(
    delete_rate_limit_override
)

list_rate_limit_overrides `module-attribute`

list_rate_limit_overrides = _warn(list_rate_limit_overrides)

clear_project_rate_limit_overrides `module-attribute`

clear_project_rate_limit_overrides = _warn(
    clear_project_rate_limit_overrides
)

set_org_rate_limit_override `module-attribute`

set_org_rate_limit_override = _warn(
    set_org_rate_limit_override
)

get_org_rate_limit_override `module-attribute`

get_org_rate_limit_override = _warn(
    get_org_rate_limit_override
)

delete_org_rate_limit_override `module-attribute`

delete_org_rate_limit_override = _warn(
    delete_org_rate_limit_override
)

list_org_rate_limit_overrides `module-attribute`

list_org_rate_limit_overrides = _warn(
    list_org_rate_limit_overrides
)

clear_org_rate_limit_overrides `module-attribute`

clear_org_rate_limit_overrides = _warn(
    clear_org_rate_limit_overrides
)

RateLimiter `dataclass`

RateLimiter(max_requests, expiry=1, policy_name='requests')

Base rate limiter class for FastAPI.

Subclass this and implement make_key() to create custom rate limiters. Return None from make_key() to skip rate limiting for that request.

ATTRIBUTE	DESCRIPTION
`max_requests`	Maximum requests allowed within the time window (must be >= 1). TYPE: `int`
`expiry`	Time window in seconds (default: 1 second, must be >= 1). TYPE: `int`

Note

The base __call__ does not support "monitor" mode. When the rate limit mode is set to "monitor", is_rate_limit_disabled() returns True and make_key() should return None, effectively disabling rate limiting. To support monitor mode (count but don't block), override __call__ and check get_rate_limit_mode() directly. See ProjectRateLimiter for an example.

Example

class IpRateLimiter(RateLimiter):
    def make_key(self, request: Request) -> str | None:
        if is_rate_limit_disabled():
            return None
        ip = extract_user_ip_from_headers(request.headers)
        if ip is None:
            return None
        return make_path_rate_limit_key(f"{request.url.path}:{ip}")

RAISES	DESCRIPTION
`ValueError`	If max_requests or expiry is less than 1.

make_key `abstractmethod`

make_key(request)

Build the Redis key for rate limiting.

PARAMETER	DESCRIPTION
`request`	The incoming Starlette/FastAPI request. TYPE: `Request`

RETURNS	DESCRIPTION
`str \| None`	Redis key string for the rate limit counter, or `None` to skip rate limiting.

Source code in application_kit/fastapi/ratelimit.py

@abc.abstractmethod
def make_key(self, request: Request) -> str | None:
    """Build the Redis key for rate limiting.

    Args:
        request: The incoming Starlette/FastAPI request.

    Returns:
        Redis key string for the rate limit counter, or `None` to skip rate limiting.
    """

PathRateLimiter `dataclass`

PathRateLimiter(
    max_requests, expiry=1, policy_name="requests"
)

Bases: RateLimiter

Rate limit by actual request path (no project isolation).

Each unique path gets its own rate limit counter. For example, /users/123 and /users/456 have separate counters.

Useful for public endpoints that don't require authentication where you want to limit by the specific resource being accessed.

Note

Does not support rate limit overrides.
Does not support "monitor" mode (treated as disabled).

ProjectRateLimiter `dataclass`

ProjectRateLimiter(
    max_requests, expiry=1, policy_name="requests"
)

Bases: RateLimiter

Applies rate limits on the endpoint, for a project.

Supports per-project, per-endpoint rate limit overrides stored in Redis. Use set_rate_limit_override() to configure overrides.

The override lookup and rate limiting are performed atomically in a single Lua script for efficiency.

make_key

make_key(request)

Generate the Redis key for rate limiting.

Source code in application_kit/fastapi/ratelimit.py

def make_key(self, request: Request) -> str | None:
    """Generate the Redis key for rate limiting."""
    result = self._get_ref_and_path(request)
    if result is None:
        return None
    ref, path = result
    return make_rate_limit_key(path, ref.organization_id, ref.project_id)

make_override_key

make_override_key(request)

Generate the Redis key for looking up rate limit overrides.

Source code in application_kit/fastapi/ratelimit.py

def make_override_key(self, request: Request) -> str | None:
    """Generate the Redis key for looking up rate limit overrides."""
    result = self._get_ref_and_path(request)
    if result is None:
        return None
    ref, path = result
    return make_override_key(path, ref.organization_id, ref.project_id)

make_org_override_key

make_org_override_key(request)

Generate the Redis key for looking up organization-level rate limit overrides.

Source code in application_kit/fastapi/ratelimit.py

def make_org_override_key(self, request: Request) -> str | None:
    """Generate the Redis key for looking up organization-level rate limit overrides."""
    result = self._get_ref_and_path(request)
    if result is None:
        return None
    ref, path = result
    return make_org_override_key(path, ref.organization_id)

RateLimitOverride

Bases: BaseModel

Rate limit override configuration stored in Redis.

is_rate_limit_disabled

is_rate_limit_disabled()

Check if rate limiting should be skipped for rate limiters that don't support monitor mode.

Returns True when RATE_LIMIT_MODE is "off" or "monitor". Rate limiters that don't implement monitor mode (e.g., PathRateLimiter, custom RateLimiter subclasses) should treat monitor mode as disabled — silently skipping rather than incorrectly blocking requests.

Only ProjectRateLimiter and apply_rate_limit support monitor mode (they check get_rate_limit_mode() directly and tag Datadog spans instead of blocking).

Example

class IpRateLimiter(RateLimiter):
    def make_key(self, request: Request) -> str | None:
        if is_rate_limit_disabled():
            return None
        ip = request.headers.get("X-Forwarded-For", "unknown")
        return make_path_rate_limit_key(f"{request.url.path}:{ip}")

Source code in application_kit/ratelimit.py

def is_rate_limit_disabled() -> bool:
    """Check if rate limiting should be skipped for rate limiters that don't support monitor mode.

    Returns True when RATE_LIMIT_MODE is "off" or "monitor". Rate limiters that
    don't implement monitor mode (e.g., `PathRateLimiter`, custom `RateLimiter`
    subclasses) should treat monitor mode as disabled — silently skipping rather
    than incorrectly blocking requests.

    Only `ProjectRateLimiter` and `apply_rate_limit` support monitor mode (they
    check `get_rate_limit_mode()` directly and tag Datadog spans instead of blocking).

    Example:
        ```python
        class IpRateLimiter(RateLimiter):
            def make_key(self, request: Request) -> str | None:
                if is_rate_limit_disabled():
                    return None
                ip = request.headers.get("X-Forwarded-For", "unknown")
                return make_path_rate_limit_key(f"{request.url.path}:{ip}")
        ```
    """
    return get_rate_limit_mode() != RateLimitMode.on

apply_rate_limit `async`

apply_rate_limit(
    request,
    response,
    max_requests,
    expiry,
    redis_client,
    increment_amount=1,
    key_suffix="",
    policy_name=None,
)

Apply rate limiting programmatically (not via Depends).

Use this when rate limit parameters depend on runtime logic, such as different limits based on request content.

PARAMETER	DESCRIPTION
`request`	The incoming FastAPI request. TYPE: `Request`
`response`	The FastAPI response to update headers on. TYPE: `Response`
`max_requests`	Maximum requests allowed in the time window (must be >= 1). TYPE: `int`
`expiry`	Time window in seconds (must be >= 1). TYPE: `int`
`redis_client`	Async Redis connection for rate limiting. TYPE: `Redis[bytes]`
`increment_amount`	Amount to increment the counter by (default: 1, must be >= 1). Use values > 1 for element-based rate limiting. TYPE: `int` DEFAULT: `1`
`key_suffix`	Optional suffix to append to the rate limit key. Use to separate different rate limit counters for the same endpoint. TYPE: `str` DEFAULT: `''`
`policy_name`	IETF policy name for structured headers. When None, auto-derived from key_suffix (e.g. "" → "requests"). TYPE: `str \| None` DEFAULT: `None`

RAISES	DESCRIPTION
`ValueError`	If max_requests, expiry, or increment_amount is less than 1.
`HTTPException`	429 status when over limit.

Source code in application_kit/fastapi/ratelimit.py

@tracer.wrap()
async def apply_rate_limit(
    request: Request,
    response: Response,
    max_requests: int,
    expiry: int,
    redis_client: redis.asyncio.Redis[bytes],
    increment_amount: int = 1,
    key_suffix: str = "",
    policy_name: str | None = None,
) -> None:
    """Apply rate limiting programmatically (not via Depends).

    Use this when rate limit parameters depend on runtime logic,
    such as different limits based on request content.

    Args:
        request: The incoming FastAPI request.
        response: The FastAPI response to update headers on.
        max_requests: Maximum requests allowed in the time window (must be >= 1).
        expiry: Time window in seconds (must be >= 1).
        redis_client: Async Redis connection for rate limiting.
        increment_amount: Amount to increment the counter by (default: 1, must be >= 1).
            Use values > 1 for element-based rate limiting.
        key_suffix: Optional suffix to append to the rate limit key.
            Use to separate different rate limit counters for the same endpoint.
        policy_name: IETF policy name for structured headers. When None,
            auto-derived from key_suffix (e.g. "" → "requests").

    Raises:
        ValueError: If max_requests, expiry, or increment_amount is less than 1.
        HTTPException: 429 status when over limit.
    """
    if max_requests < 1:
        raise ValueError("max_requests must be at least 1")
    if expiry < 1:
        raise ValueError("expiry must be at least 1 second")
    if increment_amount < 1:
        raise ValueError("increment_amount must be at least 1")

    mode = get_rate_limit_mode()
    if mode == RateLimitMode.off:
        return

    ref_and_path = _extract_ref_and_path(request)
    if ref_and_path is None:
        return
    ref, path = ref_and_path

    effective_policy_name = policy_name if policy_name is not None else derive_policy_name(key_suffix)
    key_path = f"{path}{key_suffix}" if key_suffix else path

    # Generate keys
    key = make_rate_limit_key(key_path, ref.organization_id, ref.project_id)
    override_key = make_override_key(key_path, ref.organization_id, ref.project_id)
    org_key = make_org_override_key(key_path, ref.organization_id)

    # Execute Lua script with increment_amount
    script = redis_client.register_script(PROJECT_RATE_LIMITER_LUA)
    raw = await script([key, override_key, org_key], [max_requests, expiry, increment_amount])
    result = RateLimitResult.from_lua_result(raw, policy_name=effective_policy_name)

    _handle_rate_limit_result(result, response, path, mode)

apply_element_rate_limit `async`

apply_element_rate_limit(
    request,
    response,
    max_requests,
    expiry,
    increment_amount,
    redis_client,
    key_suffix="/elements",
    policy_name=None,
)

Apply element-based rate limiting.

Convenience wrapper for apply_rate_limit with element-specific defaults. Used for endpoints where rate limiting is based on the number of elements (e.g., matrix origins × destinations) rather than just request count.

PARAMETER	DESCRIPTION
`request`	The incoming FastAPI request. TYPE: `Request`
`response`	The FastAPI response to update headers on. TYPE: `Response`
`max_requests`	Maximum elements allowed in the time window. TYPE: `int`
`expiry`	Time window in seconds. TYPE: `int`
`increment_amount`	Number of elements in this request. TYPE: `int`
`redis_client`	Async Redis connection for rate limiting. TYPE: `Redis[bytes]`
`key_suffix`	Suffix to append to the rate limit key (default: "/elements"). Separates element limits from request count limits. TYPE: `str` DEFAULT: `'/elements'`
`policy_name`	IETF policy name for structured headers. When None, auto-derived from key_suffix (e.g. "/elements" → "elements"). TYPE: `str \| None` DEFAULT: `None`

RAISES	DESCRIPTION
`HTTPException`	429 status when over limit.

Source code in application_kit/fastapi/ratelimit.py

@tracer.wrap()
async def apply_element_rate_limit(
    request: Request,
    response: Response,
    max_requests: int,
    expiry: int,
    increment_amount: int,
    redis_client: redis.asyncio.Redis[bytes],
    key_suffix: str = "/elements",
    policy_name: str | None = None,
) -> None:
    """Apply element-based rate limiting.

    Convenience wrapper for apply_rate_limit with element-specific defaults.
    Used for endpoints where rate limiting is based on the number of elements
    (e.g., matrix origins × destinations) rather than just request count.

    Args:
        request: The incoming FastAPI request.
        response: The FastAPI response to update headers on.
        max_requests: Maximum elements allowed in the time window.
        expiry: Time window in seconds.
        increment_amount: Number of elements in this request.
        redis_client: Async Redis connection for rate limiting.
        key_suffix: Suffix to append to the rate limit key (default: "/elements").
            Separates element limits from request count limits.
        policy_name: IETF policy name for structured headers. When None,
            auto-derived from key_suffix (e.g. "/elements" → "elements").

    Raises:
        HTTPException: 429 status when over limit.
    """
    effective_policy_name = policy_name if policy_name is not None else derive_policy_name(key_suffix)
    await apply_rate_limit(
        request=request,
        response=response,
        max_requests=max_requests,
        expiry=expiry,
        redis_client=redis_client,
        increment_amount=increment_amount,
        key_suffix=key_suffix,
        policy_name=effective_policy_name,
    )

application_kit.django.ratelimit

RateLimitExceeded

RateLimitExceeded(result)

Bases: HttpException

Raised when rate limit is exceeded.

Source code in application_kit/django/ratelimit.py

def __init__(self, result: RateLimitResult) -> None:
    super().__init__(status=429, message=RATE_LIMIT_ERROR_MESSAGE, headers=result.to_headers())
    self.result = result

rate_limit

rate_limit(
    max_requests,
    expiry=1,
    endpoint_name=None,
    redis_dependency="ratelimit",
    policy_name="requests",
)

Rate limit decorator for Django views.

Must be placed after authenticate_key or authenticate_user decorator to have access to the project token.

PARAMETER	DESCRIPTION
`max_requests`	Maximum requests allowed in the time window (must be >= 1). TYPE: `int`
`expiry`	Time window in seconds (default: 1 second, must be >= 1). TYPE: `int` DEFAULT: `1`
`endpoint_name`	Optional identifier for the endpoint. If not provided, automatically extracted from the request's URL pattern (e.g., '/api/v1/datasets//search'). TYPE: `str \| None` DEFAULT: `None`
`redis_dependency`	Name of the Redis dependency in bender config. TYPE: `str` DEFAULT: `'ratelimit'`
`policy_name`	IETF policy name for structured headers (default: "requests"). TYPE: `str` DEFAULT: `'requests'`

RAISES	DESCRIPTION
`ValueError`	If max_requests or expiry is less than 1.

Example

@authenticate_key() @rate_limit(max_requests=100, expiry=60) def my_view(request): return JsonResponse({"status": "ok"})

Or with explicit endpoint name:

@authenticate_key() @rate_limit(max_requests=100, expiry=60, endpoint_name="custom_name") def my_view(request): return JsonResponse({"status": "ok"})

Rate limit overrides can be configured per project using the async functions from application_kit.fastapi.ratelimit: - set_rate_limit_override() to create/update an override - get_rate_limit_override() to retrieve an override - delete_rate_limit_override() to remove an override - list_rate_limit_overrides() to list all overrides for a project

Override key format: ratelimit_override:{endpoint_name}:{org_id}:{project_id}

Source code in application_kit/django/ratelimit.py

def rate_limit(
    max_requests: int,
    expiry: int = 1,
    endpoint_name: str | None = None,
    redis_dependency: str = "ratelimit",
    policy_name: str = "requests",
) -> Callable[[T], T]:
    """Rate limit decorator for Django views.

    Must be placed after `authenticate_key` or `authenticate_user` decorator
    to have access to the project token.

    Args:
        max_requests: Maximum requests allowed in the time window (must be >= 1).
        expiry: Time window in seconds (default: 1 second, must be >= 1).
        endpoint_name: Optional identifier for the endpoint. If not provided,
            automatically extracted from the request's URL pattern
            (e.g., '/api/v1/datasets/<int:dataset_id>/search').
        redis_dependency: Name of the Redis dependency in bender config.
        policy_name: IETF policy name for structured headers (default: "requests").

    Raises:
        ValueError: If max_requests or expiry is less than 1.

    Example:
        @authenticate_key()
        @rate_limit(max_requests=100, expiry=60)
        def my_view(request):
            return JsonResponse({"status": "ok"})

        # Or with explicit endpoint name:
        @authenticate_key()
        @rate_limit(max_requests=100, expiry=60, endpoint_name="custom_name")
        def my_view(request):
            return JsonResponse({"status": "ok"})

    Rate limit overrides can be configured per project using the async functions
    from `application_kit.fastapi.ratelimit`:
        - `set_rate_limit_override()` to create/update an override
        - `get_rate_limit_override()` to retrieve an override
        - `delete_rate_limit_override()` to remove an override
        - `list_rate_limit_overrides()` to list all overrides for a project

    Override key format: `ratelimit_override:{endpoint_name}:{org_id}:{project_id}`
    """
    if max_requests < 1:
        raise ValueError("max_requests must be at least 1")
    if expiry < 1:
        raise ValueError("expiry must be at least 1 second")

    def decorator(func: T) -> T:
        if is_coroutine_function(func):
            return cast(
                T, _make_async_wrapper(func, endpoint_name, redis_dependency, max_requests, expiry, policy_name)
            )
        elif not_is_coroutine_function(func):
            return cast(T, _make_sync_wrapper(func, endpoint_name, redis_dependency, max_requests, expiry, policy_name))
        else:
            raise RuntimeError("View must be either sync or async")

    return decorator

Rate Limiting

Overview

Usage

Choosing a Rate Limiter

ProjectRateLimiter (Recommended)

PathRateLimiter (Limited Use Cases)

Parameters

Response Headers

Legacy Headers

Rate Limit Exceeded

Rate Limit Overrides

Project Override Functions

Organization Override Functions

Temporary Overrides

Programmatic Rate Limiting (FastAPI)

Element-Based Counting

Parameters

Custom Rate Limiters (FastAPI)

Automatic Endpoint Detection

Implementation Details

Configuration

Rate Limit Mode

Redis Dependency

OpenAPI Documentation

API Reference

application_kit.fastapi.ratelimit

set_rate_limit_override module-attribute

get_rate_limit_override module-attribute

delete_rate_limit_override module-attribute

list_rate_limit_overrides module-attribute

clear_project_rate_limit_overrides module-attribute

set_org_rate_limit_override module-attribute

get_org_rate_limit_override module-attribute

delete_org_rate_limit_override module-attribute

list_org_rate_limit_overrides module-attribute

clear_org_rate_limit_overrides module-attribute

RateLimiter dataclass

make_key abstractmethod

PathRateLimiter dataclass

ProjectRateLimiter dataclass

make_key

make_override_key

make_org_override_key

RateLimitOverride

is_rate_limit_disabled

apply_rate_limit async

apply_element_rate_limit async

application_kit.django.ratelimit

RateLimitExceeded

rate_limit

Or with explicit endpoint name:

set_rate_limit_override `module-attribute`

get_rate_limit_override `module-attribute`

delete_rate_limit_override `module-attribute`

list_rate_limit_overrides `module-attribute`

clear_project_rate_limit_overrides `module-attribute`

set_org_rate_limit_override `module-attribute`

get_org_rate_limit_override `module-attribute`

delete_org_rate_limit_override `module-attribute`

list_org_rate_limit_overrides `module-attribute`

clear_org_rate_limit_overrides `module-attribute`

RateLimiter `dataclass`

make_key `abstractmethod`

PathRateLimiter `dataclass`

ProjectRateLimiter `dataclass`

apply_rate_limit `async`

apply_element_rate_limit `async`