FastAPI Rate Limiting: slowapi, a Token Bucket and Redis, Tested

TL;DR

For a single-process FastAPI app, slowapi gets you per-route limits in about ten lines. Switch its strategy from the default fixed window to moving-window, because in the runs below a fixed window let 19 requests through a “10/minute” limit inside two seconds. If you want no third-party dependency, a 40-line token-bucket dependency does the job and can emit the IETF draft RateLimit headers. The moment you run more than one worker or one instance, in-memory counters stop being a limit at all, and you need Redis (or your gateway) to hold the count.

FastAPI ships no rate limiter. That is a deliberate choice: the framework stays small and expects you to bring auth, background jobs and throttling from elsewhere, which I covered in the FastAPI vs Litestar comparison (Litestar bundles a rate-limit middleware; FastAPI doesn’t). Adding one means picking from four approaches. Each fails differently.

This tutorial builds three of them, runs each one, and shows the real output: slowapi with per-route and per-API-key limits, a dependency-only token bucket, and a Redis-backed sliding window for multi-worker deployments. In between there’s a small experiment that replays the same traffic against four algorithms, because the choice of algorithm decides whether your limit holds at all.

Which FastAPI rate limiting approach do you need?

Count your processes first. The right answer depends on how many of them serve your API, and the algorithm is a second-order choice.

One rule cuts across all four. An in-memory limiter with four uvicorn workers gives each worker its own counter, so a “5/minute” limit is really somewhere between 5 and 20 per minute depending on which worker the load balancer picks, and nothing logs an error.

Step 1: slowapi in ten lines

slowapi is adapted from Flask-Limiter for Starlette and FastAPI, built on the limits library, which does the actual counting. Install it with pip install slowapi (or uv add slowapi if you’ve moved to uv; my uv tutorial covers the switch). Here is a full app with a 5-per-minute limit on one route, plus a loop that hits it seven times through FastAPI’s TestClient:

from fastapi import FastAPI, Request, Response
from fastapi.testclient import TestClient
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address, headers_enabled=True)

app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


@app.get("/search")
@limiter.limit("5/minute")
async def search(request: Request, response: Response, q: str = ""):
    return {"q": q, "results": []}


if __name__ == "__main__":
    client = TestClient(app)
    for i in range(1, 8):
        r = client.get("/search", params={"q": "fastapi"})
        print(
            f"req {i}: {r.status_code}",
            f"remaining={r.headers.get('X-RateLimit-Remaining')}",
            f"retry-after={r.headers.get('Retry-After')}",
        )
    print("body of the 429:", r.json())

Running it prints:

req 1: 200 remaining=4 retry-after=60
req 2: 200 remaining=3 retry-after=60
req 3: 200 remaining=2 retry-after=60
req 4: 200 remaining=1 retry-after=60
req 5: 200 remaining=0 retry-after=60
req 6: 429 remaining=0 retry-after=60
req 7: 429 remaining=0 retry-after=60
body of the 429: {'error': 'Rate limit exceeded: 5 per 1 minute'}

Three details in that code trip people up.

The request: Request parameter is mandatory. The slowapi docs say the decorator can’t hook into the endpoint without it. Decorator order counts too: @app.get goes on top and @limiter.limit below it.

The response: Response parameter is the easy one to miss. The first version of this script left it out, and with headers_enabled=True every request crashed with Exception: parameter 'response' must be an instance of starlette.responses.Response. When your endpoint returns a dict, slowapi needs a Response object to write the X-RateLimit-* headers onto, and FastAPI only injects one if you ask for it in the signature. The docs mention this in one line; the traceback doesn’t.

Also notice retry-after=60 on the successful responses. slowapi sends Retry-After on every response once headers are on, not only on the 429. A client that treats any Retry-After as a back-off signal would pause for a minute after every successful call, so check how your SDKs handle the header before turning headers on for a public API.

Step 2: limit per API key, with plan tiers

Limiting by IP is the default and usually the wrong key. Corporate NATs put hundreds of users behind one address, and mobile carriers rotate addresses constantly. If requests carry an API key or a user ID, key on that and fall back to IP for anonymous traffic.

slowapi accepts a callable as the limit. In 0.1.9 the callable receives the key returned by key_func, which makes plan tiers a dictionary lookup:

from fastapi import FastAPI, Request, Response
from fastapi.testclient import TestClient
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.errors import RateLimitExceeded
from slowapi.util import get_remote_address

# toy key store: api key -> plan
PLANS = {"key-free-123": "free", "key-pro-456": "pro"}
PLAN_LIMITS = {"free": "3/minute", "pro": "10/minute", "anon": "2/minute"}


def api_key_or_ip(request: Request) -> str:
    key = request.headers.get("X-API-Key")
    if key in PLANS:
        return f"key:{key}"
    return f"ip:{get_remote_address(request)}"


def limit_for(key: str) -> str:
    if key.startswith("key:"):
        return PLAN_LIMITS[PLANS[key.removeprefix("key:")]]
    return PLAN_LIMITS["anon"]


limiter = Limiter(key_func=api_key_or_ip)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)


@app.get("/reports")
@limiter.limit(limit_for)
async def reports(request: Request, response: Response):
    return {"ok": True}


@app.get("/health")
@limiter.exempt
async def health(request: Request):
    return {"status": "up"}


if __name__ == "__main__":
    client = TestClient(app)
    for who, headers in [
        ("anon", {}),
        ("free", {"X-API-Key": "key-free-123"}),
        ("pro", {"X-API-Key": "key-pro-456"}),
    ]:
        codes = [client.get("/reports", headers=headers).status_code for _ in range(12)]
        print(f"{who:>4}: {codes.count(200):>2} allowed, {codes.count(429):>2} rejected")
    health = [client.get("/health").status_code for _ in range(50)]
    print(f"/health: {health.count(200)} of 50 allowed")

Output, 12 requests per caller:

anon:  2 allowed, 10 rejected
free:  3 allowed,  9 rejected
 pro: 10 allowed,  2 rejected
/health: 50 of 50 allowed

Note the if key in PLANS check. Without it, an attacker sends a random X-API-Key on every request and gets a fresh bucket each time, so unknown keys fall back to the IP. In a real app that lookup hits your API-key table or cache. Keep @limiter.exempt for health checks: once you add default_limits, a load balancer polling /health every two seconds can trip them and get your pods marked unhealthy.

Fixed window vs moving window vs token bucket: a replay test

slowapi defaults to the fixed-window strategy, and limits offers two more: moving window and sliding window counter. The limits strategy docs describe the tradeoff in words. A fixed window is the cheapest (one counter per key) but lets bursts through at window edges, a moving window is exact but stores a timestamp per request, and the sliding window counter approximates the moving window with two counters.

To put numbers on that, this script replays fixed traffic patterns against all three limits strategies plus a hand-rolled token bucket. It swaps the library’s clock for a fake one so every run is deterministic:

"""Replay one traffic pattern against the three `limits` strategies plus a
hand-rolled token bucket, using a fake clock so the run is deterministic."""
import types

import limits.storage.memory as mem_mod
import limits.strategies as strat_mod
from limits import parse
from limits.storage import MemoryStorage
from limits.strategies import (
    FixedWindowRateLimiter,
    MovingWindowRateLimiter,
    SlidingWindowCounterRateLimiter,
)

BASE = 1_760_000_000 // 60 * 60  # a real Unix time, aligned to a minute
NOW = [0.0]
fake_time = types.SimpleNamespace(time=lambda: BASE + NOW[0])
mem_mod.time = fake_time
strat_mod.time = fake_time


class TokenBucket:
    def __init__(self, capacity: int, per_seconds: float):
        self.capacity = capacity
        self.rate = capacity / per_seconds  # tokens per second
        self.tokens = float(capacity)
        self.last = 0.0

    def hit(self) -> bool:
        now = NOW[0]
        self.tokens = min(self.capacity, self.tokens + (now - self.last) * self.rate)
        self.last = now
        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False


# 1 request at t=0, 9 at t=59, then 10 at t=61: a client that idles,
# then bursts across the window boundary.
PATTERN = [(0.0, 1), (59.0, 9), (61.0, 10)]
limit = parse("10/minute")

contenders = {
    "fixed window": FixedWindowRateLimiter(MemoryStorage()),
    "moving window": MovingWindowRateLimiter(MemoryStorage()),
    "sliding window counter": SlidingWindowCounterRateLimiter(MemoryStorage()),
}

print(f"{'strategy':<24}{'t=0':>5}{'t=59':>6}{'t=61':>6}{'allowed in 59..61s':>20}")
for name, limiter in list(contenders.items()) + [("token bucket", TokenBucket(10, 60))]:
    row = []
    for t, n in PATTERN:
        NOW[0] = t
        if isinstance(limiter, TokenBucket):
            ok = sum(limiter.hit() for _ in range(n))
        else:
            ok = sum(limiter.hit(limit, "client-1") for _ in range(n))
        row.append(ok)
    print(f"{name:<24}{row[0]:>5}{row[1]:>6}{row[2]:>6}{row[1] + row[2]:>20}")

strategy                  t=0  t=59  t=61  allowed in 59..61s
fixed window                1     9    10                  19
moving window               1     9     1                  10
sliding window counter      1     9     1                  10
token bucket                1     9     1                  10

The fixed window in limits memory storage starts its 60-second window at the first hit. The client spends one request at t=0, nine at t=59, and the window resets at t=60, so ten more go through at t=61: 19 requests in two seconds under a limit that claims 10 per minute, close to double the intended rate for anyone who times a burst across the reset.

A first version of this experiment used a fake clock starting at 0, and the sliding window counter returned nonsense: a brand-new client only got 7 of 10 requests. The cause was in the library source. limits names its buckets with int((now - expiry) / expiry), and Python’s int() truncates toward zero, so for timestamps under one window length, the “previous” and “current” buckets get the same key. That can’t happen with real Unix timestamps, which is why the script above uses a realistic base time.

Swapping in three more PATTERN values (same script, same 10/minute limit) gives this, with each cell showing how many requests of each burst were allowed:

The sliding window counter errs in both directions. In the first two rows it rejects requests that the exact moving window allows, because it weights the previous clock-aligned bucket’s count by how much of that bucket still overlaps the last minute. At t=90 the counter still charges half of the ten requests from t=0, even though all ten are more than 60 seconds old. In the third row it lets two extra through. The token bucket allows 3 at t=70 by design: it refills at one token every six seconds, so 20 seconds of idling earns three requests back.

Use moving-window for anything security-related (login, password reset, OTP verification) where exactness is the point, and where request counts per key stay small enough that storing a timestamp each is cheap. Use a token bucket for API quotas where you want to allow short bursts and a steady average. Avoid the default fixed window unless you only care about rough volume. In slowapi the switch is one argument: Limiter(key_func=..., strategy="moving-window"). I checked that all three strategy names (fixed-window, moving-window, sliding-window-counter) resolve to the matching limits class in slowapi 0.1.9.

Step 3: FastAPI rate limiting without slowapi

slowapi’s docs still call it “alpha quality code”, and its core is synchronous: open issues #3 and #130 track async support. If that puts you off, a FastAPI dependency is a clean place for a limiter. Dependencies run before the endpoint, can raise HTTPException, and can write headers to the response. A middleware also works, but it runs for every route, and it sits outside Starlette’s exception-handling layer. I checked: raising HTTPException(status_code=429) inside an @app.middleware("http") function returned a plain 500 Internal Server Error, so a middleware limiter has to build its 429 response by hand.

This token bucket keys on API key or client IP and emits the headers from the IETF RateLimit header draft: RateLimit-Policy describes the quota (q) and window (w), and RateLimit reports what’s left (r) and when it resets (t).

import asyncio
import math
import time
from dataclasses import dataclass, field

from fastapi import Depends, FastAPI, HTTPException, Request, Response
from fastapi.testclient import TestClient


@dataclass
class Bucket:
    tokens: float
    updated: float = field(default_factory=time.monotonic)


class TokenBucketLimiter:
    """In-process token bucket: `capacity` burst, refilled at capacity/period per second."""

    def __init__(self, capacity: int, period: float):
        self.capacity = capacity
        self.period = period
        self.rate = capacity / period
        self.buckets: dict[str, Bucket] = {}
        self.lock = asyncio.Lock()

    async def __call__(self, request: Request, response: Response) -> None:
        key = request.headers.get("X-API-Key") or request.client.host
        async with self.lock:
            now = time.monotonic()
            b = self.buckets.setdefault(key, Bucket(tokens=self.capacity, updated=now))
            b.tokens = min(self.capacity, b.tokens + (now - b.updated) * self.rate)
            b.updated = now
            allowed = b.tokens >= 1
            if allowed:
                b.tokens -= 1
            remaining = int(b.tokens)
            # seconds until one full token is back
            wait = 0 if allowed else math.ceil((1 - b.tokens) / self.rate)

        policy = f'"burst";q={self.capacity};w={int(self.period)}'
        state = f'"burst";r={remaining};t={wait}'
        if not allowed:
            raise HTTPException(
                status_code=429,
                detail="rate limit exceeded",
                headers={"Retry-After": str(wait), "RateLimit-Policy": policy, "RateLimit": state},
            )
        response.headers["RateLimit-Policy"] = policy
        response.headers["RateLimit"] = state


burst = TokenBucketLimiter(capacity=3, period=6)  # 3 requests, one back every 2 s
app = FastAPI()


@app.post("/generate", dependencies=[Depends(burst)])
async def generate():
    return {"job": "queued"}


@app.get("/whoami")
async def whoami(request: Request):
    return {"host": request.client.host}


if __name__ == "__main__":
    client = TestClient(app)
    t0 = time.monotonic()
    for i in range(5):
        r = client.post("/generate")
        print(f"t={time.monotonic() - t0:4.1f}s {r.status_code} RateLimit: {r.headers['RateLimit']}"
              + (f"  Retry-After: {r.headers['Retry-After']}" if r.status_code == 429 else ""))
    time.sleep(2.1)
    r = client.post("/generate")
    print(f"t={time.monotonic() - t0:4.1f}s {r.status_code} RateLimit: {r.headers['RateLimit']}")
    # a forged X-Forwarded-For does not change the key: it is the socket peer
    spoof = {"X-Forwarded-For": "203.0.113.9"}
    r = client.post("/generate", headers=spoof)
    who = client.get("/whoami", headers=spoof).json()
    print(f"forged XFF -> {r.status_code}, app sees client.host={who['host']!r}")

t= 0.0s 200 RateLimit: "burst";r=2;t=0
t= 0.0s 200 RateLimit: "burst";r=1;t=0
t= 0.0s 200 RateLimit: "burst";r=0;t=0
t= 0.0s 429 RateLimit: "burst";r=0;t=2  Retry-After: 2
t= 0.0s 429 RateLimit: "burst";r=0;t=2  Retry-After: 2
t= 2.1s 200 RateLimit: "burst";r=0;t=0
forged XFF -> 429, app sees client.host='testclient'

Three requests go through, the next two get a 429 with Retry-After: 2, and after a 2.1-second pause one token has refilled. Unlike slowapi, Retry-After only appears on the rejection, which is what RFC 6585 describes for 429 responses.

Some notes on the design. The asyncio.Lock serializes the read-modify-write on the bucket; without it, two coroutines can both read “1 token left” across an await and both pass. In this exact code there’s no await between the read and the write, so the lock is defensive, but the moment you add an async lookup (say, fetching the user’s plan from a database) it becomes necessary. time.monotonic() avoids surprises when NTP adjusts the wall clock. The buckets dict never shrinks. In production, evict idle keys, or keys from a botnet will slowly eat memory. A cost-weighted version is a small change: deduct cost tokens instead of 1, which is how you’d meter an LLM endpoint by tokens rather than requests.

The last line of output is the part people miss. A forged X-Forwarded-For header did nothing, because the limiter keys on request.client.host, the socket peer. That’s the safe default, and also why a limiter breaks behind a proxy.

Behind a reverse proxy: the client IP problem

Put NGINX, Traefik or a cloud load balancer in front of uvicorn, and request.client.host becomes the proxy’s IP unless uvicorn trusts the proxy. Then every user shares one bucket, and your first busy minute rate-limits the whole user base.

Per the uvicorn settings docs, --proxy-headers is on by default, but uvicorn only honors X-Forwarded-For from addresses listed in --forwarded-allow-ips, which defaults to 127.0.0.1 and ::1. If your proxy runs on the same host, it works out of the box. If it’s a separate container or a cloud LB, set the allowed IPs to the proxy’s address or subnet:

uvicorn app:app --host 0.0.0.0 --port 8000 \
  --proxy-headers --forwarded-allow-ips="10.0.0.0/8"

Don’t set --forwarded-allow-ips="*" on a server reachable from the internet. The uvicorn deployment docs warn to “only trust clients you can actually trust”. With *, any client can send X-Forwarded-For: <random IP> on each request and get a fresh rate-limit bucket every time. Also avoid key functions that parse X-Forwarded-For themselves unless you know exactly how many proxy hops sit in front of the app; slowapi has an open issue (#306) about how its get_ipaddr helper reads that header.

Step 4: share the limit across workers with Redis

Once you scale past one process, the counter has to live somewhere all workers can see. With slowapi that’s one argument, since limits supports Redis, Redis Cluster, Sentinel, Valkey, Memcached and MongoDB backends (storage docs):

limiter = Limiter(
    key_func=api_key_or_ip,
    strategy="moving-window",
    storage_uri="redis://redis:6379/0",
)

Watch the extras pin: slowapi’s redis extra still requires redis>=3.4.1,<4.0.0 according to its PyPI metadata, so install slowapi and a current redis package separately instead of slowapi[redis].

If you went the dependency route, here’s the Redis equivalent of the moving window: a sliding-window log in a Lua script, so the trim, count and insert happen atomically. It reads the clock from Redis with TIME, which means app servers with drifting clocks still agree on the window. The sorted-set pattern is the one Redis’s own rate-limiting tutorial uses for its sliding window log. This block was not executed against a live Redis, since the sandbox for this article has no Redis server. I only confirmed it compiles and that the app and script object build without a connection, so test it against your own instance before relying on it.

import uuid

import redis.asyncio as redis
from fastapi import Depends, FastAPI, HTTPException, Request

# Sliding-window log in one atomic script. Uses the Redis server clock,
# so app servers with drifting clocks still agree on the window.
SLIDING_LOG = """
local key    = KEYS[1]
local limit  = tonumber(ARGV[1])
local window = tonumber(ARGV[2])          -- milliseconds
local member = ARGV[3]
local t      = redis.call('TIME')
local now    = tonumber(t[1]) * 1000 + math.floor(tonumber(t[2]) / 1000)

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)
if count >= limit then
  local oldest = redis.call('ZRANGE', key, 0, 0, 'WITHSCORES')
  return {0, tonumber(oldest[2]) + window - now}
end
redis.call('ZADD', key, now, member)
redis.call('PEXPIRE', key, window)
return {1, 0}
"""

pool = redis.Redis.from_url("redis://localhost:6379/0")
sliding_log = pool.register_script(SLIDING_LOG)


def redis_limit(limit: int, window_s: int, scope: str):
    async def dependency(request: Request) -> None:
        who = request.headers.get("X-API-Key") or request.client.host
        allowed, retry_ms = await sliding_log(
            keys=[f"rl:{scope}:{who}"],
            args=[limit, window_s * 1000, uuid.uuid4().hex],
        )
        if not allowed:
            raise HTTPException(
                status_code=429,
                detail="rate limit exceeded",
                headers={"Retry-After": str(max(1, -(-retry_ms // 1000)))},
            )

    return dependency


app = FastAPI()


@app.post("/login", dependencies=[Depends(redis_limit(5, 60, "login"))])
async def login():
    return {"token": "..."}

The script stores one sorted-set member per allowed request, so memory per key is bounded by the limit itself (five entries for the login route). Rejected requests aren’t recorded, which means a client hammering the endpoint doesn’t extend its own lockout. If you want every attempt to count against the client, run the ZADD before the ZCARD check and reject when the count exceeds the limit. Decide what happens when Redis is down: slowapi has swallow_errors and in_memory_fallback_enabled options for failing open to a local limit, and for the hand-rolled version you’d wrap the call in a try and pick fail-open (availability) or fail-closed (security) per route. For a login endpoint I’d fail closed.

Testing rate-limited endpoints

Rate limits leak state between tests: the fifth test to call /search gets a 429 for no reason it can see. slowapi’s Limiter has a reset() method and an enabled flag, and both behave as expected in 0.1.9. After six calls returned [200, 200, 200, 200, 200, 429], limiter.reset() brought the next call back to 200, and with limiter.enabled = False, eight calls in a row all returned 200. An autouse pytest fixture keeps tests independent:

import pytest
from fastapi.testclient import TestClient

from app_slowapi import app, limiter

client = TestClient(app)


@pytest.fixture(autouse=True)
def fresh_limits():
    limiter.reset()
    yield
    limiter.reset()


def test_sixth_request_is_rejected():
    codes = [client.get("/search").status_code for _ in range(6)]
    assert codes == [200] * 5 + [429]


def test_starts_fresh_after_previous_test():
    assert client.get("/search").status_code == 200

$ python3 -m pytest -q test_limits.py
..                                                                       [100%]
2 passed in 0.46s

Keep at least one test that runs with the limiter on and asserts the 429, its body and its headers. Otherwise a refactor that drops the request parameter or reorders the decorators silently turns the limit off and nothing tells you.

If your stack mixes FastAPI with heavier serialization work, profile before you blame the limiter for latency; Python 3.15’s sampling profiler can attach to a running uvicorn process for exactly that. Timing 20,000 calls to limits with memory storage in this sandbox gave about 4.5 µs per fixed-window hit and 9.3 µs per moving-window hit (with all 20,000 timestamps sitting in the window). A Redis-backed check adds a network round trip on top. On hot paths the JSON decode is usually the bigger cost, which is where msgspec instead of Pydantic earns its keep.

FAQ

How do I implement rate limiting in FastAPI?

Install slowapi, create a Limiter(key_func=get_remote_address), attach it to app.state.limiter, register the RateLimitExceeded handler, and decorate routes with @limiter.limit("5/minute"). The endpoint must accept a request: Request parameter, plus response: Response if you enable headers and return a dict. Step 1 above has the complete working app.

Can I rate limit FastAPI without Redis?

Yes, if you run a single process. slowapi’s default memory storage and the token-bucket dependency in Step 3 both work without Redis. With multiple uvicorn or gunicorn workers, each process keeps its own counters and the effective limit multiplies by the worker count, so you need shared storage at that point.

How do I rate limit per user instead of per IP in FastAPI?

Write a key function that returns the authenticated user ID or a validated API key, and fall back to the client IP for anonymous requests. With slowapi, pass it as key_func; per-plan limits work by passing a callable to @limiter.limit that maps the key to a limit string, as in Step 2.

What is the best rate limiting library for FastAPI?

For most apps, slowapi: it’s MIT-licensed, wraps the mature limits library, and supports Redis, Memcached, MongoDB and Valkey storage. fastapi-limiter is the main alternative. Its 0.2.0 release (February 6, 2026) depends on pyrate-limiter instead of Redis, so older tutorials showing FastAPILimiter.init(redis) no longer match the current API. If you want no dependency, a FastAPI Depends() limiter like the one in Step 3 is about 40 lines.

Which rate limiting algorithm should I use?

Use a moving window (sliding log) for login and other security-sensitive endpoints, and a token bucket for API quotas that should tolerate short bursts. Avoid a plain fixed window for anything abuse-related: in the replay test above it allowed 19 requests in two seconds against a 10-per-minute limit.

Sources

• slowapi on GitHub and slowapi documentation: setup, request/response parameter requirements, alpha-status note
• slowapi on PyPI: release history and the redis extra pin
• limits: rate limiting strategies and storage backends: fixed, moving and sliding window counter tradeoffs
• fastapi-limiter on PyPI and GitHub: 0.2.0 and its pyrate-limiter dependency
• FastAPI middleware docs and dependencies docs
• RFC 6585, section 4: the 429 status code and Retry-After
• draft-ietf-httpapi-ratelimit-headers: the RateLimit and RateLimit-Policy header fields
• uvicorn settings and deployment docs: --proxy-headers and --forwarded-allow-ips
• Redis: build rate limiters with Redis: fixed window, sliding log and token bucket patterns

Bottom line

My default for a new service: slowapi with strategy="moving-window", a key function built on validated API keys, Redis storage from the first deploy that runs more than one worker, and a test that asserts the 429. If the project has a no-new-dependencies rule, the token-bucket dependency above does the same job and gives you standards-track headers for free.