scaffolding-fastapi

name

scaffolding-fastapi

description

Build production-grade FastAPI backends with SQLModel, Pydantic settings, and JWT authentication. Use when building REST APIs with PostgreSQL, implementing environment-based configuration with .env files, creating CRUD endpoints with JWT/JWKS verification, or setting up pydantic-settings for app configuration. NOT when building distributed microservices with Dapr (use scaffolding-fastapi-dapr for that).

FastAPI + Pydantic Settings

Build production-grade FastAPI backends with SQLModel, Pydantic settings, and .env configuration management.

Quick Start

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

# Create .env file
cat > .env << 'EOF'
DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/mydb
AUTH_SERVER_URL=http://localhost:3001
JWT_ALGORITHM=RS256
DEBUG=false
LOG_LEVEL=INFO
EOF

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

# Access docs
open http://localhost:8000/docs

---

Core Concepts

Pydantic Settings for Configuration

Use pydantic-settings to manage environment variables with type validation:

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Literal

class Settings(BaseSettings):
    """Application settings with environment variable support."""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        env_prefix="APP_",  # Optional: prefix for env vars
        case_sensitive=False,
        extra="ignore",  # Ignore extra env vars
    )

    # Database
    database_url: str
    database_pool_size: int = 5
    database_max_overflow: int = 10

    # API
    api_host: str = "0.0.0.0"
    api_port: int = 8000
    api_title: str = "My API"
    api_version: str = "1.0.0"

    # Auth
    auth_server_url: str
    jwt_algorithm: Literal["HS256", "RS256"] = "RS256"
    jwt_secret: str | None = None  # For HS256

    # CORS
    allowed_origins: str = "http://localhost:3000"
    allowed_credentials: bool = True

    # Environment
    environment: Literal["development", "staging", "production"] = "development"
    debug: bool = False
    log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"

    @property
    def is_production(self) -> bool:
        return self.environment == "production"

    @property
    def is_development(self) -> bool:
        return self.environment == "development"

    @property
    def cors_origins(self) -> list[str]:
        return [o.strip() for o in self.allowed_origins.split(",")]

# Global settings instance
settings = Settings()

Using Settings in Your App

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.config import settings

# Create async engine from settings
async_engine = create_async_engine(
    settings.database_url,
    pool_size=settings.database_pool_size,
    max_overflow=settings.database_max_overflow,
    echo=settings.debug,  # Enable SQL logging in debug mode
)

async_session_maker = sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)

async def get_session() -> AsyncSession:
    """FastAPI dependency for database sessions."""
    async with async_session_maker() as session:
        yield session

# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.config import settings

app = FastAPI(
    title=settings.api_title,
    version=settings.api_version,
    debug=settings.debug,
)

# CORS from settings
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_credentials=settings.allowed_credentials,
    allow_methods=["*"],
    allow_headers=["*"],
)

Environment Variable Reference

.env File Format

# Database Configuration
DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/mydb
DATABASE_POOL_SIZE=5
DATABASE_MAX_OVERFLOW=10

# API Configuration
API_HOST=0.0.0.0
API_PORT=8000
API_TITLE=My API
API_VERSION=1.0.0

# Authentication
AUTH_SERVER_URL=http://localhost:3001
JWT_ALGORITHM=RS256
# JWT_SECRET=your-secret-key-here  # Only for HS256

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

# Environment
ENVIRONMENT=development
DEBUG=false
LOG_LEVEL=INFO

Environment-Specific Files

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
import os

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=f".env.{os.getenv('ENV', 'development')}",
        env_file_encoding="utf-8",
    )
    # ... same settings as above

# Usage:
# ENV=production uv run uvicorn main:app
# Loads from .env.production

---

FastAPI Core Patterns

1. SQLModel Schema (Database + API)

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

class TaskBase(SQLModel):
    title: str = Field(max_length=200, index=True)
    description: Optional[str] = None
    status: Literal["pending", "in_progress", "completed"] = "pending"
    priority: Literal["low", "medium", "high"] = "medium"

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)

class TaskCreate(TaskBase):
    pass

class TaskUpdate(SQLModel):
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None

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

2. CRUD Endpoints

from fastapi import FastAPI, Depends, HTTPException, Query
from sqlmodel import select, AsyncSession
from app.database import get_session
from app.models import Task, TaskCreate, TaskUpdate, TaskRead

@app.post("/api/tasks", response_model=TaskRead, status_code=201)
async def create_task(
    task: TaskCreate,
    session: AsyncSession = Depends(get_session),
):
    db_task = Task.model_validate(task)
    session.add(db_task)
    await session.commit()
    await session.refresh(db_task)
    return db_task

@app.get("/api/tasks", response_model=list[TaskRead])
async def list_tasks(
    session: AsyncSession = Depends(get_session),
    status: 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)
    query = query.offset(offset).limit(limit)
    result = await session.exec(query)
    return result.all()

@app.get("/api/tasks/{task_id}", response_model=TaskRead)
async def get_task(
    task_id: int,
    session: AsyncSession = Depends(get_session),
):
    task = await 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)
async def update_task(
    task_id: int,
    update: TaskUpdate,
    session: AsyncSession = Depends(get_session),
):
    task = await session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    update_data = update.model_dump(exclude_unset=True)
    task.sqlmodel_update(update_data)
    task.updated_at = datetime.now()

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

@app.delete("/api/tasks/{task_id}")
async def delete_task(
    task_id: int,
    session: AsyncSession = Depends(get_session),
):
    task = await session.get(Task, task_id)
    if not task:
        raise HTTPException(status_code=404, detail="Task not found")

    await session.delete(task)
    await session.commit()
    return {"deleted": task_id}

3. JWT/JWKS Authentication

# app/auth.py
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, jwk, JWTError
import httpx
import time
from app.config import settings

security = HTTPBearer()

_jwks_cache = None
_jwks_cache_time = 0
JWKS_CACHE_TTL = 3600  # 1 hour

async def get_jwks():
    """Fetch and cache JWKS from 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

    jwks_url = f"{settings.auth_server_url}/.well-known/jwks.json"
    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 auth server JWKS."""
    try:
        jwks = await get_jwks()
        unverified_header = jwt.get_unverified_header(token)
        kid = unverified_header.get("kid")

        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")

        payload = jwt.decode(
            token,
            rsa_key,
            algorithms=[settings.jwt_algorithm],
            options={"verify_aud": False},
        )
        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"),
    }

4. Health Check Endpoints

from app.database import async_engine
from app.config import settings

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

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

---

Pydantic Settings Patterns

Required vs Optional Fields

from pydantic_settings import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    # Required - will raise ValidationError if not set
    database_url: str

    # Optional with default
    api_port: int = 8000
    debug: bool = False

    # Optional without default (None)
    jwt_secret: Optional[str] = None

    @property
    def is_secure(self) -> bool:
        return not self.debug and self.jwt_secret is not None

Field Validation

from pydantic import field_validator

class Settings(BaseSettings):
    port: int = 8000

    @field_validator("port")
    @classmethod
    def port_in_range(cls, v: int) -> int:
        if not (1 <= v <= 65535):
            raise ValueError("Port must be between 1 and 65535")
        return v

    database_url: str

    @field_validator("database_url")
    @classmethod
    def ensure_async_driver(cls, v: str) -> str:
        if "postgresql://" in v and "postgresql+asyncpg://" not in v:
            v = v.replace("postgresql://", "postgresql+asyncpg://")
        return v

Multiple Environments

# .env.development
DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/mydb_dev
DEBUG=true
LOG_LEVEL=DEBUG

# .env.staging
DATABASE_URL=postgresql+asyncpg://user:pass@staging-db:5432/mydb_stg
DEBUG=false
LOG_LEVEL=INFO

# .env.production
DATABASE_URL=postgresql+asyncpg://user:pass@prod-db:5432/mydb
DEBUG=false
LOG_LEVEL=WARNING

# Usage:
# uv run --environment production uvicorn main:app

Secrets Management

from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    # Use .secrets file for sensitive values
    # This file should be in .gitignore
    model_config = SettingsConfigDict(
        env_file=[".env", ".secrets"],
        env_file_encoding="utf-8",
    )

    # Sensitive values
    database_password: str
    jwt_secret: str
    api_key: str

# .gitignore
.env.local
.env.*.local
.secrets

---

Project Structure

backend/
├── app/
│   ├── __init__.py
│   ├── config.py          # Pydantic Settings class
│   ├── main.py            # FastAPI app
│   ├── database.py        # Async engine + session
│   ├── auth.py            # JWT/JWKS verification
│   ├── models/            # SQLModel schemas
│   ├── routers/           # API routes
│   ├── repositories/      # Data access layer
│   └── services/          # Business logic
├── tests/
│   ├── conftest.py        # Test fixtures
│   └── test_*.py          # Test files
├── .env.example           # Template for env vars
├── .env                   # Local development (gitignored)
├── .env.production        # Production (gitignored)
├── pyproject.toml
└── README.md

---

Complete Example

# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from typing import Literal

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        case_sensitive=False,
    )

    database_url: str
    auth_server_url: str
    api_title: str = "My API"
    api_version: str = "1.0.0"
    debug: bool = False

    @property
    def cors_origins(self) -> list[str]:
        return ["http://localhost:3000"] if self.debug else []

settings = Settings()

# app/database.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from app.config import settings

async_engine = create_async_engine(settings.database_url, echo=settings.debug)
async_session_maker = sessionmaker(async_engine, class_=AsyncSession, expire_on_commit=False)

async def get_session() -> AsyncSession:
    async with async_session_maker() as session:
        yield session

# app/main.py
from fastapi import FastAPI
from app.config import settings
from app.database import async_session_maker
from sqlmodel import SQLModel

app = FastAPI(
    title=settings.api_title,
    version=settings.api_version,
    debug=settings.debug,
)

@app.on_event("startup")
async def startup():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

@app.get("/")
async def root():
    return {"name": settings.api_title, "version": settings.api_version}

---

Verification

Run: python3 scripts/verify.py

Expected: ✓ scaffolding-fastapi skill ready

If Verification Fails

1. Check: references/ folder has all pattern files
2. Check: examples/ folder has config examples
3. Stop and report if still failing

Related Skills

• configuring-better-auth - JWT/JWKS auth for API endpoints
• fetching-library-docs - FastAPI docs: --library-id /fastapi/fastapi --topic dependencies

References

• references/pydantic-settings.md - Pydantic Settings patterns and .env handling
• references/fastapi-patterns.md - Complete FastAPI backend patterns
• references/sqlmodel-patterns.md - SQLModel database patterns and migrations
• references/production-testing.md - Structured logging, DI, testing, versioning