FastAPI Mastery

Role

You are an elite FastAPI architect. You design and implement high-performance async APIs

with proper dependency injection, middleware chains, streaming responses, WebSocket support,

and comprehensive test coverage using httpx.

Part 1: Application Structure

Production Layout

src/
  api/
    main.py              # App factory + startup
    dependencies.py      # Shared dependencies
    middleware.py         # Custom middleware
    routers/
      __init__.py
      users.py
      items.py
      websocket.py
    models/
      schemas.py          # Pydantic models
      database.py         # SQLAlchemy models
    services/
      user_service.py
      item_service.py
    core/
      config.py           # Settings
      security.py         # Auth utilities
tests/
  conftest.py
  test_users.py
  test_items.py

App Factory Pattern

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from .core.config import settings
from .routers import users, items, websocket
from .middleware import RequestLoggingMiddleware, RateLimitMiddleware


@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup
    from .models.database import engine, Base
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    # Shutdown
    await engine.dispose()


def create_app() -> FastAPI:
    app = FastAPI(
        title=settings.APP_NAME,
        version=settings.VERSION,
        lifespan=lifespan,
        docs_url="/docs" if settings.DEBUG else None,
    )

    # Middleware (order matters — last added = first executed)
    app.add_middleware(RequestLoggingMiddleware)
    app.add_middleware(RateLimitMiddleware, max_requests=100, window_seconds=60)
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.CORS_ORIGINS,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    # Routers
    app.include_router(users.router, prefix="/api/users", tags=["users"])
    app.include_router(items.router, prefix="/api/items", tags=["items"])
    app.include_router(websocket.router, prefix="/ws", tags=["websocket"])

    @app.get("/health")
    async def health():
        return {"status": "healthy", "version": settings.VERSION}

    return app

app = create_app()

Part 2: Pydantic v2 Models

from pydantic import BaseModel, Field, field_validator, model_validator
from datetime import datetime
from enum import Enum


class UserRole(str, Enum):
    ADMIN = "admin"
    USER = "user"
    VIEWER = "viewer"


class UserCreate(BaseModel):
    model_config = {"strict": True}

    email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    name: str = Field(..., min_length=2, max_length=100)
    role: UserRole = UserRole.USER
    age: int = Field(ge=13, le=150)

    @field_validator("name")
    @classmethod
    def normalize_name(cls, v: str) -> str:
        return v.strip().title()

    @field_validator("email")
    @classmethod
    def lowercase_email(cls, v: str) -> str:
        return v.lower()


class UserResponse(BaseModel):
    model_config = {"from_attributes": True}

    id: int
    email: str
    name: str
    role: UserRole
    created_at: datetime


class PaginatedResponse(BaseModel):
    items: list[UserResponse]
    total: int
    page: int
    per_page: int
    pages: int

    @model_validator(mode="after")
    def compute_pages(self) -> "PaginatedResponse":
        self.pages = (self.total + self.per_page - 1) // self.per_page
        return self

Part 3: Dependency Injection

from fastapi import Depends, HTTPException, Header, Request
from sqlalchemy.ext.asyncio import AsyncSession
from typing import Annotated


# Database session dependency
async def get_db() -> AsyncSession:
    async with async_session_factory() as session:
        try:
            yield session
        finally:
            await session.close()

DB = Annotated[AsyncSession, Depends(get_db)]


# Auth dependency
async def get_current_user(
    authorization: str = Header(..., alias="Authorization"),
    db: AsyncSession = Depends(get_db),
) -> User:
    if not authorization.startswith("Bearer "):
        raise HTTPException(401, "Invalid authorization header")
    token = authorization[7:]
    user = await verify_token(token, db)
    if not user:
        raise HTTPException(401, "Invalid or expired token")
    return user

CurrentUser = Annotated[User, Depends(get_current_user)]


# Role-based access
def require_role(*roles: str):
    async def check_role(user: CurrentUser):
        if user.role not in roles:
            raise HTTPException(403, f"Required role: {', '.join(roles)}")
        return user
    return Depends(check_role)


# Usage in router
@router.get("/admin")
async def admin_panel(user: CurrentUser = require_role("admin")):
    return {"message": f"Welcome admin {user.name}"}

Part 4: Middleware

import time
import logging
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response, JSONResponse
from collections import defaultdict

logger = logging.getLogger(__name__)


class RequestLoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        duration = time.perf_counter() - start

        logger.info(
            f"{request.method} {request.url.path} "
            f"status={response.status_code} "
            f"duration={duration:.3f}s"
        )

        response.headers["X-Process-Time"] = f"{duration:.3f}"
        return response


class RateLimitMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, max_requests: int = 100, window_seconds: int = 60):
        super().__init__(app)
        self.max_requests = max_requests
        self.window = window_seconds
        self.clients: dict[str, list[float]] = defaultdict(list)

    async def dispatch(self, request: Request, call_next):
        client_ip = request.client.host
        now = time.time()

        # Clean old entries
        self.clients[client_ip] = [
            t for t in self.clients[client_ip] if now - t < self.window
        ]

        if len(self.clients[client_ip]) >= self.max_requests:
            return JSONResponse(
                status_code=429,
                content={"detail": "Too many requests"},
                headers={"Retry-After": str(self.window)},
            )

        self.clients[client_ip].append(now)
        return await call_next(request)

Part 5: WebSocket

from fastapi import APIRouter, WebSocket, WebSocketDisconnect
from typing import Dict, Set
import json

router = APIRouter()


class ConnectionManager:
    def __init__(self):
        self.active: Dict[str, Set[WebSocket]] = {}  # room -> connections

    async def connect(self, websocket: WebSocket, room: str):
        await websocket.accept()
        if room not in self.active:
            self.active[room] = set()
        self.active[room].add(websocket)

    def disconnect(self, websocket: WebSocket, room: str):
        self.active.get(room, set()).discard(websocket)

    async def broadcast(self, room: str, message: dict):
        for conn in self.active.get(room, set()):
            try:
                await conn.send_json(message)
            except Exception:
                self.active[room].discard(conn)

manager = ConnectionManager()


@router.websocket("/{room}")
async def websocket_endpoint(websocket: WebSocket, room: str):
    await manager.connect(websocket, room)
    try:
        while True:
            data = await websocket.receive_json()
            await manager.broadcast(room, {
                "type": "message",
                "room": room,
                "data": data,
            })
    except WebSocketDisconnect:
        manager.disconnect(websocket, room)
        await manager.broadcast(room, {
            "type": "system",
            "data": "A user disconnected",
        })

Part 6: Streaming Responses

from fastapi import APIRouter
from fastapi.responses import StreamingResponse
import asyncio
import json

router = APIRouter()


@router.get("/stream/events")
async def server_sent_events():
    """SSE endpoint for real-time updates."""
    async def event_generator():
        while True:
            data = await get_latest_event()  # Your data source
            yield f"data: {json.dumps(data)}\n\n"
            await asyncio.sleep(1)

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@router.get("/stream/llm")
async def stream_llm_response(prompt: str):
    """Stream LLM response token by token."""
    async def generate():
        async for token in call_ollama_stream(prompt):
            yield f"data: {json.dumps({'token': token})}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")


async def call_ollama_stream(prompt: str):
    """Stream from Ollama API."""
    import httpx
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST", "http://localhost:11434/api/generate",
            json={"model": "qwen2.5:7b", "prompt": prompt},
            timeout=120,
        ) as response:
            async for line in response.aiter_lines():
                if line:
                    data = json.loads(line)
                    if not data.get("done"):
                        yield data.get("response", "")

Part 7: Background Tasks

from fastapi import BackgroundTasks, APIRouter
import logging

router = APIRouter()
logger = logging.getLogger(__name__)


async def send_notification(email: str, subject: str, body: str):
    """Background task: send email notification."""
    logger.info(f"Sending email to {email}: {subject}")
    # await email_client.send(to=email, subject=subject, body=body)


async def process_upload(file_id: str, file_path: str):
    """Background task: process uploaded file."""
    logger.info(f"Processing file {file_id}")
    # await run_ocr(file_path)
    # await generate_embeddings(file_id)
    # await update_status(file_id, "processed")


@router.post("/upload")
async def upload_file(background_tasks: BackgroundTasks):
    file_id = "abc123"
    file_path = "/tmp/upload.pdf"

    # Queue multiple background tasks
    background_tasks.add_task(process_upload, file_id, file_path)
    background_tasks.add_task(send_notification, "[email protected]", "Upload received", f"File {file_id}")

    return {"file_id": file_id, "status": "processing"}

Part 8: Testing with httpx

import pytest
from httpx import AsyncClient, ASGITransport
from src.api.main import create_app


@pytest.fixture
def app():
    return create_app()


@pytest.fixture
async def client(app):
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as ac:
        yield ac


@pytest.mark.asyncio
async def test_health(client: AsyncClient):
    response = await client.get("/health")
    assert response.status_code == 200
    assert response.json()["status"] == "healthy"


@pytest.mark.asyncio
async def test_create_user(client: AsyncClient):
    response = await client.post("/api/users", json={
        "email": "[email protected]",
        "name": "Test User",
        "role": "user",
        "age": 25,
    })
    assert response.status_code == 201
    data = response.json()
    assert data["email"] == "[email protected]"
    assert data["name"] == "Test User"


@pytest.mark.asyncio
async def test_rate_limiting(client: AsyncClient):
    # Exceed rate limit
    for _ in range(101):
        await client.get("/health")
    response = await client.get("/health")
    assert response.status_code == 429


@pytest.mark.asyncio
async def test_websocket(app):
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as client:
        async with client.websocket_connect("/ws/test-room") as ws:
            await ws.send_json({"message": "hello"})
            data = await ws.receive_json()
            assert data["type"] == "message"

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)