sqlmodel-database skill

---
name: sqlmodel-database
description: Design and implement database schemas using SQLModel with sync and async patterns. Use this skill when creating database models, setting up Neon PostgreSQL connections, defining relationships (one-to-many, many-to-many), implementing FastAPI dependency injection, or migrating schemas. Covers both sync Session and async AsyncSession patterns.
---

# SQLModel Database

Design and implement database schemas using SQLModel - the library that combines SQLAlchemy's power with Pydantic's validation.

## When to Use

- Creating database models for FastAPI applications
- Setting up PostgreSQL/Neon database connections
- Defining model relationships (one-to-many, many-to-many)
- Implementing async database operations
- Creating API request/response schemas from database models
- FastAPI dependency injection for sessions

## Quick Start

```bash
# Sync only
uv add sqlmodel

# Async support (recommended for FastAPI)
uv add sqlmodel sqlalchemy[asyncio] asyncpg  # PostgreSQL
# or
uv add sqlmodel sqlalchemy[asyncio] aiosqlite  # SQLite
```

## Core Patterns

### 1. Model Hierarchy (Base → Table → API)

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

# Base model - shared fields, validation, NO table
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"
    progress_percent: int = Field(default=0, ge=0, le=100)

# Table model - has table=True, adds id and timestamps
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)

# Create model - for POST requests (inherits validation)
class TaskCreate(TaskBase):
    pass

# Update model - all fields optional for PATCH
class TaskUpdate(SQLModel):
    title: Optional[str] = None
    description: Optional[str] = None
    status: Optional[str] = None
    priority: Optional[str] = None
    progress_percent: Optional[int] = Field(default=None, ge=0, le=100)

# Read model - for responses (includes id)
class TaskRead(TaskBase):
    id: int
    created_at: datetime
    updated_at: datetime
```

### 2. Sync Database Connection

```python
from sqlmodel import create_engine, Session, SQLModel
import os

DATABASE_URL = os.getenv("DATABASE_URL")
# PostgreSQL: postgresql://user:pass@host/db?sslmode=require
# SQLite: sqlite:///database.db

engine = create_engine(DATABASE_URL, echo=True)

def create_db_and_tables():
    SQLModel.metadata.create_all(engine)

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

### 3. Async Database Connection (Recommended)

```python
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from sqlalchemy.orm import sessionmaker
import os

# Async URLs use different drivers
DATABASE_URL = os.getenv("DATABASE_URL")
# PostgreSQL async: postgresql+asyncpg://user:pass@host/db
# SQLite async: sqlite+aiosqlite:///database.db

async_engine = create_async_engine(DATABASE_URL, echo=True)

async def create_db_and_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

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

### 4. FastAPI Integration (Async)

```python
from fastapi import FastAPI, Depends, HTTPException
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

app = FastAPI()

@app.on_event("startup")
async def on_startup():
    await create_db_and_tables()

@app.post("/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("/tasks", response_model=list[TaskRead])
async def list_tasks(
    session: AsyncSession = Depends(get_session),
    offset: int = 0,
    limit: int = 100,
):
    statement = select(Task).offset(offset).limit(limit)
    results = await session.exec(statement)
    return results.all()

@app.get("/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("/tasks/{task_id}", response_model=TaskRead)
async def update_task(
    task_id: int,
    task_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")

    # Only update fields that were explicitly set
    update_data = task_update.model_dump(exclude_unset=True)
    task.sqlmodel_update(update_data)

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

@app.delete("/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 {"ok": True}
```

### 5. Relationships

#### One-to-Many (Task belongs to Project)

```python
from sqlmodel import Field, Relationship
from typing import Optional, List

class Project(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    slug: str = Field(unique=True)

    # One project has many tasks
    tasks: List["Task"] = Relationship(back_populates="project")

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str

    # Foreign key
    project_id: Optional[int] = Field(default=None, foreign_key="project.id")

    # Relationship back to project
    project: Optional[Project] = Relationship(back_populates="tasks")
```

#### Many-to-Many (Task has many Workers, Worker has many Tasks)

```python
# Link table (no extra fields)
class TaskWorkerLink(SQLModel, table=True):
    task_id: Optional[int] = Field(
        default=None, foreign_key="task.id", primary_key=True
    )
    worker_id: Optional[str] = Field(
        default=None, foreign_key="worker.id", primary_key=True
    )

class Worker(SQLModel, table=True):
    id: str = Field(primary_key=True)  # e.g., "@claude-code"
    name: str
    type: Literal["human", "agent"]

    tasks: List["Task"] = Relationship(
        back_populates="workers",
        link_model=TaskWorkerLink
    )

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str

    workers: List[Worker] = Relationship(
        back_populates="tasks",
        link_model=TaskWorkerLink
    )
```

#### Many-to-Many with Extra Fields

```python
# Link table WITH extra fields
class TaskAssignment(SQLModel, table=True):
    task_id: Optional[int] = Field(
        default=None, foreign_key="task.id", primary_key=True
    )
    worker_id: Optional[str] = Field(
        default=None, foreign_key="worker.id", primary_key=True
    )
    assigned_at: datetime = Field(default_factory=datetime.now)
    role: str = "assignee"  # assignee, reviewer, etc.

    # Relationships to both sides
    task: "Task" = Relationship(back_populates="assignments")
    worker: "Worker" = Relationship(back_populates="assignments")

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    assignments: List[TaskAssignment] = Relationship(back_populates="task")

class Worker(SQLModel, table=True):
    id: str = Field(primary_key=True)
    name: str
    assignments: List[TaskAssignment] = Relationship(back_populates="worker")
```

### 6. Self-Referential (Parent-Child Tasks)

```python
class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str

    # Self-referential foreign key
    parent_id: Optional[int] = Field(default=None, foreign_key="task.id")

    # Parent relationship
    parent: Optional["Task"] = Relationship(
        back_populates="subtasks",
        sa_relationship_kwargs={"remote_side": "Task.id"}
    )

    # Children relationship
    subtasks: List["Task"] = Relationship(back_populates="parent")
```

### 7. Partial Updates with exclude_unset

```python
# The key pattern for PATCH endpoints
update_data = task_update.model_dump(exclude_unset=True)
db_task.sqlmodel_update(update_data)

# This ensures:
# - Fields not in request body are NOT overwritten
# - Fields explicitly set to null ARE updated to null
# - Only provided fields are modified
```

### 8. Query Patterns

```python
from sqlmodel import select, or_, and_, col

# Basic select
statement = select(Task)
results = await session.exec(statement)
tasks = results.all()

# With filters
statement = select(Task).where(Task.status == "pending")

# Multiple conditions (AND)
statement = select(Task).where(
    Task.status == "in_progress",
    Task.priority == "high"
)

# OR conditions
statement = select(Task).where(
    or_(Task.status == "pending", Task.status == "in_progress")
)

# With ordering
statement = select(Task).order_by(Task.created_at.desc())

# With pagination
statement = select(Task).offset(20).limit(10)

# Get single by ID
task = await session.get(Task, task_id)

# Get first match
statement = select(Task).where(Task.title == "Example")
result = await session.exec(statement)
task = result.first()

# Count
from sqlalchemy import func
statement = select(func.count()).select_from(Task)
result = await session.exec(statement)
count = result.one()

# Join with eager loading
from sqlalchemy.orm import selectinload
statement = select(Task).options(selectinload(Task.project))
```

### 9. Neon PostgreSQL Connection

```python
import os
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine

# Neon connection string (async)
# Original: postgresql://user:pass@host/db?sslmode=require
# Async: postgresql+asyncpg://user:pass@host/db?ssl=require

DATABASE_URL = os.getenv("DATABASE_URL")

# Convert sync URL to async if needed
if DATABASE_URL.startswith("postgresql://"):
    DATABASE_URL = DATABASE_URL.replace(
        "postgresql://",
        "postgresql+asyncpg://",
        1
    ).replace("sslmode=", "ssl=")

async_engine = create_async_engine(
    DATABASE_URL,
    echo=True,
    pool_size=5,
    max_overflow=10,
)
```

## Testing Patterns

```python
import pytest
from sqlmodel import SQLModel, create_engine, Session
from sqlmodel.pool import StaticPool
from fastapi.testclient import TestClient

@pytest.fixture
def session():
    """In-memory SQLite for fast tests."""
    engine = create_engine(
        "sqlite://",  # In-memory
        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):
    """Test client with overridden session."""
    def get_session_override():
        return session

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

def test_create_task(client):
    response = client.post("/tasks", json={"title": "Test"})
    assert response.status_code == 201
    assert response.json()["title"] == "Test"

def test_update_partial(client, session):
    # Create task
    task = Task(title="Original", priority="low")
    session.add(task)
    session.commit()

    # Partial update - only change priority
    response = client.patch(
        f"/tasks/{task.id}",
        json={"priority": "high"}
    )

    assert response.status_code == 200
    assert response.json()["title"] == "Original"  # Unchanged
    assert response.json()["priority"] == "high"   # Updated
```

## Critical: Async Session Patterns (MissingGreenlet Prevention)

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

### The Pattern: Extract → Flush → Commit

```python
@router.post("/entities", response_model=EntityRead, status_code=201)
async def create_entity(
    data: EntityCreate,
    session: AsyncSession = Depends(get_session),
    user: CurrentUser = Depends(get_current_user),
) -> EntityRead:
    # 1. Get related objects and EXTRACT primitives immediately
    worker = await get_worker(session, user.id)
    worker_id = worker.id        # Extract BEFORE any commit
    worker_type = worker.type    # Extract BEFORE any commit

    # 2. Create entity
    entity = Entity(**data.model_dump())
    session.add(entity)

    # 3. Use flush() to get generated ID without committing
    await session.flush()
    entity_id = entity.id        # Extract immediately after flush

    # 4. Do related operations (audit logging, etc.)
    # Pass primitives, NOT objects
    await log_action(
        session,
        entity_id=entity_id,
        actor_id=worker_id,
        actor_type=worker_type,  # String, not object
    )

    # 5. Single commit at the end
    await session.commit()

    # 6. Refresh if you need to return the object
    await session.refresh(entity)
    return entity
```

### Service Functions: Never Commit Internally

```python
# WRONG - service commits, caller loses transaction control
async def log_action(session: AsyncSession, ...):
    log = AuditLog(...)
    session.add(log)
    await session.commit()  # ❌ Don't do this!

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

### The Problem: Object Detachment After Commit

```python
# ❌ WRONG - causes MissingGreenlet
worker = await session.get(Worker, 1)
await session.commit()           # Worker is now DETACHED
print(worker.name)               # ERROR! Triggers lazy load in async

# ✅ CORRECT - extract before commit
worker = await session.get(Worker, 1)
worker_name = worker.name        # Extract while attached
await session.commit()
print(worker_name)               # Safe - it's a string
```

### When Passing Objects Between Functions

```python
# ❌ WRONG - passing ORM object that may be detached
async def do_work(entity: Entity):
    print(entity.name)  # May fail if entity was detached

# ✅ CORRECT - pass primitive IDs
async def do_work(entity_id: int, session: AsyncSession):
    entity = await session.get(Entity, entity_id)  # Fresh fetch
    print(entity.name)
```

---

## Common Pitfalls

### 1. Forgetting table=True
```python
# WRONG - no table created
class Task(SQLModel):
    id: int

# CORRECT
class Task(SQLModel, table=True):
    id: int
```

### 2. Using sync Session with async engine
```python
# WRONG
from sqlmodel import Session
async with Session(async_engine) as session:  # Error!

# CORRECT
from sqlmodel.ext.asyncio.session import AsyncSession
async with AsyncSession(async_engine) as session:
```

### 3. Missing await on async operations
```python
# WRONG
task = session.get(Task, 1)  # Returns coroutine, not Task!

# CORRECT
task = await session.get(Task, 1)
```

### 4. Relationship without back_populates
```python
# WRONG - can cause issues
tasks: List["Task"] = Relationship()

# CORRECT
tasks: List["Task"] = Relationship(back_populates="project")
```

### 5. MissingGreenlet after commit (see "Critical: Async Session Patterns" above)
```python
# WRONG - accessing detached object
await session.commit()
print(entity.name)  # MissingGreenlet error!

# CORRECT - extract before commit or refresh after
name = entity.name  # Before commit
await session.commit()
# OR
await session.commit()
await session.refresh(entity)  # Reattach
print(entity.name)
```

### 6. MissingGreenlet in log statements (sneaky!)

**This is a common trap** - log statements that access ORM attributes after commit:

```python
# WRONG - logger.info triggers attribute access after commit!
session.add(notification)
await session.commit()
logger.info("Created notification %d for %s", notification.id, notification.user_id)
#                                              ^^^^^^^^^^^^^^^ MissingGreenlet!

# CORRECT - refresh after commit before accessing any attributes
session.add(notification)
await session.commit()
await session.refresh(notification)  # Reattach to session
logger.info("Created notification %d for %s", notification.id, notification.user_id)

# ALTERNATIVE - extract values before commit
session.add(notification)
await session.flush()  # Get ID without committing
notif_id = notification.id  # Extract while attached
user_id = notification.user_id
await session.commit()
logger.info("Created notification %d for %s", notif_id, user_id)
```

**Why this is sneaky**: The commit succeeds, data is saved, but then the log line crashes.
The notification exists in the database, but you get an error in logs.

---

## Input Validation Patterns

### Converting 0 to None for nullable foreign keys

Swagger UI and some clients send `0` as default for optional int fields. This violates FK constraints.

```python
from pydantic import field_validator

class TaskCreate(SQLModel):
    assignee_id: int | None = None
    parent_task_id: int | None = None

    @field_validator("assignee_id", "parent_task_id", mode="after")
    @classmethod
    def zero_to_none(cls, v: int | None) -> int | None:
        """Convert 0 to None (0 is not a valid FK)."""
        return None if v == 0 else v
```

### Normalizing timezone-aware datetimes

Database stores naive UTC, but clients may send ISO 8601 with timezone.

```python
from datetime import UTC, datetime
from pydantic import field_validator

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

    @field_validator("due_date", mode="after")
    @classmethod
    def normalize_datetime(cls, v: datetime | None) -> datetime | None:
        """Strip timezone after converting to UTC."""
        if v is None:
            return None
        if v.tzinfo is not None:
            v = v.astimezone(UTC).replace(tzinfo=None)
        return v
```

## References

For additional documentation, use Context7 MCP:
```
mcp__context7__resolve-library-id with libraryName="sqlmodel"
mcp__context7__get-library-docs with topic="relationships" or "async"
```

See also: `references/migrations.md` for schema migration patterns.