Middleware¶

ThrottleMiddleware sits in the ASGI stack and intercepts every request before it reaches any route handler. Use it when you want to enforce rate limits globally or based on path/method patterns, without touching individual route definitions.

Works with any Starlette application

ThrottleMiddleware is a standard Starlette middleware and works with any Starlette-based application, not just FastAPI. If you're using plain Starlette or any other ASGI framework built on Starlette, you can add ThrottleMiddleware the same way: app.add_middleware(ThrottleMiddleware, middleware_throttles=[...]).

Use middleware for cross-cutting concerns

Middleware is the right tool when the throttle logic is independent of which specific route is hit, for example, a global IP-based limit, or a limit on all /api/ traffic regardless of endpoint. For per-route or per-handler limits, prefer dependency injection.

Basic setup¶

Wrap your underlying throttle in MiddlewareThrottle and pass it to ThrottleMiddleware. The backend is typically shared via the lifespan context.

v1.1.0+: Using regular throttles with rules

As of v1.1.0, you can pass regular Throttle instances directly to ThrottleMiddleware, thanks to the new throttle rule API. MiddlewareThrottle is a lightweight wrapper that internally uses ThrottleRule, so you can replicate its functionality by attaching rules directly to a Throttle. See MiddlewareThrottle as a convenience wrapper for details and an example.

from fastapi import FastAPI
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.middleware import MiddlewareThrottle, ThrottleMiddleware

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

throttle = HTTPThrottle(uid="global", rate="1000/min")

app.add_middleware(
    ThrottleMiddleware,
    middleware_throttles=[
        MiddlewareThrottle(throttle),
    ],
)

This applies the throttle to every HTTP request. MiddlewareThrottle without any path, methods, or predicate arguments matches all connections.

HTTP and WebSocket throttles in middleware¶

ThrottleMiddleware handles both HTTP and WebSocket connections. Pass throttles for each connection type in the same middleware_throttles list. Traffik routes them automatically based on the connection's ASGI scope type.

from fastapi import FastAPI
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.middleware import MiddlewareThrottle, ThrottleMiddleware
from traffik.throttles import WebSocketThrottle

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

http_throttle = HTTPThrottle(uid="http:global", rate="500/min")
ws_throttle = WebSocketThrottle(uid="ws:global", rate="60/min")

app.add_middleware(
    ThrottleMiddleware,
    middleware_throttles=[
        MiddlewareThrottle(http_throttle),
        MiddlewareThrottle(ws_throttle, path="/ws/"),
    ],
)

HTTP throttles are never applied to WebSocket connections, and vice versa. Traffik categorizes throttles by connection type at startup, so there is no per-request branching overhead.

Path filtering¶

The path argument on MiddlewareThrottle limits the throttle to requests whose URL path matches. Strings are interpreted as prefix matches (compiled to a regex internally). You can also pass a pre-compiled re.Pattern.

String prefixCompiled regex

A plain string matches any path that starts with that prefix.

from traffik import HTTPThrottle
from traffik.middleware import MiddlewareThrottle

throttle = HTTPThrottle(uid="api", rate="300/min")

# Matches /api/, /api/users, /api/v2/products, etc.
api_middleware_throttle = MiddlewareThrottle(throttle, path="/api/")

Pass a re.Pattern for full regex control.

import re
from traffik import HTTPThrottle
from traffik.middleware import MiddlewareThrottle

throttle = HTTPThrottle(uid="api:versioned", rate="300/min")

# Matches /api/v1/..., /api/v2/..., but not /api/legacy/...
pattern = re.compile(r"^/api/v\d+/")
versioned_middleware_throttle = MiddlewareThrottle(throttle, path=pattern)

Note

When path is a string, it is compiled as a regex pattern. If you use a bare string like "/api/", it matches any path that contains /api/ at that position (anchored from the start). Use re.compile(...) explicitly when you need full regex semantics such as anchors or alternation.

Wildcard patterns are supported

Because MiddlewareThrottle's path parameter uses ThrottleRule internally, it supports the same wildcard syntax: * matches a single path segment (no /), and ** matches any number of segments including /. For example, path="/api/*/users" matches /api/v1/users and /api/v2/users but not /api/v1/admin/users. See Throttle Rules & Wildcards for the full pattern reference.

Method filtering¶

The methods argument restricts the throttle to specific HTTP verbs. WebSocket connections ignore method filtering entirely (they have no HTTP method after the handshake).

from traffik import HTTPThrottle
from traffik.middleware import MiddlewareThrottle

write_throttle = HTTPThrottle(uid="writes", rate="100/min")

# Only applies to POST, PUT, PATCH, and DELETE requests
write_middleware_throttle = MiddlewareThrottle(
    write_throttle,
    methods={"POST", "PUT", "PATCH", "DELETE"},
)

GET requests to any path pass through this throttle without consuming quota.

Predicate filtering¶

For conditions that can't be expressed with path or method alone, provide an async predicate callable. The throttle only applies when the predicate returns True.

import typing
from starlette.requests import HTTPConnection
from traffik import HTTPThrottle
from traffik.middleware import MiddlewareThrottle

throttle = HTTPThrottle(uid="premium:api", rate="2000/min")

async def is_premium_user(
    connection: HTTPConnection,
    context: typing.Optional[typing.Mapping[str, typing.Any]] = None,
) -> bool:
    return connection.headers.get("X-User-Tier") == "premium"

premium_middleware_throttle = MiddlewareThrottle(
    throttle,
    path="/api/",
    predicate=is_premium_user,
)

Keep predicates fast

The predicate runs on every matching request before the throttle strategy executes. Avoid blocking I/O or expensive computation. If you need database lookups, consider caching the result in request.state from an earlier middleware.

Combined: path + methods + predicate¶

All three filters are evaluated conjunctively: the throttle only fires when all conditions are met.

import typing
from starlette.requests import HTTPConnection
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.middleware import MiddlewareThrottle, ThrottleMiddleware
from fastapi import FastAPI

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

upload_throttle = HTTPThrottle(uid="uploads:authenticated", rate="50/hour")

async def is_authenticated(
    connection: HTTPConnection,
    context: typing.Optional[typing.Mapping[str, typing.Any]] = None,
) -> bool:
    return "Authorization" in connection.headers

app.add_middleware(
    ThrottleMiddleware,
    middleware_throttles=[
        MiddlewareThrottle(
            upload_throttle,
            path="/api/uploads/",
            methods={"POST"},
            predicate=is_authenticated,
        )
    ],
)

Execution order within a throttle¶

For each MiddlewareThrottle, filters are evaluated from cheapest to most expensive:

Methods — a set membership check, essentially free.
Path — a compiled regex match against the URL string.
Predicate — your async callable, potentially involving I/O.

If any check fails, the remaining checks are skipped and the throttle is bypassed for that request. The throttle strategy (quota consumption) only runs after all filters pass.

Throttle sorting¶

When ThrottleMiddleware receives a list of MiddlewareThrottle instances, it sorts them before processing. Sorting controls which throttles run first, which is most useful when you want cheap (low-cost) throttles to reject requests before expensive ones do unnecessary work.

The sort parameter accepts:

Value	Behavior
`"cheap_first"` (default)	Throttles with lower `cost` run first
`"cheap_last"`	Throttles with higher `cost` run first
`False` or `None`	No sorting; use the order you provided
A callable	Sorted by your custom key function

from fastapi import FastAPI
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.middleware import MiddlewareThrottle, ThrottleMiddleware

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

cheap_throttle = HTTPThrottle(uid="cheap", rate="1000/min")
expensive_throttle = HTTPThrottle(uid="expensive", rate="100/min")

app.add_middleware(
    ThrottleMiddleware,
    middleware_throttles=[
        MiddlewareThrottle(expensive_throttle, cost=10),
        MiddlewareThrottle(cheap_throttle, cost=1),
    ],
    sort="cheap_first",  # cheap_throttle runs first (default behavior)
)

Indeteminable thottle cost is treated as infinite

A MiddlewareThrottle with no explicit cost (i.e., cost=None), and the wrapped Throttle uses a dynamic cost function, is treated as having infinite cost and sorted last under "cheap_first". This ensures unconstrained throttles don't block cheap ones from short-circuiting early.

Custom sort key¶

Provide a callable that takes a MiddlewareThrottle (or Throttle) and returns any sortable value:

app.add_middleware(
    ThrottleMiddleware,
    middleware_throttles=[...],
    sort=lambda t: t.throttle.uid,  # sort alphabetically by throttle UID
)

MiddlewareThrottle as a convenience wrapper¶

MiddlewareThrottle is a lightweight wrapper around ThrottleRule that simplifies the common case of applying path, method, and predicate filters to a throttle in middleware. Since v1.1.0, you can achieve the same result by attaching rules directly to a regular Throttle using the throttle rule API.

This example shows both approaches side-by-side:

Using MiddlewareThrottle (v1.0.0+)Using regular Throttle with rules (v1.1.0+)

from fastapi import FastAPI
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.middleware import MiddlewareThrottle, ThrottleMiddleware

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

api_throttle = HTTPThrottle(uid="api:writes", rate="200/min")

app.add_middleware(
    ThrottleMiddleware,
    middleware_throttles=[
        MiddlewareThrottle(
            api_throttle,
            path="/api/",
            methods={"POST", "PUT", "DELETE"},
        )
    ],
)

from fastapi import FastAPI
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.middleware import ThrottleMiddleware
from traffik.registry import ThrottleRule

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

# Attach rules directly to the throttle
rule = ThrottleRule(path="/api/", methods={"POST", "PUT", "DELETE"})
api_throttle = HTTPThrottle(uid="api:writes", rate="200/min", rules={rule})

app.add_middleware(
    ThrottleMiddleware,
    middleware_throttles=[api_throttle],  # Pass the throttle directly
)

Both approaches are equivalent. The first approach is more concise for middleware-only use, while the second allows you to share the same filtered throttle across middleware and route-level dependencies.

Using `MiddlewareThrottle` as a FastAPI dependency¶

MiddlewareThrottle implements __call__ and exposes a __signature__ compatible with FastAPI's dependency injection. This is a niche use case, for example when you have a MiddlewareThrottle instance configured with path and method filters, and you also want to use the same filtered throttle as a route-level dependency.

from fastapi import FastAPI, Depends
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.middleware import MiddlewareThrottle

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

throttle = HTTPThrottle(uid="api:write", rate="50/min")
write_middleware_throttle = MiddlewareThrottle(
    throttle,
    methods={"POST", "PUT", "DELETE"},
)

# Used as a dependency — the path/method filters still apply
@app.post("/resource", dependencies=[Depends(write_middleware_throttle)])
async def create_resource(payload: dict):
    return {"created": True}