Rate Limiting¶

Rate limiting infrastructure for API protection.

Configuration¶

Rate limit configuration and settings.

Rate limiting configuration for API endpoints.

class RateLimitTier[source]¶

Bases: str, Enum

Rate limit tier levels with corresponding limits.

CRITICAL = 'critical'¶

HIGH = 'high'¶

MEDIUM = 'medium'¶

LOW = 'low'¶

__new__(value)¶

class RateLimitConfig[source]¶

Bases: BaseSettings

Rate limiting configuration with tiered limits and multipliers.

model_config: ClassVar[SettingsConfigDict] = {'arbitrary_types_allowed': True, 'case_sensitive': False, 'cli_avoid_json': False, 'cli_enforce_required': False, 'cli_exit_on_error': True, 'cli_flag_prefix_char': '-', 'cli_hide_none_type': False, 'cli_ignore_unknown_args': False, 'cli_implicit_flags': False, 'cli_kebab_case': False, 'cli_parse_args': None, 'cli_parse_none_str': None, 'cli_prefix': '', 'cli_prog_name': None, 'cli_shortcuts': None, 'cli_use_class_docs_for_groups': False, 'enable_decoding': True, 'env_file': '.env', 'env_file_encoding': 'utf-8', 'env_ignore_empty': False, 'env_nested_delimiter': None, 'env_nested_max_split': None, 'env_parse_enums': None, 'env_parse_none_str': None, 'env_prefix': 'RATELIMIT_', 'extra': 'ignore', 'json_file': None, 'json_file_encoding': None, 'nested_model_default_partial_update': False, 'protected_namespaces': ('model_validate', 'model_dump', 'settings_customise_sources'), 'secrets_dir': None, 'toml_file': None, 'validate_default': True, 'yaml_config_section': None, 'yaml_file': None, 'yaml_file_encoding': None}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

critical_limit: int¶

high_limit: int¶

medium_limit: int¶

low_limit: int¶

authenticated_multiplier: float¶

staff_multiplier: float¶

enabled: bool¶

redis_key_prefix: str¶

window_seconds: int¶

get_limit(tier, *, is_authenticated=False, is_staff=False)[source]¶

Calculate rate limit based on tier and user status.

Parameters:

tier (RateLimitTier) – Rate limit tier level
is_authenticated (bool) – Whether user is authenticated
is_staff (bool) – Whether user is staff

Return type:

int

Returns:

Calculated rate limit per minute

get_tier_limits()[source]¶

Get all tier limits for different user types.

Return type:: dict[str, dict[str, int]]
Returns:: Dictionary mapping tier names to limits for anonymous, authenticated, and staff users

get_ratelimit_config()[source]¶

Get cached rate limit configuration instance.

Return type:: RateLimitConfig
Returns:: RateLimitConfig instance

Exceptions¶

Rate limit exceptions.

Rate limit exception handler with HTMX-aware responses.

This module provides a custom exception handler for 429 Too Many Requests errors that adapts responses based on the request type (HTMX, API, or browser). It automatically includes Retry-After headers to inform clients when they can retry.

Features:

HTMX requests: Returns toast notification with rate limit message
API requests: Returns JSON with Retry-After header
Browser requests: Renders full 429 error template
Automatic Retry-After header extraction from exception

Usage:

from pydotorg.core.ratelimit.exceptions import rate_limit_exception_handler
from litestar.exceptions import TooManyRequestsException

app = Litestar(
    exception_handlers={
        TooManyRequestsException: rate_limit_exception_handler,
    }
)

rate_limit_exception_handler(request, exc)[source]¶

Handle 429 Too Many Requests exceptions with adaptive responses.

Provides different response formats based on the request type: - HTMX requests: Empty response with toast notification trigger - API requests: JSON response with error details - Browser requests: Full HTML error page with friendly message

All responses include the Retry-After header to inform clients when they can retry their request.

Parameters:

request (Request) – The incoming request
exc (TooManyRequestsException) – The TooManyRequestsException that was raised

Returns:

Appropriate response based on request type

Return type:

Response | Template

Example

>>> @get("/resource")
>>> async def resource():
>>> # This will trigger rate limit exception if limit exceeded
>>>     return {"data": "resource"}

Identifier¶

Client identification for rate limiting.

Rate limit identifier for user-aware rate limiting.

async get_rate_limit_identifier(request)[source]¶

Generate a rate limit identifier based on user authentication status.

Returns different identifier prefixes based on user role:

admin:{user_id} for staff or superuser accounts
user:{user_id} for authenticated regular users
anon:{ip_address} for anonymous/unauthenticated users

The identifier is used to track rate limits per user/IP, allowing different rate limit configurations for different user types.

Parameters:: request (Request) – The incoming Litestar request object. User is populated by UserPopulationMiddleware if authenticated.
Return type:: str
Returns:: String identifier in format prefix:value for rate limit tracking.

Example:

# Authenticated staff user
identifier = await get_rate_limit_identifier(request)
# Returns: "admin:550e8400-e29b-41d4-a716-446655440000"

# Regular authenticated user
identifier = await get_rate_limit_identifier(request)
# Returns: "user:123e4567-e89b-12d3-a456-426614174000"

# Anonymous user from IP 192.168.1.1
identifier = await get_rate_limit_identifier(request)
# Returns: "anon:192.168.1.1"

Middleware¶

Rate limiting middleware.

Rate limiting middleware configuration using Redis.

This module configures rate limiting for the application using Litestar’s built-in RateLimitConfig with Redis as the backend store. The middleware enforces per-IP rate limits while excluding health checks, static assets, and API documentation.

Configuration:

Rate limits are applied per client IP address. Excluded routes:

/health (health check endpoint)
/static/* (static files)
/schema (OpenAPI schema)
/docs (API documentation)
/api (OpenAPI endpoint)

Usage:

from pydotorg.core.ratelimit.middleware import create_rate_limit_config
from pydotorg.config import settings
from litestar.stores.redis import RedisStore

rate_limit_config = create_rate_limit_config(settings)
redis_store = RedisStore.with_client(url=settings.redis_url)

app = Litestar(
    middleware=[rate_limit_config.middleware],
    stores={"rate_limit": redis_store},
    ...
)

create_rate_limit_config(settings)[source]¶

Create rate limiting configuration using Redis as the backend store.

Configures per-IP rate limiting with:

Redis-based persistent storage for distributed rate limiting
Automatic rate limit headers (X-Rate-Limit-*, Retry-After)
Exclusion of health checks, static files, and API documentation
100 requests per minute per IP (default)

Note

The Redis store must be configured in the Litestar app’s stores parameter:

from litestar.stores.redis import RedisStore
redis_store = RedisStore.with_client(url=settings.redis_url)
app = Litestar(stores={"rate_limit": redis_store}, ...)

Parameters:: settings (Settings) – Application settings containing Redis connection URL
Returns:: Configured rate limit middleware instance
Return type:: RateLimitConfig

Example

>>> from pydotorg.config import settings
>>> rate_limit_config = create_rate_limit_config(settings)
>>> rate_limit_config.rate_limit  # ("minute", 100)

create_response_cache_config(settings)[source]¶

Create response cache configuration using Redis as the backend store.

Configures response caching with Redis for improved performance. This is separate from rate limiting but uses the same Redis instance.

Note

The Redis store must be configured in the Litestar app’s stores parameter:

>>> from litestar.stores.redis import RedisStore
>>> redis_store = RedisStore.with_client(url=settings.redis_url)
>>> app = Litestar(stores={"response_cache": redis_store}, ...)

Parameters:: settings (Settings) – Application settings containing Redis connection URL
Returns:: Configured response cache instance
Return type:: ResponseCacheConfig

Example

>>> cache_config = create_response_cache_config(settings)
>>> cache_config.default_expiration  # 60