fastapi-fastmcp-asyncpq-alembic-guide.md

fastmcp-integration-guide.md

FastMCP Integration Guide

How to build MCP servers with FastMCP 3.x and mount them inside FastAPI.

Reference: The complete FastMCP documentation index is available at https://gofastmcp.com/llms.txt. Fetch it to discover all available pages, API references, and integration guides.

---

1. What is MCP + FastMCP

MCP (Model Context Protocol) is a standard protocol for LLMs to call external tools, read resources, and execute code. Clients include Claude Code, Claude Desktop, Cursor, ChatGPT, and Gemini CLI.

FastMCP is the Python framework for building MCP servers. It handles:

• Tool schema generation from Python type hints
• stdio transport (local — Claude Code on same machine)
• HTTP transport (remote — agents over the network)
• Parameter validation (Pydantic)
• Auth, middleware, pagination

You write Python functions. FastMCP exposes them to LLMs.

---

2. Installation

pip install fastmcp

FastMCP does NOT include FastAPI — install separately if mounting:

pip install fastmcp fastapi uvicorn

---

3. Creating Tools

from fastmcp import FastMCP

mcp = FastMCP("My Server")

@mcp.tool
def add(a: int, b: int) -> int:
    """Adds two integer numbers together."""
    return a + b

FastMCP auto-generates:

• Tool name from the function name (add)
• Description from the docstring
• Input schema from type annotations
• Output schema from the return type annotation

Async tools

@mcp.tool
async def fetch_user(user_id: str) -> dict:
    """Fetch a user from the database."""
    return await db.fetchrow("SELECT * FROM users WHERE id = $1", user_id)

Sync tools run in a threadpool automatically — they don't block the event loop.

Typed parameters with descriptions

from typing import Annotated
from pydantic import Field

@mcp.tool
def search(
    query: Annotated[str, Field(description="Search query")],
    limit: Annotated[int, Field(description="Max results", ge=1, le=100)] = 10,
) -> list[dict]:
    """Search the database."""
    ...

Tool annotations (hints for LLM clients)

@mcp.tool(annotations={
    "readOnlyHint": True,       # tool doesn't modify state
    "destructiveHint": False,   # not destructive
    "idempotentHint": True,     # repeatable with same result
    "openWorldHint": False,     # only interacts with internal data
})
def get_user(user_id: str) -> dict:
    """Retrieve user information."""
    ...

Clients like Claude and ChatGPT use these hints to skip confirmation prompts for read-only tools.

Hiding parameters from the LLM

Inject runtime values (user IDs, DB sessions, credentials) via Depends(). These are excluded from the tool schema — the LLM never sees them:

from fastmcp.dependencies import Depends

def get_current_user() -> str:
    return "user_123"  # resolved at runtime

@mcp.tool
def my_data(user_id: str = Depends(get_current_user)) -> dict:
    """Get data for the current user."""
    return {"user": user_id}

Error handling

Raise any exception. FastMCP converts it to an MCP error response:

from fastmcp.exceptions import ToolError

@mcp.tool
def divide(a: float, b: float) -> float:
    """Divide a by b."""
    if b == 0:
        raise ToolError("Division by zero is not allowed.")
    return a / b

ToolError messages are always sent to the client. Other exception details can be masked in production:

mcp = FastMCP("Secure Server", mask_error_details=True)

---

4. Running the Server

stdio (local — Claude Code on same machine)

if __name__ == "__main__":
    mcp.run()  # defaults to stdio

Claude Code connects via stdin/stdout. No network.

HTTP (remote — agents over the network)

Direct (simple):

if __name__ == "__main__":
    mcp.run(transport="http", host="0.0.0.0", port=8000)

Server accessible at http://localhost:8000/mcp.

ASGI app (production — more control):

app = mcp.http_app()

uvicorn server:app --host 0.0.0.0 --port 8000

---

5. Mounting MCP Inside FastAPI

This is the main pattern for our project — serve REST API + MCP from the same application:

from fastapi import FastAPI
from fastmcp import FastMCP

# Your MCP server
mcp = FastMCP("Analytics Tools")

@mcp.tool
def analyze_data(query: str) -> dict:
    """Analyze data."""
    return {"result": "..."}

# Create MCP's ASGI app
mcp_app = mcp.http_app(path="/")

# Create FastAPI app with MCP's lifespan (REQUIRED)
app = FastAPI(lifespan=mcp_app.lifespan)

@app.get("/api/status")
def status():
    return {"status": "ok"}

# Mount MCP at /mcp
app.mount("/mcp", mcp_app)

Result:

• REST API at http://localhost:8000/api/status
• MCP endpoint at http://localhost:8000/mcp

Critical: lifespan must be passed

# ✅ Correct — session manager initializes
app = FastAPI(lifespan=mcp_app.lifespan)
app.mount("/mcp", mcp_app)

# ❌ Wrong — session manager won't initialize, requests fail
app = FastAPI()
app.mount("/mcp", mcp.http_app(path="/"))

---

6. Combining Lifespans (Database + MCP)

When your FastAPI app already has a lifespan (for DB connections), use combine_lifespans to run both:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastmcp import FastMCP
from fastmcp.utilities.lifespan import combine_lifespans
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

# Your database lifespan
@asynccontextmanager
async def db_lifespan(app: FastAPI):
    engine = create_async_engine("postgresql+asyncpg://...")
    app.state.engine = engine
    app.state.session_factory = async_sessionmaker(engine, expire_on_commit=False)
    yield
    await engine.dispose()

# Create MCP server
mcp = FastMCP("My Server")

@mcp.tool
async def list_users() -> list[dict]:
    """List all users from the database."""
    # Access the session factory via the MCP server's state
    # or pass it through dependency injection
    ...

# Create MCP ASGI app
mcp_app = mcp.http_app(path="/")

# Combine both lifespans — DB + MCP
app = FastAPI(lifespan=combine_lifespans(db_lifespan, mcp_app.lifespan))
app.mount("/mcp", mcp_app)

combine_lifespans enters lifespans in order and exits in reverse order.

---

7. MCP Context (Logging, Progress, Resources)

Tools can access MCP features through the Context object:

from fastmcp import FastMCP, Context

mcp = FastMCP("My Server")

@mcp.tool
async def process_large_file(file_path: str, ctx: Context) -> str:
    """Process a large file with progress updates."""
    await ctx.info(f"Starting to process {file_path}")

    lines = read_file(file_path)
    total = len(lines)

    for i, line in enumerate(lines):
        await ctx.report_progress(progress=i, total=total)

    await ctx.info("Processing complete")
    return f"Processed {total} lines"

Context provides:

• Logging: ctx.debug(), ctx.info(), ctx.warning(), ctx.error()
• Progress: ctx.report_progress(progress, total)
• Resource access: ctx.read_resource(uri)
• LLM sampling: ctx.sample("Summarize this: ...")
• Request info: ctx.request_id, ctx.client_id

---

8. Accessing FastAPI State from MCP Tools

MCP tools run inside the same process as FastAPI. To access app state (database sessions, config, etc.), use a closure or module-level reference:

from fastapi import FastAPI
from fastmcp import FastMCP

mcp = FastMCP("My Server")

# Store a reference to app state that gets set during lifespan
_app_state: dict = {}

@mcp.tool
async def list_users() -> list[dict]:
    """List all users."""
    session_factory = _app_state["session_factory"]
    async with session_factory() as session:
        from opensilk.models import User
        result = await session.execute(select(User))
        users = result.scalars().all()
        return [{"id": str(u.id), "email": u.email} for u in users]

@asynccontextmanager
async def combined_lifespan(app: FastAPI):
    # Set up DB
    engine = create_async_engine(DATABASE_URL)
    factory = async_sessionmaker(engine, expire_on_commit=False)
    _app_state["session_factory"] = factory
    _app_state["engine"] = engine
    yield
    await engine.dispose()

# Wire up
mcp_app = mcp.http_app(path="/")
app = FastAPI(lifespan=combine_lifespans(combined_lifespan, mcp_app.lifespan))
app.mount("/mcp", mcp_app)

---

9. Generating MCP from FastAPI (auto-convert)

FastMCP can auto-convert your FastAPI endpoints into MCP tools:

mcp = FastMCP.from_fastapi(app=app)

This uses OpenAPIProvider under the hood — it reads your OpenAPI spec and exposes each endpoint as a tool.

Warning: Auto-converted tools have poor LLM performance compared to purpose-built tools. Use this for prototyping, not production:

"LLMs achieve significantly better performance with well-designed and curated MCP servers than with auto-converted OpenAPI servers." — FastMCP docs

For production, write purpose-built @mcp.tool functions with clear docstrings and simple parameters.

---

10. Production Deployment

Stateless mode (horizontal scaling)

When running multiple instances behind a load balancer, enable stateless mode — each request is self-contained, no sticky sessions needed:

app = mcp.http_app(stateless_http=True)

Or via environment variable:

FASTMCP_STATELESS_HTTP=true uvicorn app:app --workers 4

Sticky sessions don't work with MCP clients (Cursor, Claude Code don't forward cookies). Stateless mode is the only reliable scaling strategy.

nginx reverse proxy

SSE streaming requires specific nginx settings:

server {
    listen 443 ssl;
    server_name mcp.example.com;

    ssl_certificate /etc/letsencrypt/live/mcp.example.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/mcp.example.com/privkey.pem;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_http_version 1.1;
        proxy_set_header Connection '';
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;

        # CRITICAL for SSE streaming
        proxy_buffering off;
        proxy_cache off;
        proxy_read_timeout 300s;
        proxy_send_timeout 300s;
    }
}

proxy_buffering off is the most important setting. Without it, nginx buffers the entire SSE stream and delivers it only when the connection closes — breaking real-time communication.

systemd service

[Unit]
Description=OpenSilk MCP Server
After=network.target

[Service]
User=www-data
Group=www-data
WorkingDirectory=/opt/opensilk
ExecStart=/opt/opensilk/.venv/bin/uvicorn opensilk.app:app --host 127.0.0.1 --port 8000
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Health check endpoint

from starlette.responses import JSONResponse

@mcp.custom_route("/health", methods=["GET"])
async def health_check(request):
    return JSONResponse({"status": "healthy"})

Custom routes are never protected by auth — they're for load balancers and monitoring.

---

11. Authentication

Bearer token (simple)

from fastmcp.server.auth import BearerTokenAuth

auth = BearerTokenAuth(token="your-secret-token")
mcp = FastMCP("Protected Server", auth=auth)

OAuth providers

FastMCP supports GitHub, Google, Auth0, Supabase, Discord, and more. See the Authentication docs for full configuration.

---

12. Testing MCP Tools

Use the in-memory client to test without network:

from fastmcp import FastMCP
from fastmcp.client import Client
import asyncio

mcp = FastMCP("Test Server")

@mcp.tool
def add(a: int, b: int) -> int:
    """Add two numbers."""
    return a + b

async def test():
    async with Client(mcp) as client:
        # List available tools
        tools = await client.list_tools()
        print([t.name for t in tools])  # ['add']

        # Call a tool
        result = await client.call_tool("add", {"a": 1, "b": 2})
        print(result.data)  # 3

asyncio.run(test())

---

13. Complete Example: FastAPI + MCP + Database

import os
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastmcp import FastMCP
from fastmcp.utilities.lifespan import combine_lifespans
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

DATABASE_URL = os.environ["DATABASE_URL"]

# --- Database setup ---
_state: dict = {}

@asynccontextmanager
async def db_lifespan(app: FastAPI):
    engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
    _state["factory"] = async_sessionmaker(engine, expire_on_commit=False)
    yield
    await engine.dispose()

# --- MCP server ---
mcp = FastMCP("OpenSilk")

@mcp.tool(annotations={"readOnlyHint": True})
async def list_users() -> list[dict]:
    """List all registered users."""
    factory = _state["factory"]
    async with factory() as session:
        result = await session.execute(text("SELECT id, email, name FROM users"))
        return [dict(row) for row in result.fetchall()]

@mcp.tool
async def create_user(email: str, name: str) -> dict:
    """Create a new user."""
    factory = _state["factory"]
    async with factory() as session:
        result = await session.execute(
            text("INSERT INTO users (email, name) VALUES (:email, :name) RETURNING id, email, name"),
            {"email": email, "name": name},
        )
        await session.commit()
        return dict(result.fetchone())

# --- Wire up ---
mcp_app = mcp.http_app(path="/")
app = FastAPI(lifespan=combine_lifespans(db_lifespan, mcp_app.lifespan))
app.mount("/mcp", mcp_app)

@app.get("/health")
def health():
    return {"status": "ok"}

Run:

DATABASE_URL=postgresql+asyncpg://opensilk:opensilk@localhost:5432/opensilk \
uvicorn server:app --host 0.0.0.0 --port 8000

Result:

• REST health check at http://localhost:8000/health
• MCP endpoint at http://localhost:8000/mcp
• Tools: list_users, create_user

---

14. Full Demo: User Auth as MCP Tools

Ties together the PostgreSQL guide and this guide. Same User model, same password hashing, same database — but now exposed as both a REST API (for web clients) and MCP tools (for AI agents).

The LLM can register users, log in, and list users — all through MCP. The REST API serves the same data to browsers and scripts.

import os
import uuid
from contextlib import asynccontextmanager
from datetime import datetime

import bcrypt
from fastapi import FastAPI, HTTPException, APIRouter
from fastmcp import FastMCP
from fastmcp.utilities.lifespan import combine_lifespans
from pydantic import BaseModel, ConfigDict, EmailStr
from sqlalchemy import String, select, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

DATABASE_URL = os.environ["DATABASE_URL"]

# ── Database layer ───────────────────────────────────────────────

class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id:            Mapped[uuid.UUID]  = mapped_column(primary_key=True, default=uuid.uuid7)
    email:         Mapped[str]        = mapped_column(String(255), unique=True, index=True)
    name:          Mapped[str]        = mapped_column(String(100))
    password_hash: Mapped[str]        = mapped_column(String(255))
    created_at:    Mapped[datetime]   = mapped_column(server_default=text("now()"))


# ── Schemas (for REST API) ───────────────────────────────────────

class UserCreate(BaseModel):
    email: EmailStr
    name: str
    password: str

class UserLogin(BaseModel):
    email: EmailStr
    password: str

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: uuid.UUID
    email: str
    name: str
    created_at: datetime


# ── Auth helpers ─────────────────────────────────────────────────

def hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()

def verify_password(password: str, hashed: str) -> bool:
    return bcrypt.checkpw(password.encode(), hashed.encode())


# ── App state (shared between FastAPI and MCP) ───────────────────

_state: dict = {}

@asynccontextmanager
async def db_lifespan(app: FastAPI):
    engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)
    _state["engine"] = engine
    _state["factory"] = async_sessionmaker(engine, expire_on_commit=False)
    yield
    await engine.dispose()


# ── REST API routes (for browsers, scripts, curl) ────────────────

rest = APIRouter()

@rest.post("/register", response_model=UserResponse, status_code=201)
async def register(body: UserCreate):
    factory = _state["factory"]
    async with factory() as session:
        user = User(
            email=body.email,
            name=body.name,
            password_hash=hash_password(body.password),
        )
        session.add(user)
        try:
            await session.flush()
        except IntegrityError:
            raise HTTPException(409, "Email already registered")
        await session.commit()
        return user

@rest.post("/login", response_model=UserResponse)
async def login(body: UserLogin):
    factory = _state["factory"]
    async with factory() as session:
        stmt = select(User).where(User.email == body.email)
        user = (await session.execute(stmt)).scalar_one_or_none()
        if user is None or not verify_password(body.password, user.password_hash):
            raise HTTPException(401, "Invalid email or password")
        return user


# ── MCP tools (for AI agents — Claude Code, Cursor, etc.) ────────

mcp = FastMCP("User Service")

@mcp.tool
async def mcp_register(email: str, name: str, password: str) -> dict:
    """Register a new user account.

    Args:
        email: User's email address (must be unique).
        name: Display name for the user.
        password: Plaintext password (will be hashed).

    Returns:
        User object with id, email, name, and created_at.
        Returns {"error": "Email already registered"} on conflict.
    """
    factory = _state["factory"]
    async with factory() as session:
        user = User(
            email=email,
            name=name,
            password_hash=hash_password(password),
        )
        session.add(user)
        try:
            await session.flush()
        except IntegrityError:
            return {"error": "Email already registered"}
        await session.commit()
        return {
            "id": str(user.id),
            "email": user.email,
            "name": user.name,
            "created_at": user.created_at.isoformat(),
        }

@mcp.tool
async def mcp_login(email: str, password: str) -> dict:
    """Verify user credentials and return user info.

    Args:
        email: User's email address.
        password: Plaintext password to verify.

    Returns:
        User object on success, {"error": "Invalid credentials"} on failure.
    """
    factory = _state["factory"]
    async with factory() as session:
        stmt = select(User).where(User.email == email)
        user = (await session.execute(stmt)).scalar_one_or_none()
        if user is None or not verify_password(password, user.password_hash):
            return {"error": "Invalid credentials"}
        return {
            "id": str(user.id),
            "email": user.email,
            "name": user.name,
            "created_at": user.created_at.isoformat(),
        }

@mcp.tool(annotations={"readOnlyHint": True})
async def mcp_list_users() -> list[dict]:
    """List all registered users (id, email, name only — no password hashes).

    Returns:
        List of user objects.
    """
    factory = _state["factory"]
    async with factory() as session:
        stmt = select(User).order_by(User.created_at.desc())
        users = (await session.execute(stmt)).scalars().all()
        return [
            {"id": str(u.id), "email": u.email, "name": u.name}
            for u in users
        ]


# ── Wire everything together ─────────────────────────────────────

mcp_app = mcp.http_app(path="/")
app = FastAPI(lifespan=combine_lifespans(db_lifespan, mcp_app.lifespan))
app.include_router(rest)
app.mount("/mcp", mcp_app)

@app.get("/health")
def health():
    return {"status": "ok"}

What the LLM sees

An AI agent connecting to http://localhost:8000/mcp sees three tools:

mcp_register(email, name, password)   → create account
mcp_login(email, password)            → verify credentials
mcp_list_users()                      → list all users (read-only)

It does NOT see:

• password_hash (excluded from all return values)
• User SQLAlchemy model internals
• REST routes (/register, /login)
• Database connection details

What the browser sees

A web client hitting the REST API:

POST /register  {"email": "...", "name": "...", "password": "..."}  → 201
POST /login     {"email": "...", "password": "..."}                  → 200
GET  /health                                                        → 200

Key differences: REST vs MCP tools

Aspect	REST routes	MCP tools
Audience	Browsers, curl, scripts	AI agents (Claude, Cursor)
Naming	RESTful verbs (/register)	Descriptive (mcp_register)
Errors	HTTPException (409, 401)	Return {"error": "..."} dict
Response	Pydantic model (UserResponse)	Plain dict (no password_hash)
Schema	OpenAPI auto-generated	FastMCP auto-generated from type hints
Annotations	None	readOnlyHint, destructiveHint

Why return dicts from MCP tools instead of raising exceptions? LLMs handle structured error responses better than HTTP error codes. An AI agent can read {"error": "Email already registered"} and explain it to the user. An HTTP 409 is opaque.

Run it

# 1. Start Postgres
docker compose up -d

# 2. Run migrations
alembic upgrade head

# 3. Start the server
DATABASE_URL=postgresql+asyncpg://opensilk:opensilk@localhost:5432/opensilk \
uvicorn server:app --host 0.0.0.0 --port 8000

# 4. Test MCP endpoint
curl http://localhost:8000/mcp

# 5. Test REST endpoint
curl -X POST http://localhost:8000/register \
  -H "Content-Type: application/json" \
  -d '{"email":"alice@test.com","name":"Alice","password":"secret"}'

# 6. Connect from Claude Desktop
# Add to claude_desktop_config.json:
# {
#   "mcpServers": {
#     "user-service": {
#       "url": "http://localhost:8000/mcp"
#     }
#   }
# }

Now Claude Code can register users, log in, and list users — by calling MCP tools that execute against your Postgres database.

---

Reference

Resource	URL
LLM-friendly doc index	https://gofastmcp.com/llms.txt
FastAPI integration	https://gofastmcp.com/integrations/fastapi
Tools docs	https://gofastmcp.com/servers/tools
HTTP deployment	https://gofastmcp.com/deployment/http
Auth	https://gofastmcp.com/servers/auth/authentication
Middleware	https://gofastmcp.com/servers/middleware
Context	https://gofastmcp.com/servers/context
Dependency injection	https://gofastmcp.com/servers/dependency-injection

postgres-fastapi-guide.md

FastAPI + AsyncPG + Alembic: A Practical Guide

For developers coming from Rust/sqlx. Covers drivers, sessions, CRUD, migrations, and deployment.

---

1. PostgreSQL Drivers in Python

Three drivers exist. Only one matters for async FastAPI:

Driver	Sync/Async	Speed	Use it?
psycopg2	Sync only	Slow	No — blocks the event loop
psycopg3 (psycopg)	Both	Medium	No — async mode slower than asyncpg
asyncpg	Async only	Fastest	Yes — built for asyncio

SQLAlchemy wraps drivers via a dialect prefix:

postgresql+asyncpg://user:pass@host:5432/dbname
            ^^^^^^^
            "Use Postgres syntax, route queries through asyncpg"

This is the only connection string you'll ever write.

---

2. SQLAlchemy Layers (3 levels)

Engine (pool of connections)
  └─ async_sessionmaker (factory that creates sessions)
       └─ AsyncSession (one transaction's worth of work)

Engine — created once at startup. Holds a connection pool. Like sqlx's PgPool.

Session — created per-request. Tracks loaded objects, dirty state, manages the transaction. Like a sqlx transaction scope.

The session IS the transaction. Everything inside one session is atomic:

async with factory() as session:         ← BEGIN (lazily, on first query)
    await session.execute(select(...))   ← SELECT (within txn)
    session.add(user)                    ← staged in memory
    await session.flush()                ← INSERT sent to Postgres
    await session.commit()               ← COMMIT — all changes permanent
                                          ← or: exception → ROLLBACK

---

3. Three Async Gotchas

These bite everyone. Memorize them.

Gotcha 1: expire_on_commit=False is mandatory

After commit(), SQLAlchemy by default "expires" all objects — next attribute access triggers a refresh query. In sync code that's invisible. In async code it crashes:

# expire_on_commit=True (default) — CRASHES
await session.commit()
print(user.name)   # 💥 MissingGreenletError

# expire_on_commit=False — works
await session.commit()
print(user.name)   # ✅ value still in memory

Gotcha 2: No lazy loading. Ever.

Accessing a relationship that wasn't explicitly loaded → crash (same reason). You must load eagerly:

# 💥 CRASHES — .posts not loaded
user = (await db.execute(select(User))).scalar_one()
user.posts

# ✅ Explicit eager load
stmt = select(User).options(selectinload(User.posts))
user = (await db.execute(stmt)).scalar_one()
user.posts   # already in memory

This mirrors sqlx: you write the JOIN, you get the data. No surprises.

Gotcha 3: pool_pre_ping=True

Idle connections get killed by Postgres/firewalls after a timeout. Next request grabs a dead connection → crash. pool_pre_ping sends a SELECT 1 before handing out a connection:

engine = create_async_engine(DATABASE_URL, pool_pre_ping=True)

---

4. Project Structure

No globals. Engine lives on app.state via lifespan. Sessions are injected per-request via FastAPI dependencies.

src/opensilk/
    db.py              # Base, lifespan, get_db, DbSession
    models.py          # SQLAlchemy models (database tables)
    schemas.py         # Pydantic schemas (API request/response)
    app.py             # FastAPI app, wires lifespan + routes
    routes/
        users.py       # imports DbSession from db.py

db.py — the shared layer

from contextlib import asynccontextmanager
from typing import Annotated, AsyncGenerator

from fastapi import Depends, FastAPI, Request
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


def create_db_lifespan(database_url: str):
    """Returns a lifespan that creates/disposes the engine."""

    @asynccontextmanager
    async def lifespan(app: FastAPI):
        engine = create_async_engine(database_url, pool_pre_ping=True)
        app.state.engine = engine
        app.state.session_factory = async_sessionmaker(
            engine,
            expire_on_commit=False,
        )
        yield
        await engine.dispose()

    return lifespan


async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]:
    factory = request.app.state.session_factory
    async with factory() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise


DbSession = Annotated[AsyncSession, Depends(get_db)]

app.py — wires it together

import os
from fastapi import FastAPI
from opensilk.db import create_db_lifespan
from opensilk.routes import users

app = FastAPI(lifespan=create_db_lifespan(os.environ["DATABASE_URL"]))
app.include_router(users.router)

routes/users.py — any route file

from fastapi import APIRouter
from opensilk.db import DbSession

router = APIRouter()

@router.get("/users")
async def list_users(db: DbSession):
    ...

How dependency injection works

GET /users
  │
  ▼
FastAPI sees db: DbSession parameter
  │
  ▼
Calls get_db(request)
  ├── factory = request.app.state.session_factory
  ├── async with factory() as session:    ← opens connection from pool
  │       yield session                    ← hands session to handler
  │       await session.commit()           ← runs AFTER handler returns
  │
  ▼
Your handler runs with a live AsyncSession

If the handler raises an exception, commit() never runs → automatic rollback. Connection returns to pool.

---

5. Models vs Schemas

Strict separation: database model (SQLAlchemy) vs API schema (Pydantic). Different routes need different field subsets.

models.py — database representation

import uuid
from datetime import datetime
from sqlalchemy import Integer, String, text
from sqlalchemy.orm import Mapped, mapped_column
from opensilk.db import Base


class User(Base):
    __tablename__ = "users"

    id:            Mapped[uuid.UUID]  = mapped_column(primary_key=True, default=uuid.uuid7)  # Python 3.14+
    email:         Mapped[str]        = mapped_column(String(255), unique=True, index=True)
    name:          Mapped[str]        = mapped_column(String(100))
    password_hash: Mapped[str]        = mapped_column(String(255))
    birth_year:    Mapped[int | None] = mapped_column(Integer, nullable=True)
    created_at:    Mapped[datetime]   = mapped_column(server_default=text("now()"))

Mapped[T] tells SQLAlchemy the Python type. mapped_column(...) tells it the SQL type + constraints.

• default=uuid.uuid7 — Python generates the UUID before INSERT (Python 3.14+; use uuid.uuid4 on older versions)
• server_default=text("now()") — Postgres generates it during INSERT

schemas.py — API validation

from datetime import datetime
from uuid import UUID
from pydantic import BaseModel, ConfigDict, EmailStr


class UserCreate(BaseModel):
    email: EmailStr
    name: str
    password: str


class UserLogin(BaseModel):
    email: EmailStr
    password: str


class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # read from ORM objects
    id: UUID
    email: str
    name: str
    created_at: datetime

Notice: UserCreate has password (plaintext from client). UserResponse does not have password_hash — never leak hashes. The split enforces this at the type level.

---

6. CRUD Operations

SELECT (read)

# List all
stmt = select(User).order_by(User.created_at.desc())
users = (await db.execute(stmt)).scalars().all()

# Get one by id
stmt = select(User).where(User.id == user_id)
user = (await db.execute(stmt)).scalar_one_or_none()
if user is None:
    raise HTTPException(404)

• scalar_one_or_none() — one object or None. Crashes if >1 match.
• scalars().all() — list of objects.
• one() — exactly one, crashes if 0 or 2+.

INSERT (create)

user = User(
    email=body.email,
    name=body.name,
    password_hash=hash_password(body.password),
)
db.add(user)        # staged in memory — no SQL yet
await db.flush()    # INSERT ... RETURNING sent to Postgres
return user         # id + created_at populated automatically

UPDATE (modify)

user.name = body.name   # just set the attribute
await db.flush()        # UPDATE users SET name=$1 WHERE id=$2
return user

No .update() call. SQLAlchemy tracks dirty attributes — changing user.name marks it dirty, flush() sends the UPDATE.

DELETE (remove)

await db.delete(user)   # stage deletion
await db.flush()        # DELETE FROM users WHERE id=$1

Raw SQL (when you need full control)

Same DbSession, wrap SQL in text():

from sqlalchemy import text

result = await db.execute(
    text("SELECT * FROM users WHERE email = :email"),
    {"email": "alice@test.com"},
)
row = result.fetchone()

Mix freely — ORM for simple CRUD, raw SQL for complex joins. Same session, same transaction, same pool.

---

7. flush() vs commit()

	flush()	commit()
What it does	Sends staged SQL to Postgres	Commits the transaction
Visible to other connections?	No — still inside transaction	Yes — permanent
Can rollback?	Yes	No
Populates auto-generated values?	Yes via RETURNING	N/A

How flush populates objects

db.add(user)
  → session remembers: "this object needs INSERT"
  → user.id = <set by Python default=uuid7>
  → user.created_at = None  ← can't know yet

await db.flush()
  → SQLAlchemy generates: INSERT INTO users (id, email, name, password_hash)
                           VALUES ($1, $2, $3, $4)
                           RETURNING users.created_at
  → Postgres inserts, computes created_at = now(), returns it
  → SQLAlchemy populates: user.created_at = 2026-06-15T...

You never write RETURNING yourself. SQLAlchemy adds it automatically for any column with server_default, autoincrement, or other DB-generated values.

Commit styles

Auto-commit (get_db owns the commit):

# get_db:
async with factory() as session:
    yield session
    await session.commit()     # runs after handler returns successfully

# Handler — just flush, don't commit:
async def register(body: UserCreate, db: DbSession):
    user = User(...)
    db.add(user)
    await db.flush()
    return user                # commit happens in get_db

Pro: if handler raises, no commit → automatic rollback. Transaction boundary is in one place.

Con: slightly indirect — "why flush not commit?"

Manual commit (handler owns the commit):

async def register(body: UserCreate, db: DbSession):
    user = User(...)
    db.add(user)
    await db.flush()
    await db.commit()          # explicit — permanent
    return user

Pro: explicit control. You see exactly when the transaction finalizes.

Con: you might forget to commit.

Pick one style and be consistent. Most production codebases use manual commit.

---

8. Alembic Migrations

What Alembic is

A version control system for your database schema. It maintains one table — alembic_version — containing the current revision ID. Each migration is a Python file with upgrade() and downgrade() functions.

One-time setup

pip install alembic
alembic init -t async migrations    # -t async = crucial for asyncpg

Wire three things in migrations/env.py:

# 1. Import your models (so autogenerate can see them)
from opensilk.db import Base
from opensilk.models import User     # MUST import every model

# 2. Point to your metadata
target_metadata = Base.metadata

# 3. Set the DB URL from env
config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"])

Daily workflow

# 1. You changed a model (added column, new table, etc.)
#    Generate the migration from the diff:
alembic revision --autogenerate -m "add birth_year to users"

# 2. REVIEW the generated file — non-negotiable
#    Alembic can't detect renames (does drop+create instead)

# 3. Apply:
alembic upgrade head

# 4. Rollback if needed:
alembic downgrade -1

What a migration file looks like

"""add birth_year to users

Revision ID: b2c3d4e5f6a7
Revises: a1b2c3d4e5f6
"""
from alembic import op
import sqlalchemy as sa


def upgrade() -> None:
    op.add_column("users", sa.Column("birth_year", sa.Integer(), nullable=True))

def downgrade() -> None:
    op.drop_column("users", "birth_year")

op is the operations API — create_table, add_column, drop_column, create_index, alter_column. It compiles to the right SQL for your dialect.

Pro practices

Always review autogenerate. Alembic diffs model metadata vs live DB. It can't know you renamed a column — it sees "column X gone, column Y new" and generates drop+add. Fix it to op.alter_column(..., new_column_name=...).

Test downgrade. Run upgrade head → downgrade -1 → upgrade head in CI. If downgrade() is broken, you can't roll back a bad deploy.

Never edit a deployed migration. Once it runs in production, it's immutable. Create a new migration to fix mistakes.

Naming conventions prevent breakage. Without them, Postgres auto-generates constraint names that vary between environments and break migrations that reference them:

from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    metadata = MetaData(naming_convention={
        "ix": "ix_%(column_0_label)s",
        "uq": "uq_%(table_name)s_%(column_0_name)s",
        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
        "pk": "pk_%(table_name)s",
    })

Cheat sheet

Command	What it does
alembic revision --autogenerate -m "msg"	Generate migration from model changes
alembic revision -m "msg"	Empty migration (write manually)
alembic upgrade head	Apply all pending migrations
alembic upgrade +1	Apply next migration only
alembic downgrade -1	Roll back one migration
alembic downgrade base	Roll back ALL migrations
alembic current	Show current revision in DB
alembic history	Show migration chain

---

9. Deploying a Schema Change to Production

Example: adding birth_year to the users table. Simple SSH server.

Step 1: Add the field to the model (local)

class User(Base):
    ...
    birth_year: Mapped[int | None] = mapped_column(Integer, nullable=True)

nullable=True is critical. Existing rows need a value. You can't add a NOT NULL column to a table with existing data without a default.

Step 2: Generate migration (local)

alembic revision --autogenerate -m "add birth_year to users"

Step 3: Review

Check it says add_column — NOT drop_table + create_table.

Step 4: Test locally

alembic upgrade head      # apply
alembic downgrade -1      # roll back
alembic upgrade head      # re-apply

All three must work.

Step 5: Commit + push

git add .
git commit -m "feat: add birth_year to users"
git push

Step 6: Deploy (SSH)

ssh user@your-server
cd /opt/opensilk
git pull
source .venv/bin/activate

# Migration FIRST — before app restart
alembic upgrade head

# Then restart the app
sudo systemctl restart opensilk

The order matters

alembic upgrade head    ← schema has birth_year now
                        ← old app still running (ignores extra column, fine)
systemctl restart app   ← new app starts (expects birth_year, it's there)

If you reversed it — restart app before migration — the new code would try to SELECT birth_year FROM users on a table without that column → crash.

Production: migrations run OUTSIDE the app

Never embed alembic upgrade head in FastAPI startup code. Race condition: multiple replicas start simultaneously, all try to migrate. Run it in the deployment entrypoint:

#!/bin/sh
# entrypoint.sh
alembic upgrade head
uvicorn opensilk.app:app --host 0.0.0.0 --port 8000

Making a column NOT NULL later (two-migration pattern)

# Migration 1: add nullable column
alembic revision --autogenerate -m "add birth_year nullable"
# Deploy, let app populate values...

# Migration 2: backfill + enforce NOT NULL (manual)
alembic revision -m "birth_year not null"

def upgrade() -> None:
    op.execute("UPDATE users SET birth_year = 2000 WHERE birth_year IS NULL")
    op.alter_column("users", "birth_year", nullable=False)

---

10. Full Example: User CRUD

auth.py

import bcrypt

def hash_password(password: str) -> str:
    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()

def verify_password(password: str, hashed: str) -> bool:
    return bcrypt.checkpw(password.encode(), hashed.encode())

routes/users.py

from uuid import UUID
from fastapi import APIRouter, HTTPException
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from opensilk.db import DbSession
from opensilk.models import User
from opensilk.schemas import UserCreate, UserLogin, UserResponse
from opensilk.auth import hash_password, verify_password

router = APIRouter()


@router.post("/register", response_model=UserResponse, status_code=201)
async def register(body: UserCreate, db: DbSession):
    user = User(
        email=body.email,
        name=body.name,
        password_hash=hash_password(body.password),
    )
    db.add(user)
    try:
        await db.flush()
    except IntegrityError:
        raise HTTPException(409, "Email already registered")
    return user


@router.post("/login", response_model=UserResponse)
async def login(body: UserLogin, db: DbSession):
    stmt = select(User).where(User.email == body.email)
    user = (await db.execute(stmt)).scalar_one_or_none()
    if user is None or not verify_password(body.password, user.password_hash):
        raise HTTPException(401, "Invalid email or password")
    return user


@router.get("/users", response_model=list[UserResponse])
async def list_users(db: DbSession):
    stmt = select(User).order_by(User.created_at.desc())
    return (await db.execute(stmt)).scalars().all()


@router.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: UUID, db: DbSession):
    stmt = select(User).where(User.id == user_id)
    user = (await db.execute(stmt)).scalar_one_or_none()
    if user is None:
        raise HTTPException(404, "User not found")
    return user


@router.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: UUID, db: DbSession):
    stmt = select(User).where(User.id == user_id)
    user = (await db.execute(stmt)).scalar_one_or_none()
    if user is None:
        raise HTTPException(404, "User not found")
    await db.delete(user)
    await db.flush()

Dependencies

fastapi
uvicorn
sqlalchemy[asyncio]>=2.0.0
asyncpg
alembic
pydantic>=2.0.0
pydantic[email]
bcrypt