FastAPI Best Practices

Production-grade patterns for FastAPI applications. Covers API design, async patterns, streaming, security, and performance.

When to Apply

• Creating new API endpoints
• Implementing SSE streaming
• Adding middleware or error handling
• Integrating with databases (asyncpg, SQLAlchemy)
• Implementing authentication/authorization

1. API Design (CRITICAL)

Use Pydantic models for request/response

class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=5000)
    mode: Literal["smalltalk", "technical_analysis", "best_setups"]
    language: str = "en"

class ChatResponse(BaseModel):
    response: str
    tokens_used: int
    cached: bool = False

Proper status codes

• 200: Success
• 201: Created
• 400: Bad request (validation error)
• 401: Unauthorized
• 403: Forbidden (quota exceeded)
• 404: Not found
• 429: Rate limited
• 500: Internal server error

Consistent error responses

class ErrorResponse(BaseModel):
    error: str
    detail: str | None = None
    code: str

@app.exception_handler(AppError)
async def app_error_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content=ErrorResponse(error=exc.message, code=exc.code).model_dump()
    )

2. SSE Streaming (CRITICAL for chat)

Proper streaming pattern

@router.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    async def event_generator():
        try:
            async for chunk in process_chat(request):
                yield f"data: {json.dumps(chunk)}\n\n"
            yield "data: [DONE]\n\n"
        except Exception as e:
            logger.error(f"Stream error: {e}")
            yield f"data: {json.dumps({'error': str(e)})}\n\n"
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"
        }
    )

Stream timeout handling

• Set per-chunk timeouts, not global
• Yield keepalive comments every 15s
• Handle client disconnection gracefully

3. Async Database Patterns (HIGH)

Connection pooling

# Startup
@app.on_event("startup")
async def startup():
    app.state.db_pool = await asyncpg.create_pool(
        dsn=settings.database_url,
        min_size=5,
        max_size=20,
        command_timeout=30
    )

# Usage
async def get_market_data(pool, symbol: str):
    async with pool.acquire() as conn:
        return await conn.fetch(
            "SELECT * FROM market_data WHERE symbol = $1 ORDER BY time DESC LIMIT 100",
            symbol
        )

Transaction management

async with pool.acquire() as conn:
    async with conn.transaction():
        await conn.execute(insert_query, *params)
        await conn.execute(update_query, *params)

4. Middleware (HIGH)

Request timing

@app.middleware("http")
async def timing_middleware(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.3f}"
    if duration > 5.0:
        logger.warning(f"Slow request: {request.url.path} took {duration:.2f}s")
    return response

CORS configuration

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.allowed_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

5. Background Tasks (MEDIUM)

Use BackgroundTasks for fire-and-forget

@router.post("/analyze")
async def analyze(request: AnalysisRequest, background_tasks: BackgroundTasks):
    result = await quick_analysis(request)
    background_tasks.add_task(save_analysis_log, request, result)
    return result

Use scheduler for periodic tasks

• APScheduler for periodic data collection
• Separate scheduler from API server
• Health checks for scheduler status

6. Security (HIGH)

Rate limiting

from slowapi import Limiter
limiter = Limiter(key_func=get_remote_address)

@router.post("/chat")
@limiter.limit("30/minute")
async def chat(request: Request):
    ...

Input validation

• Validate all inputs with Pydantic
• Set max lengths on string fields
• Validate numeric ranges
• Sanitize before database queries

Authentication

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
        if user_id is None:
            raise HTTPException(status_code=401)
        return await get_user(user_id)
    except JWTError:
        raise HTTPException(status_code=401)

7. Deployment

• Use uvicorn with multiple workers
• Configure graceful shutdown
• Health check endpoint at /health
• Structured JSON logging
• Monitor with Prometheus metrics