Rate Limiting

The ratelimit module provides shared infrastructure for rate limiting across Django and FastAPI. This includes Lua scripts for atomic Redis operations, data models for rate limit configuration, and key generation utilities.

Lua Script

PROJECT_RATE_LIMITER_LUA handles atomic rate limit checking with optional override support. It checks whether current + increment_amount would exceed the limit before incrementing (using INCRBY for element-based counting), checks for per-project overrides stored in Redis hashes, falls back to organization-level overrides, and returns {is_over_limit, request_count, ttl, effective_max_requests, effective_expiry} in a single atomic operation. The effective values reflect any applied overrides, which is important for accurate IETF w= (window) parameters. The priority chain is: project override → org override → defaults. When no override keys are provided (empty strings), it behaves as a simple rate limiter.

Key Format

Rate limit keys follow consistent patterns:

Counter: ratelimit:{endpoint}:{organization_id}:{project_id}
Project override: ratelimit_override:{endpoint}:{organization_id}:{project_id}
Org override: ratelimit_org_override:{endpoint}:{organization_id}

The endpoint portion matches your framework's route pattern (e.g., /datasets/{dataset_id}/search for FastAPI).

Core

application_kit.ratelimit

Shared rate limiting components for Django and FastAPI.

This module contains the common Lua scripts, data models, and key generation functions used by both the Django and FastAPI rate limiting implementations.

RateLimitOverride

Bases: BaseModel

Rate limit override configuration stored in Redis.

RateLimitResult `dataclass`

RateLimitResult(
    is_over_limit,
    request_count,
    ttl,
    max_requests,
    policy_name="requests",
    window=0,
)

Result of a rate limit check.

Supports IETF draft draft-ietf-httpapi-ratelimit-headers-10 via named policies. Each result carries a policy_name (e.g. "requests", "elements") and an optional window (full window size in seconds). Multiple results can be combined into comma-separated RateLimit-Policy and RateLimit structured headers.

to_policy_item

to_policy_item()

Format this result as an IETF RateLimit-Policy list member.

Returns e.g. '"requests";q=100;w=60' when window > 0, or '"requests";q=100' when window is 0 (omitted).

Source code in application_kit/ratelimit.py

def to_policy_item(self) -> str:
    """Format this result as an IETF ``RateLimit-Policy`` list member.

    Returns e.g. ``'"requests";q=100;w=60'`` when window > 0,
    or ``'"requests";q=100'`` when window is 0 (omitted).
    """
    item = f'"{self.policy_name}";q={self.max_requests}'
    if self.window > 0:
        item += f";w={self.window}"
    return item

to_status_item

to_status_item()

Format this result as an IETF RateLimit list member.

Returns e.g. '"requests";r=82;t=45'.

Source code in application_kit/ratelimit.py

def to_status_item(self) -> str:
    """Format this result as an IETF ``RateLimit`` list member.

    Returns e.g. ``'"requests";r=82;t=45'``.
    """
    return f'"{self.policy_name}";r={self.remaining};t={self.ttl}'

to_headers

to_headers()

Generate IETF rate limit headers for a single policy.

Source code in application_kit/ratelimit.py

def to_headers(self) -> dict[str, str]:
    """Generate IETF rate limit headers for a single policy."""
    return {
        "RateLimit-Policy": self.to_policy_item(),
        "RateLimit": self.to_status_item(),
    }

from_lua_result `classmethod`

from_lua_result(raw, *, policy_name='requests')

Create a RateLimitResult from the Lua script output.

PARAMETER	DESCRIPTION
`raw`	List returned by PROJECT_RATE_LIMITER_LUA: [is_over_limit, request_count, ttl, max_requests, expiry] TYPE: `list[SupportsInt]`
`policy_name`	IETF policy name for structured headers. TYPE: `str` DEFAULT: `'requests'`

RETURNS	DESCRIPTION
`RateLimitResult`	RateLimitResult instance

Source code in application_kit/ratelimit.py

@classmethod
def from_lua_result(
    cls,
    raw: list[SupportsInt],
    *,
    policy_name: str = "requests",
) -> "RateLimitResult":
    """Create a RateLimitResult from the Lua script output.

    Args:
        raw: List returned by PROJECT_RATE_LIMITER_LUA:
             [is_over_limit, request_count, ttl, max_requests, expiry]
        policy_name: IETF policy name for structured headers.

    Returns:
        RateLimitResult instance
    """
    return cls(
        is_over_limit=bool(raw[_LUA_IDX_IS_OVER_LIMIT]),
        request_count=int(raw[_LUA_IDX_REQUEST_COUNT]),
        ttl=int(raw[_LUA_IDX_TTL]),
        max_requests=int(raw[_LUA_IDX_MAX_REQUESTS]),
        policy_name=policy_name,
        window=int(raw[_LUA_IDX_EXPIRY]),
    )

normalize_path

normalize_path(path)

Normalize a path for consistent rate limit key generation.

Ensures the path starts with a leading slash
Strips trailing slashes (except for root "/")

This ensures that "/api/users", "api/users", and "/api/users/" all produce the same rate limit key.

PARAMETER	DESCRIPTION
`path`	The path to normalize (e.g., route pattern or endpoint name) TYPE: `str`

RETURNS	DESCRIPTION
`str`	Normalized path string

Source code in application_kit/ratelimit.py

def normalize_path(path: str) -> str:
    """Normalize a path for consistent rate limit key generation.

    - Ensures the path starts with a leading slash
    - Strips trailing slashes (except for root "/")

    This ensures that "/api/users", "api/users", and "/api/users/" all
    produce the same rate limit key.

    Args:
        path: The path to normalize (e.g., route pattern or endpoint name)

    Returns:
        Normalized path string
    """
    if path and not path.startswith("/"):
        path = f"/{path}"
    return path.rstrip("/") or "/"

derive_policy_name

derive_policy_name(key_suffix)

Derive IETF policy name from the Redis key suffix.

Suffixes ending with "elements" are treated as element-based policies, everything else as request-based:

"" → "requests" "/traffic" → "traffic_requests" "/elements" → "elements" "/traffic/elements" → "traffic_elements"

Source code in application_kit/ratelimit.py

def derive_policy_name(key_suffix: str) -> str:
    """Derive IETF policy name from the Redis key suffix.

    Suffixes ending with "elements" are treated as element-based policies,
    everything else as request-based:

      "" → "requests"
      "/traffic" → "traffic_requests"
      "/elements" → "elements"
      "/traffic/elements" → "traffic_elements"
    """
    # Collapse consecutive slashes, convert to underscores, strip edges
    parts = [p for p in key_suffix.split("/") if p]
    slug = "_".join(parts)
    if not slug:
        return "requests"
    if slug.endswith("elements"):
        return slug
    return f"{slug}_requests"

make_path_rate_limit_key

make_path_rate_limit_key(path)

Generate the Redis key for path-only rate limit counter (no project isolation).

Key format: ratelimit:{path}

PARAMETER	DESCRIPTION
`path`	The request path (should be normalized with `normalize_path` first). TYPE: `str`

RETURNS	DESCRIPTION
`str`	Redis key string for the rate limit counter.

Source code in application_kit/ratelimit.py

def make_path_rate_limit_key(path: str) -> str:
    """Generate the Redis key for path-only rate limit counter (no project isolation).

    Key format: ratelimit:{path}

    Args:
        path: The request path (should be normalized with ``normalize_path`` first).

    Returns:
        Redis key string for the rate limit counter.
    """
    return f"{_RATE_LIMIT_PREFIX}:{path}"

make_rate_limit_key

make_rate_limit_key(
    endpoint_name, organization_id, project_id
)

Generate the Redis key for rate limit counter.

Source code in application_kit/ratelimit.py

def make_rate_limit_key(
    endpoint_name: str,
    organization_id: int | None,
    project_id: int | None,
) -> str:
    """Generate the Redis key for rate limit counter."""
    return _make_key(_RATE_LIMIT_PREFIX, endpoint_name, organization_id, project_id)

make_override_key

make_override_key(
    endpoint_name, organization_id, project_id
)

Generate the Redis key for rate limit override lookup.

The endpoint_name is normalized (leading slash, no trailing slash) to ensure consistent keys regardless of how the caller formats the path.

Source code in application_kit/ratelimit.py

def make_override_key(
    endpoint_name: str,
    organization_id: int | None,
    project_id: int | None,
) -> str:
    """Generate the Redis key for rate limit override lookup.

    The endpoint_name is normalized (leading slash, no trailing slash) to ensure
    consistent keys regardless of how the caller formats the path.
    """
    return _make_key(_OVERRIDE_PREFIX, normalize_path(endpoint_name), organization_id, project_id)

parse_override_key

parse_override_key(key)

Parse an override key to extract endpoint, organization_id, and project_id.

PARAMETER	DESCRIPTION
`key`	Redis key in format "ratelimit_override:{endpoint}:{org_id}:{project_id}" TYPE: `str`

RETURNS	DESCRIPTION
`tuple[str, int, int] \| None`	Tuple of (endpoint_name, organization_id, project_id) or None if parsing fails.

Note

This handles endpoint names that contain colons by parsing from both ends: - The prefix is known and stripped from the left - org_id and project_id are parsed from the right (they're always integers) - Everything in between is the endpoint name

Source code in application_kit/ratelimit.py

def parse_override_key(key: str) -> tuple[str, int, int] | None:
    """Parse an override key to extract endpoint, organization_id, and project_id.

    Args:
        key: Redis key in format "ratelimit_override:{endpoint}:{org_id}:{project_id}"

    Returns:
        Tuple of (endpoint_name, organization_id, project_id) or None if parsing fails.

    Note:
        This handles endpoint names that contain colons by parsing from both ends:
        - The prefix is known and stripped from the left
        - org_id and project_id are parsed from the right (they're always integers)
        - Everything in between is the endpoint name
    """
    result = _parse_key_from_right(key, _OVERRIDE_PREFIX, num_trailing_ints=2)
    if result is None:
        return None
    endpoint, org_id_str, project_id_str = result
    return endpoint, int(org_id_str), int(project_id_str)

make_org_override_key

make_org_override_key(endpoint_name, organization_id)

Generate the Redis key for organization-level rate limit override lookup.

Key format: ratelimit_org_override:{endpoint_name}:{organization_id}

PARAMETER	DESCRIPTION
`endpoint_name`	The endpoint name or route pattern. TYPE: `str`
`organization_id`	The organization ID, or None to skip org override lookup. TYPE: `int \| None`

RETURNS	DESCRIPTION
`str`	Redis key string for the org override hash, or empty string if organization_id is None.

Source code in application_kit/ratelimit.py

def make_org_override_key(endpoint_name: str, organization_id: int | None) -> str:
    """Generate the Redis key for organization-level rate limit override lookup.

    Key format: ratelimit_org_override:{endpoint_name}:{organization_id}

    Args:
        endpoint_name: The endpoint name or route pattern.
        organization_id: The organization ID, or None to skip org override lookup.

    Returns:
        Redis key string for the org override hash, or empty string if organization_id is None.
    """
    if organization_id is None:
        return ""
    endpoint_name = normalize_path(endpoint_name)
    return f"{_ORG_OVERRIDE_PREFIX}:{endpoint_name}:{organization_id}"

parse_org_override_key

parse_org_override_key(key)

Parse an org override key to extract endpoint and organization_id.

PARAMETER	DESCRIPTION
`key`	Redis key in format "ratelimit_org_override:{endpoint}:{org_id}" TYPE: `str`

RETURNS	DESCRIPTION
`tuple[str, int] \| None`	Tuple of (endpoint_name, organization_id) or None if parsing fails.

Note

This handles endpoint names that contain colons by parsing from both ends: - The prefix is known and stripped from the left - org_id is parsed from the right (it's always an integer) - Everything in between is the endpoint name

Source code in application_kit/ratelimit.py

def parse_org_override_key(key: str) -> tuple[str, int] | None:
    """Parse an org override key to extract endpoint and organization_id.

    Args:
        key: Redis key in format "ratelimit_org_override:{endpoint}:{org_id}"

    Returns:
        Tuple of (endpoint_name, organization_id) or None if parsing fails.

    Note:
        This handles endpoint names that contain colons by parsing from both ends:
        - The prefix is known and stripped from the left
        - org_id is parsed from the right (it's always an integer)
        - Everything in between is the endpoint name
    """
    result = _parse_key_from_right(key, _ORG_OVERRIDE_PREFIX, num_trailing_ints=1)
    if result is None:
        return None
    endpoint, org_id_str = result
    return endpoint, int(org_id_str)

get_rate_limit_mode

get_rate_limit_mode()

Get rate limit mode from Bender configuration.

RETURNS	DESCRIPTION
`RateLimitMode`	RateLimitMode.on (default) - normal rate limiting, block when over limit
`RateLimitMode`	RateLimitMode.off - completely disabled, skip all rate limit checks
`RateLimitMode`	RateLimitMode.monitor - track violations but don't block, tag to Datadog

Source code in application_kit/ratelimit.py

def get_rate_limit_mode() -> RateLimitMode:
    """Get rate limit mode from Bender configuration.

    Returns:
        RateLimitMode.on (default) - normal rate limiting, block when over limit
        RateLimitMode.off - completely disabled, skip all rate limit checks
        RateLimitMode.monitor - track violations but don't block, tag to Datadog
    """
    try:
        return cast(RateLimitMode, get_configuration("RATE_LIMIT_MODE"))
    except ConfigurationError:
        return RateLimitMode.on

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

parse_last_policy_params

parse_last_policy_params(header_value)

Parse parameters from the last item in a structured rate limit header.

Given '"requests";q=100;w=60, "elements";q=5000;w=60', returns {"q": "5000", "w": "60"} (from the last item).

This matches legacy behavior where the last limiter's headers would overwrite previous values.

Source code in application_kit/ratelimit.py

def parse_last_policy_params(header_value: str) -> dict[str, str]:
    """Parse parameters from the last item in a structured rate limit header.

    Given ``'"requests";q=100;w=60, "elements";q=5000;w=60'``, returns
    ``{"q": "5000", "w": "60"}`` (from the last item).

    This matches legacy behavior where the last limiter's headers would
    overwrite previous values.
    """
    last_item = header_value.rsplit(",", 1)[-1].strip()
    params: dict[str, str] = {}
    for part in last_item.split(";")[1:]:
        key, _, value = part.strip().partition("=")
        if key:
            params[key] = value
    return params

FastAPI

application_kit.fastapi.ratelimit

RateLimitOverride

Bases: BaseModel

Rate limit override configuration stored in Redis.

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)

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,
    )

Django

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

Lua Script

Key Format

Core

application_kit.ratelimit

RateLimitOverride

RateLimitResult dataclass

to_policy_item

to_status_item

to_headers

from_lua_result classmethod

normalize_path

derive_policy_name

make_path_rate_limit_key

make_rate_limit_key

make_override_key

parse_override_key

make_org_override_key

parse_org_override_key

get_rate_limit_mode

is_rate_limit_disabled

parse_last_policy_params

FastAPI

application_kit.fastapi.ratelimit

RateLimitOverride

RateLimiter dataclass

make_key abstractmethod

PathRateLimiter dataclass

ProjectRateLimiter dataclass

make_key

make_override_key

make_org_override_key

is_rate_limit_disabled

apply_rate_limit async

apply_element_rate_limit async

Django

application_kit.django.ratelimit

RateLimitExceeded

rate_limit

Or with explicit endpoint name:

RateLimitResult `dataclass`

from_lua_result `classmethod`

RateLimiter `dataclass`

make_key `abstractmethod`

PathRateLimiter `dataclass`

ProjectRateLimiter `dataclass`

apply_rate_limit `async`

apply_element_rate_limit `async`