Skip to content

Rate Limiting

Application Kit provides rate limiting with support for per-project, per-endpoint limits and Redis-backed overrides.

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
  • Configurable time windows
  • Rate limit overrides stored in Redis
  • Standard rate limit headers in responses
flowchart LR
    Request([Request]) --> Auth[Authentication]
    Auth --> RL{Rate Limit<br/>Check}
    RL -->|Under limit| Handler[Request Handler]
    RL -->|Over limit| Reject[429 Too Many Requests]
    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

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

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 single rate limit counter per path, shared across all callers. This means:

  • All requests to the same path share one counter
  • No per-project isolation
  • Rate limit overrides are not supported

Only use this when ProjectRateLimiter isn't possible.

Parameters

Parameter Type Default Description
max_requests int required Maximum requests allowed in the time window
expiry int 1 Time window in seconds
endpoint_name str auto-detected Identifier for the endpoint
redis_dependency str "ratelimit" Name of the Redis dependency

Response Headers

All rate-limited responses include standard headers:

Header Description
RateLimit-Limit Maximum requests allowed in the window
RateLimit-Remaining Requests remaining in the current window
RateLimit-Reset Seconds until the rate limit window resets

Rate Limit Exceeded

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

FastAPI returns an HTTPException with status code 429 and the message "Ratelimit was hit, try again later."

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

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

Headers included: RateLimit-Limit, RateLimit-Remaining, RateLimit-Reset

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

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

Headers included: RateLimit-Limit, RateLimit-Remaining, RateLimit-Reset

Rate Limit Overrides

Override default rate limits for specific projects without code changes. Overrides are stored in Redis and checked atomically with rate limiting.

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.

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


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

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:

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
)

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

Custom Rate Limiters (FastAPI)

For advanced use cases, subclass RateLimiter and implement make_key():

from starlette.requests import Request

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


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)
        return f"ratelimit:{request.scope['route'].path}:{ip}"


class ClientIpRateLimiter(RateLimiter):
    """Rate limit by X-Forwarded-For header."""

    def make_key(self, request: Request) -> str | None:
        if is_rate_limit_disabled():
            return None
        ip = request.headers.get("X-Forwarded-For", "unknown")
        return f"ratelimit:{request.scope['route'].path}:{ip}"

Usage:

@router.get(
    "/public-endpoint",
    dependencies=[Depends(IpRateLimiter(max_requests=50, expiry=1))],
)
async def public_endpoint():
    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

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:

  1. Atomic counting: Increment and check happen in a single Redis operation
  2. Override lookup: Override configuration is fetched and applied atomically
  3. TTL management: Rate limit windows are automatically managed via Redis TTL
flowchart TD
    Request([Request]) --> A
    subgraph LuaScript ["Lua Script (Atomic)"]
        A[Check Override Key] --> B{Override exists?}
        B -->|Yes| C[Use override limits]
        B -->|No| D[Use default limits]
        C --> E[Get current count]
        D --> E
        E --> F{Count >= Limit?}
        F -->|No| G[Increment counter]
        F -->|Yes| H[Over limit]
        G --> I{First request?}
        I -->|Yes| J[Set TTL on key]
        I -->|No| K[Return count]
        J --> K
    end
    subgraph Responses [" "]
        Response429([429 + Headers])
        Response200([200 + Headers])
    end
    H --> Response429
    K --> Response200

Redis key format:

  • Rate limit counter: ratelimit:{endpoint}:{organization_id}:{project_id}
  • Override config: ratelimit_override:{endpoint}:{organization_id}:{project_id}

Configuration

Rate Limit Mode

Rate limiting behavior is controlled by the RATE_LIMIT_MODE environment variable:

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

Set the mode via environment variable:

export RATE_LIMIT_MODE=on  # or "off" or "monitor" (Django only)

Monitor mode is Django-only

FastAPI only supports on and off modes. The monitor mode (which counts requests without blocking and tags Datadog spans) is only available in Django.

Redis Dependency

Add the ratelimit Redis dependency to your application.json:

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

OpenAPI Documentation

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

application_kit.fastapi.ratelimit

RateLimiter dataclass

RateLimiter(max_requests, expiry=1)

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.

TYPE: int

expiry

Time window in seconds (default: 1 second).

TYPE: int

Example
class IpRateLimiter(RateLimiter):
    def make_key(self, request: Request) -> str | None:
        ip = request.headers.get("X-Forwarded-For", "unknown")
        return f"ratelimit:{request.scope['route'].path}:{ip}"

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
80
81
82
83
84
85
86
87
88
89
@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)

Bases: RateLimiter

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

All requests to the same path share a single rate limit counter, regardless of which project they belong to. Useful for public endpoints that don't require authentication.

Does not support rate limit overrides.

ProjectRateLimiter dataclass

ProjectRateLimiter(max_requests, expiry=1)

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_override_key

make_override_key(request)

Generate the Redis key for looking up rate limit overrides.

Source code in application_kit/fastapi/ratelimit.py
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def make_override_key(self, request: Request) -> str | None:
    """Generate the Redis key for looking up rate limit overrides."""
    project_reference = None
    try:
        project_reference = request.state.project_token.reference
    except AttributeError:
        return None

    if project_reference is None:
        return None

    # Normalize path by stripping trailing slash to match make_key()
    path = request.scope["route"].path.rstrip("/") or "/"
    return f"ratelimit_override:{path}:{project_reference.organization_id}:{project_reference.project_id}"

is_rate_limit_disabled

is_rate_limit_disabled()

Check if rate limiting is disabled based on RATE_LIMIT_MODE configuration.

Returns True when RATE_LIMIT_MODE is set to "off", False otherwise. Use this in custom rate limiters to respect the global rate limit setting.

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 f"ratelimit:{request.scope['route'].path}:{ip}"
Source code in application_kit/fastapi/ratelimit.py
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def is_rate_limit_disabled() -> bool:
    """Check if rate limiting is disabled based on RATE_LIMIT_MODE configuration.

    Returns True when RATE_LIMIT_MODE is set to "off", False otherwise.
    Use this in custom rate limiters to respect the global rate limit setting.

    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 f"ratelimit:{request.scope['route'].path}:{ip}"
        ```
    """
    mode = cast(RateLimitMode, get_configuration("RATE_LIMIT_MODE"))
    return mode == RateLimitMode.off

set_rate_limit_override async

set_rate_limit_override(
    redis,
    organization_id,
    project_id,
    endpoint_pattern,
    max_requests,
    expiry,
    override_ttl=None,
)

Set a rate limit override for a specific project and endpoint.

PARAMETER DESCRIPTION
redis

Async Redis connection

TYPE: Redis[bytes]

organization_id

The organization ID

TYPE: int

project_id

The project ID

TYPE: int

endpoint_pattern

The route path pattern (e.g., "/datasets/{dataset_id}/path")

TYPE: str

max_requests

Maximum requests allowed in the window

TYPE: int

expiry

Time window in seconds

TYPE: int

override_ttl

Optional TTL for the override itself (auto-expire)

TYPE: int | None DEFAULT: None

Source code in application_kit/fastapi/ratelimit.py
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
async def set_rate_limit_override(
    redis: "redis.asyncio.Redis[bytes]",
    organization_id: int,
    project_id: int,
    endpoint_pattern: str,
    max_requests: int,
    expiry: int,
    override_ttl: int | None = None,
) -> None:
    """Set a rate limit override for a specific project and endpoint.

    Args:
        redis: Async Redis connection
        organization_id: The organization ID
        project_id: The project ID
        endpoint_pattern: The route path pattern (e.g., "/datasets/{dataset_id}/path")
        max_requests: Maximum requests allowed in the window
        expiry: Time window in seconds
        override_ttl: Optional TTL for the override itself (auto-expire)
    """
    key = make_override_key(endpoint_pattern, organization_id, project_id)

    await redis.hset(key, mapping={"max_requests": str(max_requests), "expiry": str(expiry)})
    if override_ttl:
        await redis.expire(key, override_ttl)

get_rate_limit_override async

get_rate_limit_override(
    redis, organization_id, project_id, endpoint_pattern
)

Get current rate limit override if exists.

Source code in application_kit/fastapi/ratelimit.py
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
async def get_rate_limit_override(
    redis: "redis.asyncio.Redis[bytes]",
    organization_id: int,
    project_id: int,
    endpoint_pattern: str,
) -> RateLimitOverride | None:
    """Get current rate limit override if exists."""
    key = make_override_key(endpoint_pattern, organization_id, project_id)
    values = await redis.hmget(key, "max_requests", "expiry")

    if values[0] is not None and values[1] is not None:
        return RateLimitOverride(
            max_requests=int(values[0]),
            expiry=int(values[1]),
        )
    return None

delete_rate_limit_override async

delete_rate_limit_override(
    redis, organization_id, project_id, endpoint_pattern
)

Delete a rate limit override. Returns True if it existed.

Source code in application_kit/fastapi/ratelimit.py
255
256
257
258
259
260
261
262
263
264
async def delete_rate_limit_override(
    redis: "redis.asyncio.Redis[bytes]",
    organization_id: int,
    project_id: int,
    endpoint_pattern: str,
) -> bool:
    """Delete a rate limit override. Returns True if it existed."""
    key = make_override_key(endpoint_pattern, organization_id, project_id)
    deleted = await redis.delete(key)
    return deleted > 0

list_rate_limit_overrides async

list_rate_limit_overrides(
    redis, organization_id, project_id
)

List all rate limit overrides for a project.

Source code in application_kit/fastapi/ratelimit.py
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
async def list_rate_limit_overrides(
    redis: "redis.asyncio.Redis[bytes]",
    organization_id: int,
    project_id: int,
) -> list[tuple[str, RateLimitOverride]]:
    """List all rate limit overrides for a project."""
    pattern = f"ratelimit_override:*:{organization_id}:{project_id}"
    overrides: list[tuple[str, RateLimitOverride]] = []

    async for key in redis.scan_iter(match=pattern):
        hash_data = await redis.hgetall(key)
        if hash_data:
            # Extract endpoint pattern from key (between first colon and org_id)
            key_str = key.decode() if isinstance(key, bytes) else key
            # Format: ratelimit_override:{endpoint}:{org_id}:{project_id}
            # Split from the right to handle endpoints that may contain colons
            parts = key_str.rsplit(":", 2)
            if len(parts) == 3:
                endpoint_pattern = parts[0].removeprefix("ratelimit_override:")
                override = RateLimitOverride(
                    max_requests=int(hash_data[b"max_requests"]),
                    expiry=int(hash_data[b"expiry"]),
                )
                overrides.append((endpoint_pattern, override))

    return overrides

clear_project_rate_limit_overrides async

clear_project_rate_limit_overrides(
    redis, organization_id, project_id
)

Clear all rate limit overrides for a project. Returns count of deleted overrides.

Source code in application_kit/fastapi/ratelimit.py
295
296
297
298
299
300
301
302
303
304
305
306
307
308
async def clear_project_rate_limit_overrides(
    redis: "redis.asyncio.Redis[bytes]",
    organization_id: int,
    project_id: int,
) -> int:
    """Clear all rate limit overrides for a project. Returns count of deleted overrides."""
    pattern = f"ratelimit_override:*:{organization_id}:{project_id}"
    deleted_count = 0

    async for key in redis.scan_iter(match=pattern):
        await redis.delete(key)
        deleted_count += 1

    return deleted_count

application_kit.django.ratelimit

RateLimitExceeded

RateLimitExceeded(limit, remaining, reset)

Bases: HttpException

Raised when rate limit is exceeded.

Source code in application_kit/django/ratelimit.py
50
51
52
53
54
55
56
57
58
59
def __init__(self, limit: int, remaining: int, reset: int) -> None:
    headers = {
        "RateLimit-Limit": str(limit),
        "RateLimit-Remaining": str(remaining),
        "RateLimit-Reset": str(reset),
    }
    super().__init__(status=429, message="Rate limit exceeded. Try again later.", headers=headers)
    self.limit = limit
    self.remaining = remaining
    self.reset = reset

rate_limit

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

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.

TYPE: int

expiry

Time window in seconds (default: 1 second).

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'

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
  • set_rate_limit_override() from application_kit.fastapi.ratelimit
  • Override key format: ratelimit_override:{endpoint_name}:{org_id}:{project_id}
Source code in application_kit/django/ratelimit.py
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
def rate_limit(
    max_requests: int,
    expiry: int = 1,
    endpoint_name: str | None = None,
    redis_dependency: str = "ratelimit",
) -> 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.
        expiry: Time window in seconds (default: 1 second).
        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.

    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:
        - `set_rate_limit_override()` from `application_kit.fastapi.ratelimit`
        - Override key format: `ratelimit_override:{endpoint_name}:{org_id}:{project_id}`
    """

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

    return decorator