Module 05

Advanced FastAPI

From CRUD developer to production-ready engineer. Mastering WebSockets, Caching, and Performance Tuning.

Middleware

⚙️

What is Middleware?

Middleware is code that runs before and after every request. It allows you to process requests before they reach your path operations and modify responses before they leave.

Logging

Authentication Check

Adding Headers

Request Tracking

Basic Middleware Example

python

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    print(f"Incoming request: {request.url}")
    
    response = await call_next(request)
    
    print(f"Response status: {response.status_code}")
    return response

Background Tasks

👉 Response is sent immediately, task runs later → improves system perceived performance.

🚀 When to use?

✔ Email Sending
✔ Notification Triggers
✔ Heavy Logging
✔ ML Post-processing

Background Task Implementation

python

from fastapi import BackgroundTasks

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(message + "\n")

@app.post("/process")
def process(data: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, data)
    return {"msg": "Processing started"}

WebSockets (Real-Time Apps)

A protocol for real-time, bidirectional communication between client and server. Unlike HTTP, it maintains a continuous stateful connection.

HTTP Protocol

Request-Response model. Stateless connection.

WebSocket Protocol

Continuous connection. Stateful & Bidirectional.

Real-time Endpoint

python

from fastapi import WebSocket

@app.websocket("/ws") async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    
    while True:
        data = await ws.receive_text()
        await ws.send_text(f"Message: {data}")

Caching (VERY IMPORTANT)

Why Memory Matters?

Caching stores frequently used data in fast storage (In-memory or Redis) to reduce database load and improve response latency.

🧠 Simple In-Memory Caching

Dict Cache

python

cache = {}

@app.get("/data")
def get_data():
    if "data" in cache:
        return cache["data"]
    
    result = {"value": "Expensive Data"}
    cache["data"] = result
    return result

🔌 Redis (Production Standard)

Used for fast reads, session storage, and global rate limiting across clusters.

Fast ReadsSession StoreRate LimitingGlobal Access

Advanced Dependency Injection

FastAPI's powerful DI system ensures cleaner code, reusability, and easy testability by decoupling logic from execution.

Reusable Logic

python

def get_token(token: str):
    return token

@app.get("/items")
def read_items(token: str = Depends(get_token)):
    return {"token": token}

Performance Optimization

Use Async Always

Non-blocking I/O operations.

@app.get("/async") async def func():

Connection Pooling

Reduce connection overhead (asyncpg).

AsyncSessionLocal = sessionmaker(...)

Leverage Caching

Reduce DB pressure significantly.

cache = get_redis_cache()

Avoid Blocking Code

Use await asyncio.sleep over time.sleep.

await asyncio.sleep(5)

🚀 Production Server Tip

Terminal

bash

gunicorn -k uvicorn.workers.UvicornWorker main:app

Rate Limiting (Advanced Concept)

🛡️

Why Rate Limit?

Prevents API abuse and protects against DDoS attacks by limiting the number of requests a specific user or IP can make in a given timeframe (e.g., 100 requests/minute).

Custom Exception Handling

⚠️ Standard Exception

HTTP Error

python

from fastapi import HTTPException

@app.get("/error")
def error():
    raise HTTPException(status_code=404, detail="Not Found")

🌍 Global Handler

Middleware Handler

python

from fastapi.responses import JSONResponse

@app.exception_handler(Exception)
async def global_exception(request, exc):
    return JSONResponse(status_code=500, content={"msg": "Something went wrong"})

Streaming Response (Advanced)

Useful for sending large data chunks or real-time generator outputs without consuming massive server memory as it sends data piece by piece.

Generator Stream

python

from fastapi.responses import StreamingResponse

def generate():
    for i in range(5):
        yield f"{i}\n"

@app.get("/stream")
def stream():
    return StreamingResponse(generate())

Lifespan Events (Startup / Shutdown)

Manage your resources (DB connections, ML models, Cache clients) globally.

Events.py

python

@app.on_event("startup")
def startup():
    print("App started - DB Connected")

@app.on_event("shutdown")
def shutdown():
    print("App stopped - DB Cleaned up")

WebSocket + Auth (Advanced Insight)

🛡️ Security in Real-time

Combine WebSockets with JWT for maximum security. Always authenticate before accepting the connection to prevent unauthorized socket usage.

Interview Preparation

🧠 Theory Questions

What is middleware?▼

Code executed before and after every request, useful for logging, headers, and global auth.

Background tasks benefits?▼

Running non-blocking tasks (emails, notifications) after sending responses to improve user UX.

WebSocket pros?▼

Real-time, stateful, and bidirectional communication between client and server.

Why use caching?▼

Improve system performance, reduce database load, and lower response latency.

Sync vs Async performance?▼

Async handles high concurrency much better by not blocking threads during I/O waits.

Dependency Injection?▼

Decouples reusable logic (DB sessions, auth) using FastAPI's Depends system.

How to optimize performace?▼

Async code, Caching (Redis), Connection Pooling, and Gunicorn workers.

💻 Coding Challenges

Q1: Middleware

Challenge Snippet

python

@app.middleware("http")\nasync def custom(request, call_next):\n    return await call_next(request)

Q2: Background Task

Challenge Snippet

python

@app.post("/task")\ndef task(bg: BackgroundTasks):\n    bg.add_task(print, "Log")\n    return {"msg": "done"}

Q3: WebSocket

Challenge Snippet

python

@app.websocket("/ws")\nasync def ws(socket: WebSocket):\n    await socket.accept()

Q4: Simple Cache

Challenge Snippet

python

cache = {}\nif "data" in cache: return cache["data"]\ncache["data"] = compute()