FastAPI Backend

Build production-grade FastAPI backends with SQLModel, Pydantic v2, and JWT/JWKS authentication patterns.

When to Use

• Building REST API endpoints with FastAPI
• Creating SQLModel schemas for Neon PostgreSQL
• Implementing JWT verification against Better Auth JWKS
• Designing OpenAPI contracts for frontend consumption
• Adding audit logging to API operations
• Ensuring human-agent parity in API design

Quick Start

# Project setup
uv init backend && cd backend
uv add fastapi sqlmodel pydantic httpx python-jose uvicorn

# Development
uv run uvicorn main:app --reload --port 8000

# Access docs
open http://localhost:8000/docs  # Swagger UI with Authorize button

Core Patterns

1. SQLModel Schema (Database + API)

SQLModel combines SQLAlchemy and Pydantic. Use table=True for database models:

from sqlmodel import SQLModel, Field
from datetime import datetime
from typing import Optional, Literal

# Base model (shared fields, no table)
class TaskBase(SQLModel):
    title: str = Field(max_length=200)
    description: Optional[str] = None
    status: Literal["pending", "in_progress", "review", "completed", "blocked"] = "pending"
    priority: Literal["low", "medium", "high", "critical"] = "medium"
    progress_percent: int = Field(default=0, ge=0, le=100)
    assigned_to: Optional[str] = None
    project_slug: Optional[str] = None
    parent_id: Optional[int] = None

# Database model (has table)
class Task(TaskBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    created_at: datetime = Field(default_factory=datetime.now)
    updated_at: datetime = Field(default_factory=datetime.now)

# API models (no table, for request/response)
class TaskCreate(TaskBase):
    pass

class TaskUpdate(SQLModel):
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None
    priority: Optional[str] = None
    progress_percent: Optional[int] = None
    assigned_to: Optional[str] = None

class TaskRead(TaskBase):
    id: int
    created_at: datetime
    updated_at: datetime

2. Neon PostgreSQL Connection

from sqlmodel import create_engine, Session
import os

# Neon connection string
DATABASE_URL = os.getenv("DATABASE_URL")  # postgresql://user:pass@host/db?sslmode=require

engine = create_engine(DATABASE_URL, echo=True)

def get_session():
    with Session(engine) as session:
        yield session

3. CRUD Endpoints

from fastapi import FastAPI, Depends, HTTPException, Query
from sqlmodel import Session, select

app = FastAPI(title="TaskFlow API", version="1.0.0")

@app.post("/api/tasks", response_model=TaskRead, status_code=201)
def create_task(
    task: TaskCreate,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
):
    db_task = Task.model_validate(task)
    session.add(db_task)
    session.commit()
    session.refresh(db_task)

    # Audit log
    log_action(session, "created", current_user.id, task_id=db_task.id)

    return db_task

@app.get("/api/tasks", response_model=list[TaskRead])
def list_tasks(
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
    status: Optional[str] = Query(None),
    assigned_to: Optional[str] = Query(None),
    project: Optional[str] = Query(None),
    limit: int = Query(50, le=100),
    offset: int = Query(0, ge=0),
):
    query = select(Task)

    if status:
        query = query.where(Task.status == status)
    if assigned_to:
        query = query.where(Task.assigned_to == assigned_to)
    if project:
        query = query.where(Task.project_slug == project)

    query = query.offset(offset).limit(limit)
    return session.exec(query).all()

@app.get("/api/tasks/{task_id}", response_model=TaskRead)
def get_task(
    task_id: int,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")
    return task

@app.patch("/api/tasks/{task_id}", response_model=TaskRead)
def update_task(
    task_id: int,
    task_update: TaskUpdate,
    session: Session = Depends(get_session),
    current_user: User = Depends(get_current_user),
):
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    update_data = task_update.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(task, key, value)

    task.updated_at = datetime.now()
    session.add(task)
    session.commit()
    session.refresh(task)

    # Audit log
    log_action(session, "updated", current_user.id, task_id=task.id, context=update_data)

    return task

4. JWT Authentication (Better Auth JWKS)

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, jwk, JWTError
import httpx
import time

security = HTTPBearer()

# Cache JWKS keys
_jwks_cache = None
_jwks_cache_time = 0
JWKS_CACHE_TTL = 3600  # 1 hour

AUTH_SERVER_URL = os.getenv("AUTH_SERVER_URL", "http://localhost:3001")
JWKS_URL = f"{AUTH_SERVER_URL}/api/auth/jwks"

async def get_jwks():
    """Fetch and cache JWKS from Better Auth server."""
    global _jwks_cache, _jwks_cache_time

    now = time.time()
    if _jwks_cache and (now - _jwks_cache_time) < JWKS_CACHE_TTL:
        return _jwks_cache

    async with httpx.AsyncClient() as client:
        response = await client.get(JWKS_URL)
        response.raise_for_status()
        _jwks_cache = response.json()
        _jwks_cache_time = now
        return _jwks_cache

async def verify_token(token: str) -> dict:
    """Verify JWT against Better Auth JWKS."""
    try:
        # Get JWKS
        jwks = await get_jwks()

        # Get unverified header to find key ID
        unverified_header = jwt.get_unverified_header(token)
        kid = unverified_header.get("kid")

        # Find matching key
        rsa_key = None
        for key in jwks.get("keys", []):
            if key.get("kid") == kid:
                rsa_key = key
                break

        if not rsa_key:
            raise HTTPException(status_code=401, detail="Key not found")

        # Verify token
        payload = jwt.decode(
            token,
            rsa_key,
            algorithms=["RS256"],
            options={"verify_aud": False}  # Adjust based on your setup
        )
        return payload

    except JWTError as e:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail=f"Invalid token: {str(e)}",
            headers={"WWW-Authenticate": "Bearer"},
        )

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
) -> dict:
    """Extract and verify current user from JWT."""
    token = credentials.credentials
    payload = await verify_token(token)

    return {
        "id": payload.get("sub"),
        "email": payload.get("email"),
        "role": payload.get("role", "user"),
    }

5. Swagger UI Authentication

FastAPI automatically adds an "Authorize" button when using HTTPBearer:

from fastapi import FastAPI
from fastapi.security import HTTPBearer

app = FastAPI(
    title="TaskFlow API",
    description="Human-Agent Task Management API",
    version="1.0.0",
)

# This adds the "Authorize" button to Swagger UI
security = HTTPBearer()

# Testing flow:
# 1. Login to your SSO (browser) → Get JWT
# 2. Open http://localhost:8000/docs
# 3. Click "Authorize" button
# 4. Paste JWT token (without "Bearer " prefix)
# 5. All requests now include Authorization header

6. Audit Logging

from sqlalchemy import JSON

class AuditLog(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    task_id: Optional[int] = None
    project_slug: Optional[str] = None
    actor_id: str
    actor_type: Literal["human", "agent"]
    action: str
    context: Optional[dict] = Field(default=None, sa_column_kwargs={"type_": JSON})
    timestamp: datetime = Field(default_factory=datetime.now)

def log_action(
    session: Session,
    action: str,
    actor_id: str,
    task_id: Optional[int] = None,
    project_slug: Optional[str] = None,
    context: Optional[dict] = None,
):
    """Create audit log entry."""
    # Determine actor type from worker registry
    worker = session.exec(select(Worker).where(Worker.id == actor_id)).first()
    actor_type = worker.type if worker else "human"

    log = AuditLog(
        task_id=task_id,
        project_slug=project_slug,
        actor_id=actor_id,
        actor_type=actor_type,
        action=action,
        context=context,
    )
    session.add(log)
    session.commit()
    return log

7. Agent Parity (MCP Compatibility)

Design endpoints that work for both CLI and MCP clients:

# Same endpoint serves:
# - CLI: taskflow start 1 → POST /api/tasks/1/start
# - MCP: claim_task(1) → POST /api/tasks/1/start
# - Web: Button click → POST /api/tasks/1/start

@app.post("/api/tasks/{task_id}/start", response_model=TaskRead)
def start_task(
    task_id: int,
    session: Session = Depends(get_session),
    current_user: dict = Depends(get_current_user),
):
    """Start a task (claim and begin work)."""
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    if task.status != "pending":
        raise HTTPException(
            status_code=400,
            detail=f"Cannot start task with status '{task.status}'"
        )

    task.status = "in_progress"
    task.assigned_to = current_user["id"]
    task.updated_at = datetime.now()

    session.add(task)
    session.commit()
    session.refresh(task)

    log_action(session, "started", current_user["id"], task_id=task.id)

    return task

@app.post("/api/tasks/{task_id}/progress", response_model=TaskRead)
def update_progress(
    task_id: int,
    percent: int = Query(..., ge=0, le=100),
    note: Optional[str] = Query(None),
    session: Session = Depends(get_session),
    current_user: dict = Depends(get_current_user),
):
    """Update task progress."""
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    if task.status != "in_progress":
        raise HTTPException(status_code=400, detail="Task must be in_progress")

    task.progress_percent = percent
    task.updated_at = datetime.now()

    session.add(task)
    session.commit()
    session.refresh(task)

    log_action(
        session, "progressed", current_user["id"],
        task_id=task.id,
        context={"percent": percent, "note": note}
    )

    return task

@app.post("/api/tasks/{task_id}/complete", response_model=TaskRead)
def complete_task(
    task_id: int,
    session: Session = Depends(get_session),
    current_user: dict = Depends(get_current_user),
):
    """Complete a task."""
    task = session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    if task.status not in ["in_progress", "review"]:
        raise HTTPException(
            status_code=400,
            detail=f"Cannot complete task with status '{task.status}'"
        )

    task.status = "completed"
    task.progress_percent = 100
    task.updated_at = datetime.now()

    session.add(task)
    session.commit()
    session.refresh(task)

    log_action(session, "completed", current_user["id"], task_id=task.id)

    return task

Project Structure

backend/
├── main.py              # FastAPI app, routes
├── models.py            # SQLModel schemas
├── database.py          # Neon connection
├── auth.py              # JWT/JWKS verification
├── audit.py             # Audit logging
├── dependencies.py      # Shared dependencies
└── tests/
    ├── conftest.py      # Test fixtures
    ├── test_tasks.py    # Task endpoint tests
    └── test_auth.py     # Auth tests

Critical: Async Session Patterns (MissingGreenlet Prevention)

After session.commit(), SQLAlchemy objects become detached. Accessing attributes triggers lazy loading which fails in async context with MissingGreenlet error.

The Pattern: Extract → Flush → Commit

@app.post("/api/tasks", response_model=TaskRead, status_code=201)
async def create_task(
    task: TaskCreate,
    session: AsyncSession = Depends(get_session),
    current_user: dict = Depends(get_current_user),
):
    # 1. Extract primitives from user BEFORE any commits
    actor_id = current_user["id"]

    # 2. Create entity and flush to get ID
    db_task = Task.model_validate(task)
    session.add(db_task)
    await session.flush()
    task_id = db_task.id  # Extract immediately after flush

    # 3. Call services with primitives (NOT objects)
    await log_action(session, actor_id=actor_id, task_id=task_id)

    # 4. Single commit at end
    await session.commit()
    await session.refresh(db_task)
    return db_task

Service Functions: Never Commit Internally

# WRONG - breaks caller's transaction
async def log_action(session: AsyncSession, ...):
    log = AuditLog(...)
    session.add(log)
    await session.commit()  # ❌ Caller loses control

# CORRECT - caller owns transaction
async def log_action(session: AsyncSession, ...):
    log = AuditLog(...)
    session.add(log)
    return log  # No commit

Input Validation for API Schemas

from pydantic import field_validator
from datetime import UTC, datetime

class TaskCreate(SQLModel):
    assignee_id: int | None = None
    due_date: datetime | None = None

    @field_validator("assignee_id", mode="after")
    @classmethod
    def zero_to_none(cls, v: int | None) -> int | None:
        """Swagger UI sends 0 for empty int fields."""
        return None if v == 0 else v

    @field_validator("due_date", mode="after")
    @classmethod
    def normalize_datetime(cls, v: datetime | None) -> datetime | None:
        """Strip timezone for naive UTC database columns."""
        if v and v.tzinfo:
            return v.astimezone(UTC).replace(tzinfo=None)
        return v

Testing with Pytest

# tests/conftest.py
import pytest
from fastapi.testclient import TestClient
from sqlmodel import SQLModel, create_engine, Session
from sqlmodel.pool import StaticPool

from main import app, get_session

@pytest.fixture
def session():
    engine = create_engine(
        "sqlite://",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    SQLModel.metadata.create_all(engine)
    with Session(engine) as session:
        yield session

@pytest.fixture
def client(session):
    def get_session_override():
        return session

    app.dependency_overrides[get_session] = get_session_override
    client = TestClient(app)
    yield client
    app.dependency_overrides.clear()

@pytest.fixture
def auth_headers():
    """Mock authenticated headers for testing."""
    return {"Authorization": "Bearer test-token"}

# tests/test_tasks.py
def test_create_task(client, auth_headers, mocker):
    # Mock auth
    mocker.patch("auth.get_current_user", return_value={"id": "@testuser"})

    response = client.post(
        "/api/tasks",
        json={"title": "Test Task"},
        headers=auth_headers,
    )
    assert response.status_code == 201
    assert response.json()["title"] == "Test Task"

Common Patterns

Error Handling

from fastapi import HTTPException
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": exc.detail,
            "status_code": exc.status_code,
        },
    )

# Validation errors are handled automatically by Pydantic
# Returns 422 with detailed error messages

CORS Configuration

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=os.getenv("ALLOWED_ORIGINS", "http://localhost:3000").split(","),
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Health Check

@app.get("/health")
def health_check():
    return {"status": "healthy", "version": "1.0.0"}

@app.get("/api/health/db")
def db_health(session: Session = Depends(get_session)):
    try:
        session.exec(select(1))
        return {"database": "connected"}
    except Exception as e:
        raise HTTPException(status_code=503, detail=f"Database error: {e}")

References

For additional documentation, use Context7 MCP:

mcp__context7__resolve-library-id with libraryName="fastapi"
mcp__context7__get-library-docs with topic="authentication" or "sqlmodel"

See also: references/jwt-verification.md for detailed JWKS patterns.

Environment Variables

# Database
DATABASE_URL=postgresql://user:pass@host/db?sslmode=require

# Auth
AUTH_SERVER_URL=http://localhost:3001
JWKS_URL=http://localhost:3001/api/auth/jwks

# CORS
ALLOWED_ORIGINS=http://localhost:3000,http://localhost:3001

# Optional
LOG_LEVEL=INFO