FastAPI

FastAPI Tutorial

Learn how to integrate rate limiting in your FastAPI application.

Halt provides a dedicated `HaltMiddleware` for FastAPI that integrates seamlessly with dependency injection and background tasks.

1. Installation

pip install halt-rate[fastapi]

2. Basic Setup

Add the middleware to your FastAPI application.

from fastapi import FastAPI
from halt import RateLimiter, InMemoryStore, presets
from halt.adapters.fastapi import HaltMiddleware

app = FastAPI()

# Create a limiter
limiter = RateLimiter(
    store=InMemoryStore(),
    policy=presets.PUBLIC_API
)

# Add middleware
app.add_middleware(HaltMiddleware, limiter=limiter)

3. Customizing Limits per Route

You can override limits for specific routes using dependencies or by creating separate limiters. Currently, the middleware applies globally. To apply per-route, you can use dependencies.

from fastapi import Depends
            
async def rate_limit(request: Request):
    # Custom logic
    pass

@app.get("/expensive", dependencies=[Depends(rate_limit)])
async def expensive_op():
    return {"status": "ok"}

*Note: Full per-route decorator support is coming in v0.2.0.*

4. Production with async Redis

For a multi-process deployment, use the async AsyncRedisStore. The FastAPI middleware awaits the limiter via limiter.acheck() internally, so an async store works end-to-end with no extra wiring.

import redis.asyncio as aioredis
from fastapi import FastAPI
from halt import RateLimiter, AsyncRedisStore, presets
from halt.adapters.fastapi import HaltMiddleware

app = FastAPI()

store = AsyncRedisStore(client=aioredis.Redis.from_url("redis://localhost:6379"))
limiter = RateLimiter(store=store, policy=presets.PUBLIC_API)

app.add_middleware(HaltMiddleware, limiter=limiter)

See the Redis store docs for fail-open/closed behavior, metrics, and cluster notes.

5. Checking limits yourself (async)

Outside the middleware, call await limiter.acheck(request) directly the async counterpart of check(). It uses an async store's aevaluate when present and falls back to the sync path otherwise. There's also a ready-made dependency factory:

from halt.adapters.fastapi import create_async_limiter_dependency

rate_limit = create_async_limiter_dependency(limiter)

@app.get("/data", dependencies=[Depends(rate_limit)])
async def data():
    return {"ok": True}