API Reference¶

Quick reference for every public class and function in Traffik. Each entry lists the import path, constructor parameters, and key methods. Follow the guide links for full usage examples.

Throttles¶

HTTPThrottle¶

Rate limiter for HTTP requests. The most common entry point. Can also be accessed as RequestThrottle

from traffik import HTTPThrottle

Guide: Dependencies · Decorators · Direct Usage

Key methods:

WebSocketThrottle¶

Rate limiter for WebSocket connections.

from traffik import WebSocketThrottle

Guide: Dependencies · Direct Usage

Takes mostly the same parameters as HTTPThrottle but with a few exceptions, and scoped to WebSocket connections. Use inside a WebSocket handler to throttle per-message.

MiddlewareThrottle¶

Throttle applied at the ASGI middleware layer before route handlers execute.

from traffik.throttles import MiddlewareThrottle

Guide: Middleware

Backends¶

InMemoryBackend¶

In-process storage backend. No external dependencies. Not shared across processes.

from traffik.backends.inmemory import InMemoryBackend

Guide: Backends

Integration with FastAPI:

backend = InMemoryBackend()
app = FastAPI(lifespan=backend.lifespan)

RedisBackend¶

Distributed backend backed by Redis. Shares counters across all app instances.

from traffik.backends.redis import RedisBackend

Guide: Backends

MemcachedBackend¶

Distributed backend backed by Memcached.

from traffik.backends.memcached import MemcachedBackend

Guide: Backends

Rate¶

Rate¶

Immutable rate limit definition. Holds the request limit and window duration.

from traffik import Rate
# or
from traffik.rates import Rate

Guide: Rate Format

Key attributes: limit, expire (total ms), rps, rpm, rph, rpd, is_subsecond, unlimited

Parse a rate string:

rate = Rate.parse("100/min")   # 100 per minute
rate = Rate.parse("10/5s")     # 10 per 5 seconds
rate = Rate.parse("5 per hour")

Supported units: ms, s / sec / second, m / min / minute, h / hr / hour, d / day.

Registry¶

ThrottleRegistry¶

Manages throttle registration, rule attachment, and group disable/enable.

from traffik.registry import ThrottleRegistry

Guide: Throttle Registry

Throttles register themselves automatically on construction. You rarely need to create a registry manually — the GLOBAL_REGISTRY is used by default.

GLOBAL_REGISTRY¶

The default ThrottleRegistry used when no registry is specified.

from traffik.registry import GLOBAL_REGISTRY

Rules¶

ThrottleRule¶

A gating rule: the throttle only fires when all attached rules pass.

from traffik.registry import ThrottleRule

Guide: Throttle Rules & Wildcards

All specified conditions are combined with AND: path AND method AND predicate must all match.

BypassThrottleRule¶

Skips throttling when the rule matches (inverse logic compared to ThrottleRule).

from traffik.registry import BypassThrottleRule

Guide: Throttle Rules & Wildcards

Same parameters as ThrottleRule. When all conditions match, the throttle is bypassed for that request.

Headers¶

Header¶

A single rate limit response header with a name, value resolver, and inclusion condition.

from traffik.headers import Header

Guide: Response Headers

Built-in builders (all require when=):

Sentinels:

• Header.DISABLE — Pass as a value in get_headers() overrides to suppress a specific header for that call.

Headers¶

A collection of rate limit response headers resolved per request.

from traffik.headers import Headers

from traffik.headers import Headers, Header

my_headers = Headers({
    "X-RateLimit-Limit": Header.LIMIT(when="always"),
    "X-RateLimit-Remaining": Header.REMAINING(when="always"),
    "Retry-After": Header.RESET_SECONDS(when="throttled"),
})

Supports | for non-mutating merge and .copy() for shallow duplication.

DEFAULT_HEADERS_ALWAYS¶

Preset Headers collection: X-RateLimit-Limit, X-RateLimit-Remaining, and Retry-After resolved on every response.

from traffik.headers import DEFAULT_HEADERS_ALWAYS

DEFAULT_HEADERS_THROTTLED¶

Preset Headers collection: same three headers resolved only on throttled (429) responses.

from traffik.headers import DEFAULT_HEADERS_THROTTLED

Types and Sentinels¶

EXEMPTED¶

Sentinel returned from an identifier function to skip throttling for a connection entirely. No quota is consumed and no backend call is made.

from traffik import EXEMPTED

async def my_identifier(request):
    if request.headers.get("x-admin-key") == ADMIN_KEY:
        return EXEMPTED  # skip throttling
    return request.client.host

StrategyStat¶

Immutable statistics snapshot returned by throttle.stat().

from traffik.types import StrategyStat

Utilities¶

get_remote_address¶

Default connection identifier: returns the client IP address as a string.

from traffik import get_remote_address

throttle = HTTPThrottle("api:v1", rate="100/min", identifier=get_remote_address)

Configuration¶

Global lock defaults, settable via environment variables or these functions. Throttle-level lock_blocking, lock_ttl, and lock_blocking_timeout parameters override these globals.

from traffik.config import (
    get_lock_ttl, set_lock_ttl,
    get_lock_blocking, set_lock_blocking,
    get_lock_blocking_timeout, set_lock_blocking_timeout,
)