Rate and concurrency limits

What it's for: how many requests each client may make in a window, and how many may run at once, counted across every process.

An API with a public search limited per client address, routes for signed-in users that share one budget per user (a report takes ten times as much of it as a list of orders), and a slow report endpoint that renders at most two reports at a time across every process. FastAPIRateLimiter counts the requests per minute, and FastAPIConcurrencyLimiter the ones in flight.

Run it

From a clone of the repository:

make deps                 # .venv, with every extra
source .venv/bin/activate
uvicorn examples.limits.main:app

The example has three users, alice, bob and carol, whose tokens are alice-token, bob-token and carol-token.

Try it

The public search allows 10 requests a minute per address. Every response says how many are left (r), and for how many seconds at least (t):

$ curl -i "http://127.0.0.1:8000/search?q=oli"
HTTP/1.1 200 OK
ratelimit: "10/60s";r=9;t=16
ratelimit-policy: "10/60s";q=10;w=60

[{"name":"Olive oil","price":9.5},{"name":"Olives","price":3.2}]

The eleventh within the minute is refused, with the seconds to wait:

$ for i in $(seq 9); do curl -s -o /dev/null "http://127.0.0.1:8000/search?q=oli"; done
$ curl -i "http://127.0.0.1:8000/search?q=oli"
HTTP/1.1 429 Too Many Requests
ratelimit: "10/60s";r=0;t=22
ratelimit-policy: "10/60s";q=10;w=60
retry-after: 22

{"detail":"Too Many Requests"}

Signed-in users have one budget of 30 a minute for /orders and /reports together:

$ curl -i http://127.0.0.1:8000/orders -H "Authorization: Bearer alice-token"
HTTP/1.1 200 OK
ratelimit: "api";r=29;t=16
ratelimit-policy: "api";q=30;w=60

[{"id":1,"customer":"alice","total":12.5},{"id":2,"customer":"alice","total":25.0},{"id":3,"customer":"alice","total":37.5}]

A report takes 10 of it, and three seconds to render:

$ curl -i -X POST http://127.0.0.1:8000/reports -H "Authorization: Bearer alice-token" \
    -H "Content-Type: application/json" -d '{"month": "2026-09"}'
HTTP/1.1 200 OK
ratelimit: "api";r=19;t=16
ratelimit-policy: "api";q=30;w=60

{"month":"2026-09","customer":"alice","pages":12}

After a second report Alice has 9 left: not enough for a third, but enough to list her orders. Bob's budget is his own:

$ curl -si -X POST http://127.0.0.1:8000/reports -H "Authorization: Bearer alice-token" \
    -H "Content-Type: application/json" -d '{"month": "2026-10"}' | grep ratelimit:
ratelimit: "api";r=9;t=13

$ curl -i -X POST http://127.0.0.1:8000/reports -H "Authorization: Bearer alice-token" \
    -H "Content-Type: application/json" -d '{"month": "2026-08"}'
HTTP/1.1 429 Too Many Requests
ratelimit: "api";r=9;t=10
ratelimit-policy: "api";q=30;w=60
retry-after: 13

{"detail":"Too Many Requests"}

$ curl -si http://127.0.0.1:8000/orders -H "Authorization: Bearer alice-token" | grep ratelimit:
ratelimit: "api";r=8;t=10

$ curl -si http://127.0.0.1:8000/orders -H "Authorization: Bearer bob-token" | grep ratelimit:
ratelimit: "api";r=29;t=10

Without a valid token, a request is refused before it reaches any limiter:

$ curl -i http://127.0.0.1:8000/orders
HTTP/1.1 401 Unauthorized
www-authenticate: Bearer

{"detail":"Unauthorized"}

Three reports at once: two render, and the third waits 2 seconds for a slot, then gets a 503 with Retry-After: 3:

$ for month in 2026-07 2026-08 2026-09; do \
    curl -s -o /dev/null -w "$month: %{http_code} after %{time_total}s\n" \
      -X POST http://127.0.0.1:8000/reports -H "Authorization: Bearer carol-token" \
      -H "Content-Type: application/json" -d "{\"month\": \"$month\"}" & \
  done; wait
2026-08: 503 after 2.009854s
2026-09: 200 after 3.009745s
2026-07: 200 after 3.009700s

What to look at

• Who a client is: /search counts by the address that the ASGI server reports (IPv6 addresses by their /64 network), never by X-Forwarded-For. uvicorn takes the address from that header only on connections from --forwarded-allow-ips (127.0.0.1 and ::1 by default, which is why a curl on the same machine can set it): set it to your proxy's address.
• Authentication first: authenticate is a dependency of the router, so it runs before the routes' own dependencies. The limiter's identifier reads the user it found, and a request without a valid token gets its 401 before it takes anybody's budget.
• One budget, several routes: the limiters made with api_budget have the same name and rate, so they share each user's 30 a minute; cost=10 makes a report take ten of them, and a user with 9 left can still list orders. Requests that fail validation count too: FastAPI runs the dependencies before it answers 422.
• The budget before the slots: on /reports the rate limiter comes first, so a client over its budget is refused at once instead of waiting for a slot.
• Slots for the slow part: the slot is held while the report renders and given back after the response is sent. name="reports" lets a job outside requests take the same slots with async with report_slots.slot().
• Shared state: InMemoryBackend counts in one process. With Redis or Valkey every process shares the counters and slots; otherwise each one would allow 10 searches a minute and 2 reports at a time of its own.

The code

examples/limits/main.py

"""An API with rate limits per client and per user, and a concurrency limit on its slowest
endpoint.

    uvicorn examples.limits.main:app

- `GET /search` is public: 10 requests a minute per client address (IPv6 addresses by their /64
  network, what an ISP gives a single customer).
- `GET /orders` and `POST /reports` are for signed-in users, and share one budget of 30 requests a
  minute per user, of which each report takes 10.
- `POST /reports` renders at most 2 reports at a time, across every process: a request waits up
  to 2 seconds for a slot, then gets a 503.

Over a rate limit, clients get a 429. Refused requests say when to retry (`Retry-After`), and
the other responses how much is left (`RateLimit` and `RateLimit-Policy`), for clients to slow
down before they get a 429.

The counters and the slots live in the cache's backend: InMemoryBackend here, so that it runs
without a server. Production uses Redis or Valkey, so that every process shares them.
"""

import asyncio
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from functools import partial
from typing import Annotated

import uvicorn
from fastapi import APIRouter, Depends, FastAPI, HTTPException, Query, Request, status
from pydantic import BaseModel, Field
from starlette.requests import HTTPConnection

from fastapi_alcazaba import Cache, FastAPIConcurrencyLimiter, FastAPIRateLimiter
from fastapi_alcazaba.backends.inmemory import InMemoryBackend

# Stands in for your users and their access tokens (a database, or a JWT check).
TOKENS = {"alice-token": "alice", "bob-token": "bob", "carol-token": "carol"}


class Product(BaseModel):
    name: str
    price: float


class Order(BaseModel):
    id: int
    customer: str
    total: float


class ReportRequest(BaseModel):
    month: str = Field(pattern=r"^\d{4}-(0[1-9]|1[0-2])$", examples=["2026-09"])


class Report(ReportRequest):
    customer: str
    pages: int


CATALOG = [
    Product(name="Olive oil", price=9.5),
    Product(name="Olives", price=3.2),
    Product(name="Oregano", price=2.1),
    Product(name="Orange jam", price=4.8),
]


@asynccontextmanager
async def lifespan(application: FastAPI) -> AsyncIterator[None]:
    # InMemoryBackend counts in this process only. Production uses Redis or Valkey
    # (RedisBackend(redis.asyncio.from_url(...)), with the URL in the app's settings): with
    # several processes, each would otherwise allow its own 10 searches a minute and its own 2
    # reports at a time.
    Cache(InMemoryBackend(), prefix="limits").install(application)
    yield


app = FastAPI(lifespan=lifespan)


@app.get("/search", dependencies=[Depends(FastAPIRateLimiter(10, minutes=1))])
async def search(q: Annotated[str, Query(min_length=2, max_length=50)]) -> list[Product]:
    """Public: 10 requests a minute per client address. Behind a proxy, let uvicorn take the
    address from its headers (`--proxy-headers --forwarded-allow-ips=<proxy address>`): the
    limiter never reads `X-Forwarded-For`, which any client can set."""
    return [product for product in CATALOG if q.lower() in product.name.lower()]


def authenticate(request: Request) -> str:
    """Stands in for your authentication: the user whose bearer token it is, else a 401."""
    scheme, _, token = request.headers.get("authorization", "").partition(" ")
    user = TOKENS.get(token) if scheme.lower() == "bearer" else None
    if user is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, headers={"WWW-Authenticate": "Bearer"})
    request.state.user_id = user
    return user


def user_id(connection: HTTPConnection) -> str:
    """Whose budget a request takes: the user that authenticate() found."""
    user: str = connection.state.user_id
    return user


User = Annotated[str, Depends(authenticate)]

# The router's dependencies run before the routes' own: user_id() finds the user, and requests
# without a valid token are refused before they take anybody's budget.
api = APIRouter(dependencies=[Depends(authenticate)])

# Limiters with the same name and rate share one budget, whatever the route; `cost=` makes a
# route take more of it.
api_budget = partial(FastAPIRateLimiter, 30, minutes=1, name="api", identifier=user_id)

# At most 2 reports at a time, whoever asks for them: each one keeps a CPU busy for seconds.
# Named, so that a job outside requests can take the same slots: `async with report_slots.slot()`.
report_slots = FastAPIConcurrencyLimiter(2, name="reports", wait=2, retry_after=3)


@api.get("/orders", dependencies=[Depends(api_budget())])
async def orders(user: User) -> list[Order]:
    return [Order(id=n, customer=user, total=12.5 * n) for n in range(1, 4)]


async def render_report(user: str, month: str) -> Report:
    """Stands in for the slow part: rendering a PDF, OCR, a call to an AI model."""
    await asyncio.sleep(3)
    return Report(customer=user, month=month, pages=12)


# The budget before the slots: a client over its budget is refused at once, instead of waiting
# for a slot first.
@api.post("/reports", dependencies=[Depends(api_budget(cost=10)), Depends(report_slots)])
async def create_report(report: ReportRequest, user: User) -> Report:
    """10 of the user's 30 a minute, and one of the 2 slots while it renders."""
    return await render_report(user, report.month)


app.include_router(api)


if __name__ == "__main__":
    uvicorn.run("main:app", reload=True)