Decorators¶

The @throttled decorator applies throttling directly to a route function. It keeps the rate limit configuration visible right above the handler, which some developers find easier to scan than a dependencies=[...] list on the route decorator.

Traffik ships two versions of @throttled: one for Starlette and one for FastAPI. They behave differently in how the throttle receives the connection object.

Two decorator variants¶

	`traffik.decorators.generic.throttled`	`traffik.decorators.fastapi.throttled`
Works with	Starlette and FastAPI	FastAPI only
Requires `Request` param	Yes - the route function must declare a `Request` or `WebSocket` parameter	No - FastAPI's DI injects the connection automatically
Mechanism	Inspects function args for an `HTTPConnection` at call time	Wraps the route with a hidden `Depends(throttle)`
OpenAPI impact	None	None
Recommended for	Starlette apps; Starlette-compatible code	FastAPI apps

`traffik.decorators.fastapi.throttled`: FastAPI version¶

Import from traffik.decorators.fastapi. Your route function does not need to declare a Request parameter. FastAPI resolves the connection internally.

from fastapi import FastAPI
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.decorators.fastapi import throttled  # FastAPI-specific

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

throttle = HTTPThrottle(uid="items:read", rate="100/min")

@app.get("/items")
@throttled(throttle)
async def list_items():
    return {"items": []}

The decorator order matters: @app.get(...) must be the outermost decorator, @throttled(...) the next one in.

Tip

Because traffik.decorators.fastapi.throttled relies on FastAPI's dependency injection, the throttle check happens before any other dependencies in the function signature are resolved. This means a rejected request never triggers database queries or other expensive dependencies.

`traffik.decorators.generic.throttled`: Starlette version¶

Import from traffik.decorators.generic (or directly from traffik). The route function must declare a Request or WebSocket parameter. The decorator inspects the function arguments at call time to find the connection object.

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.decorators.generic import throttled  # Starlette version

backend = InMemoryBackend()

throttle = HTTPThrottle(uid="items:read", rate="100/min")

@throttled(throttle)
async def list_items(request: Request) -> JSONResponse:
    return JSONResponse({"items": []})

app = Starlette(
    routes=[Route("/items", list_items, methods=["GET"])],
    lifespan=backend.lifespan,
)

The Starlette decorator requires an HTTPConnection parameter

If the decorated function has no Request or WebSocket parameter (positional or keyword), Traffik raises a ValueError at call time. For FastAPI routes without a Request parameter, use traffik.decorators.throttled instead.

The Starlette version also works in FastAPI if you prefer it, just make sure your handler declares the Request.

from fastapi import FastAPI
from starlette.requests import Request
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.throttles import throttled  # Starlette version, used in FastAPI

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

throttle = HTTPThrottle(uid="items:read-sl", rate="100/min")

@app.get("/items")
@throttled(throttle)
async def list_items(request: Request):  # <-- required for starlette.throttles version
    return {"items": []}

Multiple throttles with `@throttled`¶

Pass multiple throttles to @throttled. They are checked sequentially: the first throttle is checked first, and if it rejects the request, the remaining throttles are never consulted.

FastAPI decoratorStarlette decorator

from fastapi import FastAPI
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.decorators.fastapi import throttled

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

burst_throttle = HTTPThrottle(uid="search:burst", rate="10/s")
sustained_throttle = HTTPThrottle(uid="search:sustained", rate="200/min")

@app.get("/search")
@throttled(burst_throttle, sustained_throttle)
async def search(q: str):
    return {"query": q, "results": []}

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.throttles import throttled

backend = InMemoryBackend()

burst_throttle = HTTPThrottle(uid="search:burst", rate="10/s")
sustained_throttle = HTTPThrottle(uid="search:sustained", rate="200/min")

@throttled(burst_throttle, sustained_throttle)
async def search(request: Request) -> JSONResponse:
    q = request.query_params.get("q", "")
    return JSONResponse({"query": q, "results": []})

app = Starlette(
    routes=[Route("/search", search, methods=["GET"])],
    lifespan=backend.lifespan,
)

Sequential checking: first failure stops the rest

With @throttled(first, second, third), if first rejects the request, second and third are never evaluated. This is efficient, no unnecessary quota checks, but it also means the order of throttles matters. Put the most restrictive (or cheapest to evaluate) throttle first.

WebSocket routes¶

@throttled works with WebSocket routes too. Use WebSocketThrottle and apply the decorator the same way. This throttles at the connection level, recording one hit when the WebSocket handshake occurs.

FastAPI decoratorStarlette decorator

from fastapi import FastAPI, WebSocket
from traffik.throttles import WebSocketThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.decorators.fastapi import throttled

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

ws_throttle = WebSocketThrottle(uid="ws:connect", rate="30/min")

@app.websocket("/ws")
@throttled(ws_throttle)
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_json({"status": "connected"})
    await websocket.close()

from starlette.applications import Starlette
from starlette.routing import WebSocketRoute
from starlette.websockets import WebSocket
from traffik.throttles import WebSocketThrottle, throttled
from traffik.backends.inmemory import InMemoryBackend

backend = InMemoryBackend()

ws_throttle = WebSocketThrottle(uid="ws:connect", rate="30/min")

@throttled(ws_throttle)
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    await websocket.send_json({"status": "connected"})
    await websocket.close()

app = Starlette(
    routes=[WebSocketRoute("/ws", websocket_endpoint)],
    lifespan=backend.lifespan,
)

For per-message throttling inside an established WebSocket connection, see Direct Usage.

Shorthand string specs¶

Instead of building a Throttle up front, @throttled(...) also accepts shorthand strings. This is handy for simple, one-off limits where a full HTTPThrottle(...) call would just add noise.

"<rate>"
"<rate>|<strategy>"
"<rate>|<strategy>|<type>"
"<uid>"
"<uid>|<rate>"
"<uid>|<rate>|<strategy>"
"<uid>|<rate>|<strategy>|<type>"

Segments are separated by |, not :, so a uid can freely use : for namespacing (e.g. "api:items", the convention used throughout these docs) without conflicting with the spec grammar. <strategy> is one of the aliases registered in traffik.throttles.specs.STRATEGY_ALIASES (e.g. "fixed_window", "token_bucket", "sliding_window_log"), and <type> is "http" or "ws", defaulting to "http".

FastAPI decoratorStarlette decorator

from fastapi import FastAPI
from traffik.backends.inmemory import InMemoryBackend
from traffik.decorators.fastapi import throttled

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

@app.get("/items")
@throttled("api:items|100/min|token_bucket")
async def list_items():
    return {"items": []}

from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
from traffik.backends.inmemory import InMemoryBackend
from traffik.decorators.generic import throttled

backend = InMemoryBackend()

@throttled("api:items|100/min|token_bucket")
async def list_items(request: Request) -> JSONResponse:
    return JSONResponse({"items": []})

app = Starlette(
    routes=[Route("/items", list_items, methods=["GET"])],
    lifespan=backend.lifespan,
)

Pre-built throttles and shorthand strings can be mixed freely in the same call, and a bare rate (no strategy or type) defaults to the fixed-window strategy and the "http" type:

@throttled(burst_throttle, "200/min")  # one pre-built, one shorthand
async def route(...): ...

A string with no rate at all (just "<uid>") doesn't build a throttle, it looks the uid up in the registry and reuses whatever is already registered under it, raising if nothing is there yet:

HTTPThrottle(uid="api:items", rate="100/min")  # registered once, elsewhere

@throttled("api:items")  # reuses the throttle above
async def list_items(): ...

uid= and type= keyword arguments on @throttled(...) supply defaults for specs that don't embed their own, and are rejected if they conflict with a value a spec does embed. Anything a shorthand string can't express (a custom strategy, HTTP-only use_method, and so on) still needs a Throttle built directly.

Decorators vs. dependencies: when to prefer each¶

Both approaches produce the same runtime behavior. The choice is largely a matter of style:

Decorators keep the throttle configuration adjacent to the handler. If you scan a file top-to-bottom, the rate limit is immediately visible before you read the function body.
Dependencies keep route configuration in the route decorator. Useful when you want all route metadata, including status codes, response models, and dependencies, in one place.

You can mix them freely. A common pattern is to use a router-level dependency for a broad limit and a decorator for a tight per-handler limit:

from fastapi import FastAPI, APIRouter, Depends
from traffik import HTTPThrottle
from traffik.backends.inmemory import InMemoryBackend
from traffik.decorators.fastapi import throttled

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

global_throttle = HTTPThrottle(uid="api:global", rate="1000/min")
export_throttle = HTTPThrottle(uid="api:export", rate="5/min")

router = APIRouter(
    prefix="/api/v1",
    dependencies=[Depends(global_throttle)],
)

@router.get("/data")
async def get_data():
    return {"data": []}

@router.get("/export")
@throttled(export_throttle)
async def export_data():
    # hits global_throttle (via router dep) + export_throttle (via decorator)
    return {"export": "..."}

app.include_router(router)

Decorators¶

Two decorator variants¶

traffik.decorators.fastapi.throttled: FastAPI version¶

traffik.decorators.generic.throttled: Starlette version¶

Multiple throttles with @throttled¶

WebSocket routes¶

Shorthand string specs¶

Decorators vs. dependencies: when to prefer each¶

`traffik.decorators.fastapi.throttled`: FastAPI version¶

`traffik.decorators.generic.throttled`: Starlette version¶

Multiple throttles with `@throttled`¶