Sqlmodel Expert

SQLModel Expert

Advanced SQLModel patterns and comprehensive Alembic migrations for production databases.

Quick Start

Define a Basic Model

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

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str = Field(index=True)
    description: Optional[str] = None
    completed: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)

Initialize Database

# Using provided script
python scripts/init_db.py --url postgresql://user:pass@localhost/db

# Or manually
from sqlmodel import create_engine
engine = create_engine("postgresql://user:pass@localhost/db")
SQLModel.metadata.create_all(engine)

Create Migration

# Using provided helper script
./scripts/migrate.sh create "add user table"

# Or directly with Alembic
alembic revision --autogenerate -m "add user table"
alembic upgrade head

Core Topics

1. Advanced Model Patterns

See: references/advanced-models.md

• Relationships: One-to-many, many-to-many, self-referential
• Inheritance: Single table, joined table, polymorphism
• Validation: Pydantic validators, custom constraints
• Mixins: Timestamp, soft delete, reusable patterns
• Field Types: Enums, JSON, arrays, custom types
• Indexes: Single, composite, partial indexes
• Constraints: Unique, check, foreign key cascades

2. Comprehensive Migrations

See: references/migrations.md

• Alembic Setup: Configuration, env.py for SQLModel
• Creating Migrations: Autogenerate vs manual
• Schema Changes: Add/drop columns, rename, change types
• Data Migrations: Complex data transformations
• Production Workflow: Zero-downtime migrations
• Rollback Strategies: Safe downgrade patterns
• Troubleshooting: Common issues and solutions

3. Query Optimization

See: references/queries-optimization.md

• N+1 Problem: Solutions with eager loading
• Query Patterns: Joins, aggregations, subqueries
• Performance: Indexes, batch operations, profiling
• Advanced Queries: Window functions, CTEs
• Bulk Operations: Insert, update, delete at scale
• Testing: Query counting, explain analyze

Common Patterns

One-to-Many Relationship

from typing import List
from sqlmodel import Field, Relationship, SQLModel

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

    # One team has many heroes
    heroes: List["Hero"] = Relationship(back_populates="team")

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    team_id: Optional[int] = Field(foreign_key="team.id")

    # Many heroes belong to one team
    team: Optional[Team] = Relationship(back_populates="heroes")

Many-to-Many with Link Table

class HeroTeamLink(SQLModel, table=True):
    hero_id: int = Field(foreign_key="hero.id", primary_key=True)
    team_id: int = Field(foreign_key="team.id", primary_key=True)
    joined_at: datetime = Field(default_factory=datetime.utcnow)

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    teams: List["Team"] = Relationship(
        back_populates="heroes",
        link_model=HeroTeamLink
    )

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    heroes: List[Hero] = Relationship(
        back_populates="teams",
        link_model=HeroTeamLink
    )

Solving N+1 Query Problem

from sqlalchemy.orm import selectinload

# BAD - N+1 queries
users = session.exec(select(User)).all()
for user in users:
    posts = user.posts  # Each triggers a query!

# GOOD - Eager loading (2 queries total)
statement = select(User).options(selectinload(User.posts))
users = session.exec(statement).all()
for user in users:
    posts = user.posts  # No additional query!

Creating a Migration

# 1. Modify your model
class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str
    phone: str  # New field added

# 2. Generate migration
# alembic revision --autogenerate -m "add phone to user"

# 3. Review generated migration
def upgrade() -> None:
    op.add_column('user', sa.Column('phone', sa.String(), nullable=True))

def downgrade() -> None:
    op.drop_column('user', 'phone')

# 4. Apply migration
# alembic upgrade head

Migration Helper Scripts

Initialize Database

python scripts/init_db.py --url postgresql://user:pass@localhost/db

Migration Operations

./scripts/migrate.sh init              # Initialize Alembic
./scripts/migrate.sh create "message"  # Create migration
./scripts/migrate.sh upgrade           # Apply migrations
./scripts/migrate.sh downgrade         # Rollback one
./scripts/migrate.sh current           # Show current
./scripts/migrate.sh history           # Show history
./scripts/migrate.sh test              # Test up & down

Example Models

Use the example models in assets/example-models.py as templates:

• User model with timestamp mixin
• Task model with enums and relationships
• Team model with many-to-many
• Tag system with link tables
• Separate read/write/update models

Copy to your project:

cp assets/example-models.py your-project/app/models.py

Best Practices Checklist

Model Design

• [ ] Use type hints for all fields
• [ ] Separate read/write/update models
• [ ] Use mixins for common fields (timestamps, soft delete)
• [ ] Define indexes on foreign keys and frequently queried columns
• [ ] Use enums for constrained choices
• [ ] Implement proper validation with Pydantic validators

Relationships

• [ ] Use back_populates for bidirectional relationships
• [ ] Create explicit link tables for many-to-many
• [ ] Consider cascade delete behavior
• [ ] Use eager loading to prevent N+1 queries
• [ ] Index foreign key columns

Migrations

• [ ] Always review autogenerated migrations
• [ ] One logical change per migration
• [ ] Test both upgrade and downgrade
• [ ] Use descriptive migration names
• [ ] Never edit applied migrations
• [ ] Add data migrations when changing schemas
• [ ] Backup database before production migrations

Query Optimization

• [ ] Use eager loading (selectinload) for relationships
• [ ] Select only needed columns
• [ ] Use indexes for WHERE/ORDER BY columns
• [ ] Batch operations instead of loops
• [ ] Profile slow queries
• [ ] Use connection pooling

Troubleshooting Guide

Migration Issues

Problem: Alembic doesn't detect model changes

# Solution: Ensure models are imported in env.py
from app.models import User, Task, Team  # Import all models
target_metadata = SQLModel.metadata

Problem: Failed migration

# Check current state
alembic current

# Manually fix issue, then stamp
alembic stamp head

# Or downgrade and retry
alembic downgrade -1
alembic upgrade head

Query Performance

Problem: Slow queries

# Enable query logging
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# Use EXPLAIN ANALYZE
explain = session.exec(text("EXPLAIN ANALYZE SELECT ...")).all()

# Profile queries
# See references/queries-optimization.md for detailed patterns

Problem: N+1 queries

# Use selectinload
statement = select(User).options(selectinload(User.posts))

# Or joinedload
from sqlalchemy.orm import joinedload
statement = select(User).options(joinedload(User.posts))

Production Workflow

Development

1. Modify SQLModel models 2. Generate migration: ./scripts/migrate.sh create "description" 3. Review generated migration file 4. Test migration: ./scripts/migrate.sh test 5. Commit migration file

Staging

1. Deploy application code 2. Run migrations: alembic upgrade head 3. Verify data integrity 4. Test application

Production

1. Backup database: pg_dump mydb > backup.sql 2. Deploy in maintenance window 3. Run migrations: alembic upgrade head 4. Monitor logs and metrics 5. Verify application functionality

Zero-Downtime Migration Strategy

For large production databases:

# Phase 1: Add new column (nullable)
def upgrade():
    op.add_column('user', sa.Column('new_email', sa.String(), nullable=True))

# Deploy app version that writes to both columns

# Phase 2: Backfill data
def upgrade():
    op.execute("UPDATE user SET new_email = email WHERE new_email IS NULL")

# Phase 3: Make non-nullable
def upgrade():
    op.alter_column('user', 'new_email', nullable=False)

# Deploy app version that reads from new column

# Phase 4: Drop old column
def upgrade():
    op.drop_column('user', 'email')

Additional Resources

• Advanced Patterns: See references/advanced-models.md for inheritance, polymorphism, composite keys
• Migration Guide: See references/migrations.md for Alembic mastery
• Query Optimization: See references/queries-optimization.md for performance tuning

This skill provides everything needed for professional SQLModel development and database management.

Advanced SQLModel Patterns

Table of Contents

1. Model Definition Patterns 2. Relationships (One-to-Many, Many-to-Many) 3. Inheritance and Polymorphism 4. Composite Keys and Constraints 5. Custom Field Types 6. Table Partitioning Strategies

---

1. Model Definition Patterns

Basic Model with Validation

from sqlmodel import Field, SQLModel
from pydantic import EmailStr, validator
from datetime import datetime
from typing import Optional

class User(SQLModel, table=True):
    """User model with validation and defaults"""
    __tablename__ = "users"

    id: Optional[int] = Field(default=None, primary_key=True)
    username: str = Field(index=True, unique=True, min_length=3, max_length=50)
    email: EmailStr = Field(unique=True)
    full_name: str
    hashed_password: str
    is_active: bool = Field(default=True)
    is_superuser: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)
    updated_at: datetime = Field(default_factory=datetime.utcnow)

    @validator('username')
    def username_alphanumeric(cls, v):
        assert v.isalnum(), 'must be alphanumeric'
        return v

Separate Read/Write Models

from sqlmodel import Field, SQLModel
from typing import Optional

# Base model (shared fields)
class UserBase(SQLModel):
    username: str = Field(index=True, unique=True)
    email: str = Field(unique=True)
    full_name: str
    is_active: bool = True

# Table model (database schema)
class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    hashed_password: str

# Create model (API input)
class UserCreate(UserBase):
    password: str

# Read model (API output - excludes password)
class UserRead(UserBase):
    id: int

# Update model (partial updates)
class UserUpdate(SQLModel):
    username: Optional[str] = None
    email: Optional[str] = None
    full_name: Optional[str] = None
    is_active: Optional[bool] = None

Timestamp Mixin

from datetime import datetime
from sqlmodel import Field, SQLModel

class TimestampMixin(SQLModel):
    """Mixin to add timestamp fields"""
    created_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)
    updated_at: datetime = Field(default_factory=datetime.utcnow, nullable=False)

class SoftDeleteMixin(SQLModel):
    """Mixin for soft delete functionality"""
    deleted_at: Optional[datetime] = Field(default=None, nullable=True)
    is_deleted: bool = Field(default=False)

class Task(TimestampMixin, SoftDeleteMixin, SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    description: Optional[str] = None
    completed: bool = Field(default=False)

---

2. Relationships

One-to-Many Relationship

from typing import List, Optional
from sqlmodel import Field, Relationship, SQLModel

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

    # Relationship: one team has many heroes
    heroes: List["Hero"] = Relationship(back_populates="team")

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    secret_name: str
    age: Optional[int] = None

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

    # Relationship: many heroes belong to one team
    team: Optional[Team] = Relationship(back_populates="heroes")

Querying with relationships:

from sqlmodel import Session, select

# Eager loading (prevents N+1 queries)
from sqlalchemy.orm import selectinload

statement = select(Team).options(selectinload(Team.heroes))
teams = session.exec(statement).all()

for team in teams:
    print(f"Team: {team.name}")
    for hero in team.heroes:
        print(f"  - {hero.name}")

Many-to-Many Relationship

from typing import List, Optional
from sqlmodel import Field, Relationship, SQLModel

# Link table (association table)
class HeroTeamLink(SQLModel, table=True):
    """Link table for many-to-many relationship"""
    __tablename__ = "hero_team_link"

    hero_id: Optional[int] = Field(
        default=None,
        foreign_key="hero.id",
        primary_key=True
    )
    team_id: Optional[int] = Field(
        default=None,
        foreign_key="team.id",
        primary_key=True
    )
    # Additional fields on the link
    joined_at: datetime = Field(default_factory=datetime.utcnow)
    role: Optional[str] = None

class Hero(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    secret_name: str

    # Many-to-many relationship
    teams: List["Team"] = Relationship(
        back_populates="heroes",
        link_model=HeroTeamLink
    )

class Team(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    headquarters: str

    # Many-to-many relationship
    heroes: List[Hero] = Relationship(
        back_populates="teams",
        link_model=HeroTeamLink
    )

Self-Referential Relationship

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

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

    # Relationships
    manager: Optional["User"] = Relationship(
        back_populates="subordinates",
        sa_relationship_kwargs={"remote_side": "User.id"}
    )
    subordinates: List["User"] = Relationship(back_populates="manager")

Cascading Deletes

from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship as sa_relationship

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

    # Cascade delete posts when user is deleted
    posts: List["Post"] = Relationship(
        back_populates="author",
        sa_relationship_kwargs={"cascade": "all, delete-orphan"}
    )

class Post(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    content: str
    user_id: int = Field(foreign_key="user.id")

    author: User = Relationship(back_populates="posts")

---

3. Inheritance and Polymorphism

Single Table Inheritance

from sqlmodel import Field, SQLModel
from typing import Optional

class Person(SQLModel, table=True):
    """Base person table with discriminator"""
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    type: str  # Discriminator column

    # Employee-specific fields (nullable for non-employees)
    employee_id: Optional[str] = None
    department: Optional[str] = None

    # Customer-specific fields (nullable for non-customers)
    customer_number: Optional[str] = None
    loyalty_points: Optional[int] = None

# Create views/models for specific types
class EmployeeCreate(SQLModel):
    name: str
    employee_id: str
    department: str

class CustomerCreate(SQLModel):
    name: str
    customer_number: str
    loyalty_points: int = 0

Joined Table Inheritance (Better approach)

from sqlmodel import Field, SQLModel
from typing import Optional

# Base table
class Person(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    name: str
    email: str
    created_at: datetime = Field(default_factory=datetime.utcnow)

# Separate tables for subtypes
class Employee(SQLModel, table=True):
    id: Optional[int] = Field(default=None, foreign_key="person.id", primary_key=True)
    employee_id: str = Field(unique=True)
    department: str
    salary: float

class Customer(SQLModel, table=True):
    id: Optional[int] = Field(default=None, foreign_key="person.id", primary_key=True)
    customer_number: str = Field(unique=True)
    loyalty_points: int = Field(default=0)

---

4. Composite Keys and Constraints

Composite Primary Key

from sqlmodel import Field, SQLModel
from typing import Optional

class UserRole(SQLModel, table=True):
    """User role assignment with composite primary key"""
    __tablename__ = "user_roles"

    user_id: int = Field(foreign_key="user.id", primary_key=True)
    role_id: int = Field(foreign_key="role.id", primary_key=True)
    granted_at: datetime = Field(default_factory=datetime.utcnow)
    granted_by: Optional[int] = Field(foreign_key="user.id")

Unique Constraints

from sqlalchemy import UniqueConstraint

class Product(SQLModel, table=True):
    __tablename__ = "products"
    __table_args__ = (
        UniqueConstraint('sku', 'warehouse_id', name='unique_product_warehouse'),
    )

    id: Optional[int] = Field(default=None, primary_key=True)
    sku: str = Field(index=True)
    name: str
    warehouse_id: int = Field(foreign_key="warehouse.id")

Check Constraints

from sqlalchemy import CheckConstraint

class BankAccount(SQLModel, table=True):
    __table_args__ = (
        CheckConstraint('balance >= 0', name='check_positive_balance'),
        CheckConstraint('overdraft_limit >= 0', name='check_positive_overdraft'),
    )

    id: Optional[int] = Field(default=None, primary_key=True)
    account_number: str = Field(unique=True)
    balance: float = Field(default=0.0)
    overdraft_limit: float = Field(default=0.0)

---

5. Custom Field Types

Enum Fields

from enum import Enum
from sqlmodel import Field, SQLModel

class TaskStatus(str, Enum):
    TODO = "todo"
    IN_PROGRESS = "in_progress"
    DONE = "done"

class TaskPriority(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"

class Task(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    title: str
    status: TaskStatus = Field(default=TaskStatus.TODO)
    priority: TaskPriority = Field(default=TaskPriority.MEDIUM)

JSON Fields

from typing import Optional, Dict, Any
from sqlalchemy import JSON, Column

class UserPreferences(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(foreign_key="user.id", unique=True)

    # JSON field for flexible settings
    settings: Dict[str, Any] = Field(
        default_factory=dict,
        sa_column=Column(JSON)
    )

    # Example usage:
    # settings = {
    #     "theme": "dark",
    #     "notifications": {"email": True, "push": False},
    #     "language": "en"
    # }

Array Fields (PostgreSQL)

from typing import List
from sqlalchemy import ARRAY, String, Column

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

    # Array field (PostgreSQL only)
    tags: List[str] = Field(
        default_factory=list,
        sa_column=Column(ARRAY(String))
    )

---

6. Index Strategies

Single Column Index

class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(index=True, unique=True)  # Creates index automatically
    username: str = Field(index=True)

Composite Index

from sqlalchemy import Index

class Order(SQLModel, table=True):
    __tablename__ = "orders"
    __table_args__ = (
        Index('idx_user_created', 'user_id', 'created_at'),
        Index('idx_status_priority', 'status', 'priority'),
    )

    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int = Field(foreign_key="user.id")
    status: str
    priority: int
    created_at: datetime = Field(default_factory=datetime.utcnow)

Partial Index (PostgreSQL)

from sqlalchemy import Index

class Task(SQLModel, table=True):
    __tablename__ = "tasks"
    __table_args__ = (
        # Index only non-completed tasks
        Index(
            'idx_active_tasks',
            'user_id',
            'created_at',
            postgresql_where=text('completed = false')
        ),
    )

    id: Optional[int] = Field(default=None, primary_key=True)
    user_id: int
    title: str
    completed: bool = Field(default=False)
    created_at: datetime = Field(default_factory=datetime.utcnow)

---

Best Practices

1. Always Use Type Hints

# Good
class User(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    email: str = Field(unique=True)
    age: int

# Bad
class User(SQLModel, table=True):
    id = Field(default=None, primary_key=True)
    email = Field(unique=True)
    age: int

2. Use Indexes Wisely

• Index foreign keys
• Index columns used in WHERE clauses
• Index columns used in ORDER BY
• Don't over-index (slows down writes)

3. Separate Read/Write Models

• Use different models for API input/output
• Prevents accidental exposure of sensitive fields
• Allows different validation rules

4. Use Relationships Carefully

• Prefer lazy loading for large collections
• Use selectinload() for eager loading to prevent N+1
• Consider back_populates for bidirectional relationships

5. Naming Conventions

# Tables: plural, lowercase with underscores
__tablename__ = "user_preferences"

# Columns: lowercase with underscores
created_at: datetime
is_active: bool

# Foreign keys: singular_table_id
user_id: int = Field(foreign_key="user.id")

Comprehensive Alembic Migrations Guide

Table of Contents

1. Alembic Setup and Configuration 2. Creating Migrations 3. Schema Changes Patterns 4. Data Migrations 5. Migration Best Practices 6. Rollback Strategies 7. Production Migration Workflow 8. Troubleshooting

---

1. Alembic Setup and Configuration

Initial Setup

# Install Alembic
pip install alembic

# Initialize Alembic in your project
alembic init alembic

# This creates:
# alembic/
# ├── env.py          # Migration environment
# ├── script.py.mako  # Migration template
# └── versions/       # Migration files
# alembic.ini         # Alembic configuration

Configure alembic.ini

# alembic.ini
[alembic]
script_location = alembic
prepend_sys_path = .

# Database URL (use environment variable in production)
sqlalchemy.url = postgresql://user:password@localhost/dbname

# Or use environment variable:
# sqlalchemy.url = ${DATABASE_URL}

file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d%%(minute).2d-%%(rev)s_%%(slug)s

Configure env.py for SQLModel

# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
import os

# Import your SQLModel models
from app.models import SQLModel  # Your base model
from app.database import get_database_url

# Alembic Config object
config = context.config

# Override sqlalchemy.url from environment
config.set_main_option("sqlalchemy.url", get_database_url())

# Interpret config file for Python logging
if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# Set target metadata for 'autogenerate'
target_metadata = SQLModel.metadata


def run_migrations_offline() -> None:
    """Run migrations in 'offline' mode."""
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )

    with context.begin_transaction():
        context.run_migrations()


def run_migrations_online() -> None:
    """Run migrations in 'online' mode."""
    connectable = engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

---

2. Creating Migrations

Autogenerate Migration (Recommended)

# Generate migration from model changes
alembic revision --autogenerate -m "Add user table"

# Result: alembic/versions/2024_01_15_1430-abc123_add_user_table.py

Generated migration file:

"""Add user table

Revision ID: abc123
Revises:
Create Date: 2024-01-15 14:30:00

"""
from alembic import op
import sqlalchemy as sa
import sqlmodel


# revision identifiers
revision = 'abc123'
down_revision = None
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.create_table(
        'user',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('username', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
        sa.Column('email', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
        sa.Column('hashed_password', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_user_email'), 'user', ['email'], unique=True)
    op.create_index(op.f('ix_user_username'), 'user', ['username'], unique=False)


def downgrade() -> None:
    op.drop_index(op.f('ix_user_username'), table_name='user')
    op.drop_index(op.f('ix_user_email'), table_name='user')
    op.drop_table('user')

Manual Migration

# Create empty migration file
alembic revision -m "custom migration"

---

3. Schema Changes Patterns

Adding a Column

def upgrade() -> None:
    op.add_column(
        'user',
        sa.Column('phone_number', sa.String(20), nullable=True)
    )

def downgrade() -> None:
    op.drop_column('user', 'phone_number')

Adding a Non-Nullable Column with Default

def upgrade() -> None:
    # Step 1: Add column as nullable
    op.add_column(
        'user',
        sa.Column('role', sa.String(50), nullable=True)
    )

    # Step 2: Set default value for existing rows
    op.execute("UPDATE user SET role = 'user' WHERE role IS NULL")

    # Step 3: Make column non-nullable
    op.alter_column('user', 'role', nullable=False)

def downgrade() -> None:
    op.drop_column('user', 'role')

Renaming a Column

def upgrade() -> None:
    op.alter_column(
        'user',
        'name',
        new_column_name='full_name'
    )

def downgrade() -> None:
    op.alter_column(
        'user',
        'full_name',
        new_column_name='name'
    )

Changing Column Type

def upgrade() -> None:
    # PostgreSQL
    op.alter_column(
        'user',
        'age',
        type_=sa.String(3),
        postgresql_using='age::text'
    )

    # SQLite (requires recreation)
    # See "Complex SQLite Migrations" below

def downgrade() -> None:
    op.alter_column(
        'user',
        'age',
        type_=sa.Integer()
    )

Adding Foreign Key

def upgrade() -> None:
    op.add_column(
        'post',
        sa.Column('user_id', sa.Integer(), nullable=True)
    )

    op.create_foreign_key(
        'fk_post_user_id',  # Constraint name
        'post',              # Source table
        'user',              # Referenced table
        ['user_id'],         # Source columns
        ['id'],              # Referenced columns
        ondelete='CASCADE'   # Optional: cascade delete
    )

def downgrade() -> None:
    op.drop_constraint('fk_post_user_id', 'post', type_='foreignkey')
    op.drop_column('post', 'user_id')

Adding Index

def upgrade() -> None:
    op.create_index(
        'idx_user_email_username',
        'user',
        ['email', 'username'],
        unique=False
    )

def downgrade() -> None:
    op.drop_index('idx_user_email_username', table_name='user')

Adding Unique Constraint

def upgrade() -> None:
    op.create_unique_constraint(
        'uq_user_email',
        'user',
        ['email']
    )

def downgrade() -> None:
    op.drop_constraint('uq_user_email', 'user', type_='unique')

Creating a New Table

def upgrade() -> None:
    op.create_table(
        'task',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(200), nullable=False),
        sa.Column('description', sa.Text(), nullable=True),
        sa.Column('completed', sa.Boolean(), nullable=False, server_default='false'),
        sa.Column('user_id', sa.Integer(), nullable=False),
        sa.Column('created_at', sa.DateTime(), nullable=False),
        sa.ForeignKeyConstraint(['user_id'], ['user.id'], ondelete='CASCADE'),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index('idx_task_user_id', 'task', ['user_id'])

def downgrade() -> None:
    op.drop_index('idx_task_user_id', table_name='task')
    op.drop_table('task')

---

4. Data Migrations

Simple Data Update

from alembic import op
from sqlalchemy import text

def upgrade() -> None:
    # Update data
    op.execute(
        text("UPDATE user SET is_active = true WHERE is_active IS NULL")
    )

def downgrade() -> None:
    # Revert data (if possible)
    op.execute(
        text("UPDATE user SET is_active = NULL WHERE is_active = true")
    )

Complex Data Migration with SQLModel

from alembic import op
from sqlalchemy.orm import Session
from app.models import User, Task  # Your SQLModel models

def upgrade() -> None:
    # Get database connection
    bind = op.get_bind()
    session = Session(bind=bind)

    # Perform complex data manipulation
    users = session.query(User).all()
    for user in users:
        # Create default task for each user
        task = Task(
            title=f"Welcome task for {user.username}",
            description="Get started with the app",
            user_id=user.id,
            completed=False
        )
        session.add(task)

    session.commit()

def downgrade() -> None:
    bind = op.get_bind()
    session = Session(bind=bind)

    # Delete welcome tasks
    session.query(Task).filter(
        Task.title.like("Welcome task for %")
    ).delete()

    session.commit()

Batch Operations for Large Tables

from alembic import op
from sqlalchemy import text

def upgrade() -> None:
    # Process in batches to avoid locking large tables
    batch_size = 1000
    offset = 0

    while True:
        result = op.execute(
            text(f"""
                UPDATE user
                SET email_verified = false
                WHERE email_verified IS NULL
                AND id IN (
                    SELECT id FROM user
                    WHERE email_verified IS NULL
                    ORDER BY id
                    LIMIT {batch_size} OFFSET {offset}
                )
            """)
        )

        if result.rowcount == 0:
            break

        offset += batch_size

def downgrade() -> None:
    pass  # Not reversible

---

5. Migration Best Practices

1. Always Review Autogenerated Migrations

# Autogenerate creates this:
def upgrade() -> None:
    op.drop_column('user', 'password')

# But you should add data migration:
def upgrade() -> None:
    # Copy data before dropping
    op.add_column('user', sa.Column('hashed_password', sa.String()))
    op.execute("UPDATE user SET hashed_password = password")
    op.drop_column('user', 'password')

2. Use Descriptive Migration Names

# Good
alembic revision --autogenerate -m "add_user_email_verification_fields"
alembic revision --autogenerate -m "create_task_priority_index"

# Bad
alembic revision --autogenerate -m "update"
alembic revision --autogenerate -m "changes"

3. One Logical Change Per Migration

# Good - focused migration
def upgrade() -> None:
    op.add_column('user', sa.Column('email_verified', sa.Boolean()))
    op.add_column('user', sa.Column('email_verified_at', sa.DateTime()))

# Better - split into separate migrations if unrelated
# Migration 1: Add email verification
# Migration 2: Add user roles

4. Test Migrations Both Ways

# Test upgrade
alembic upgrade head

# Test downgrade
alembic downgrade -1

# Test upgrade again
alembic upgrade head

5. Never Edit Applied Migrations

# Never modify a migration that's been applied to production
# Instead, create a new migration to fix issues

---

6. Running Migrations

Basic Commands

# Show current revision
alembic current

# Show migration history
alembic history --verbose

# Upgrade to latest
alembic upgrade head

# Upgrade one step
alembic upgrade +1

# Downgrade one step
alembic downgrade -1

# Downgrade to specific revision
alembic downgrade abc123

# Downgrade all
alembic downgrade base

# Show SQL without executing
alembic upgrade head --sql

# Stamp database at specific revision (mark as applied without running)
alembic stamp head

Migration to Specific Revision

# Upgrade to specific revision
alembic upgrade abc123

# Downgrade to specific revision
alembic downgrade xyz789

---

7. Production Migration Workflow

Pre-Deployment Checklist

# 1. Test migrations locally
alembic upgrade head
alembic downgrade base
alembic upgrade head

# 2. Review all migration files
cat alembic/versions/*.py

# 3. Backup production database
pg_dump mydb > backup_$(date +%Y%m%d_%H%M%S).sql

# 4. Test on staging environment
# Deploy to staging
alembic upgrade head

# 5. Monitor for issues
# Check application logs
# Verify data integrity

Safe Production Migration

# Dockerfile entrypoint
#!/bin/bash
set -e

# Run migrations
echo "Running database migrations..."
alembic upgrade head

# Start application
echo "Starting application..."
uvicorn main:app --host 0.0.0.0 --port 8000

Zero-Downtime Migration Strategy

# Phase 1: Add new column (nullable)
def upgrade() -> None:
    op.add_column('user', sa.Column('new_email', sa.String(), nullable=True))

# Deploy application version that writes to both columns

# Phase 2: Backfill data
def upgrade() -> None:
    op.execute("UPDATE user SET new_email = email WHERE new_email IS NULL")

# Phase 3: Make column non-nullable
def upgrade() -> None:
    op.alter_column('user', 'new_email', nullable=False)

# Deploy application version that reads from new column

# Phase 4: Drop old column
def upgrade() -> None:
    op.drop_column('user', 'email')

# Phase 5: Rename new column
def upgrade() -> None:
    op.alter_column('user', 'new_email', new_column_name='email')

---

8. Troubleshooting

Issue: Alembic Can't Detect Changes

# Problem: Models changed but autogenerate doesn't detect them

# Solution 1: Check env.py imports all models
from app.models import User, Task, Team  # Import all models

# Solution 2: Verify target_metadata is set correctly
target_metadata = SQLModel.metadata

# Solution 3: Use compare_type=True for column type changes
context.configure(
    connection=connection,
    target_metadata=target_metadata,
    compare_type=True
)

Issue: Migration Conflicts

# Multiple heads (branching)
alembic heads

# Merge branches
alembic merge -m "merge branches" head1 head2

Issue: Failed Migration

# Check current state
alembic current

# Manually fix database issue
# Then stamp as if migration succeeded
alembic stamp head

# Or downgrade and retry
alembic downgrade -1
# Fix the issue
alembic upgrade head

Issue: SQLite Limitations

# SQLite doesn't support many ALTER operations
# Use batch operations

from alembic import op
import sqlalchemy as sa

def upgrade() -> None:
    with op.batch_alter_table('user') as batch_op:
        batch_op.add_column(sa.Column('new_field', sa.String()))
        batch_op.alter_column('old_field', new_column_name='renamed_field')

def downgrade() -> None:
    with op.batch_alter_table('user') as batch_op:
        batch_op.drop_column('new_field')
        batch_op.alter_column('renamed_field', new_column_name='old_field')

---

Advanced Patterns

Multiple Database Support

# alembic/env.py
def run_migrations_online() -> None:
    # Get database URLs from environment
    databases = {
        'main': os.getenv('MAIN_DB_URL'),
        'analytics': os.getenv('ANALYTICS_DB_URL'),
    }

    for name, url in databases.items():
        engine = create_engine(url)
        with engine.connect() as connection:
            context.configure(
                connection=connection,
                target_metadata=target_metadata
            )
            with context.begin_transaction():
                context.run_migrations()

Custom Migration Template

# alembic/script.py.mako
"""${message}

Revision ID: ${up_revision}
Revises: ${down_revision | comma,n}
Create Date: ${create_date}
Author: ${author}  # Custom field
Ticket: ${ticket}  # Custom field
"""
from alembic import op
import sqlalchemy as sa
${imports if imports else ""}

revision = ${repr(up_revision)}
down_revision = ${repr(down_revision)}
branch_labels = ${repr(branch_labels)}
depends_on = ${repr(depends_on)}


def upgrade() -> None:
    ${upgrades if upgrades else "pass"}


def downgrade() -> None:
    ${downgrades if downgrades else "pass"}

This comprehensive guide covers all aspects of SQLModel migrations with Alembic!

SQLModel Query Patterns and Optimization

Table of Contents

1. Basic Query Patterns 2. Advanced Queries 3. N+1 Query Problem Solutions 4. Query Optimization Techniques 5. Bulk Operations 6. Raw SQL and Performance 7. Testing and Profiling

---

1. Basic Query Patterns

Simple Queries

from sqlmodel import Session, select
from app.models import User

# Get all users
statement = select(User)
users = session.exec(statement).all()

# Get one user
statement = select(User).where(User.id == 1)
user = session.exec(statement).first()

# Get or 404
user = session.get(User, user_id)
if not user:
    raise HTTPException(status_code=404, detail="User not found")

Filtering

# Simple filter
statement = select(User).where(User.is_active == True)

# Multiple conditions (AND)
statement = select(User).where(
    User.is_active == True,
    User.email_verified == True
)

# OR conditions
from sqlalchemy import or_
statement = select(User).where(
    or_(
        User.email == "user@example.com",
        User.username == "user123"
    )
)

# IN clause
user_ids = [1, 2, 3, 4]
statement = select(User).where(User.id.in_(user_ids))

# LIKE clause
statement = select(User).where(User.username.like("%john%"))

# BETWEEN
from datetime import datetime, timedelta
start_date = datetime.utcnow() - timedelta(days=7)
statement = select(User).where(User.created_at.between(start_date, datetime.utcnow()))

# IS NULL / IS NOT NULL
statement = select(User).where(User.deleted_at.is_(None))
statement = select(User).where(User.deleted_at.isnot(None))

Ordering

# Order by single column
statement = select(User).order_by(User.created_at.desc())

# Order by multiple columns
statement = select(User).order_by(
    User.is_active.desc(),
    User.created_at.desc()
)

# Dynamic ordering
from sqlalchemy import asc, desc

order_direction = "desc"
order_column = User.created_at

if order_direction == "desc":
    statement = select(User).order_by(desc(order_column))
else:
    statement = select(User).order_by(asc(order_column))

Pagination

# Offset-based pagination
def get_users(skip: int = 0, limit: int = 100):
    statement = select(User).offset(skip).limit(limit)
    return session.exec(statement).all()

# Cursor-based pagination (better for large datasets)
def get_users_cursor(cursor_id: int = None, limit: int = 100):
    statement = select(User)

    if cursor_id:
        statement = statement.where(User.id > cursor_id)

    statement = statement.order_by(User.id).limit(limit)
    return session.exec(statement).all()

---

2. Advanced Queries

Joins

from sqlmodel import select
from app.models import User, Post

# Inner join
statement = (
    select(User, Post)
    .join(Post, User.id == Post.user_id)
)
results = session.exec(statement).all()

# Left outer join
from sqlalchemy import outerjoin
statement = (
    select(User, Post)
    .outerjoin(Post, User.id == Post.user_id)
)

# Join with filtering
statement = (
    select(User, Post)
    .join(Post)
    .where(Post.published == True)
    .where(User.is_active == True)
)

Aggregations

from sqlalchemy import func

# Count
statement = select(func.count(User.id))
count = session.exec(statement).one()

# Count with filter
statement = select(func.count(User.id)).where(User.is_active == True)

# Group by
statement = (
    select(User.country, func.count(User.id))
    .group_by(User.country)
)
results = session.exec(statement).all()

# Having clause
statement = (
    select(User.country, func.count(User.id))
    .group_by(User.country)
    .having(func.count(User.id) > 10)
)

# Multiple aggregations
statement = (
    select(
        User.country,
        func.count(User.id).label('user_count'),
        func.avg(User.age).label('avg_age'),
        func.max(User.created_at).label('latest_signup')
    )
    .group_by(User.country)
)

Subqueries

# Scalar subquery
subquery = (
    select(func.count(Post.id))
    .where(Post.user_id == User.id)
    .scalar_subquery()
)

statement = select(User, subquery.label('post_count'))

# Subquery in WHERE
active_user_ids = (
    select(User.id)
    .where(User.is_active == True)
    .subquery()
)

statement = select(Post).where(Post.user_id.in_(active_user_ids))

# Common Table Expression (CTE)
recent_posts = (
    select(Post)
    .where(Post.created_at > datetime.utcnow() - timedelta(days=7))
    .cte('recent_posts')
)

statement = (
    select(User)
    .join(recent_posts, User.id == recent_posts.c.user_id)
    .where(recent_posts.c.published == True)
)

Window Functions

from sqlalchemy import func, over

# Row number
statement = select(
    User.username,
    User.country,
    func.row_number().over(
        partition_by=User.country,
        order_by=User.created_at.desc()
    ).label('row_num')
)

# Rank users by post count per country
post_count = (
    select(
        User.id.label('user_id'),
        func.count(Post.id).label('post_count')
    )
    .join(Post)
    .group_by(User.id)
    .subquery()
)

statement = select(
    User.username,
    User.country,
    post_count.c.post_count,
    func.rank().over(
        partition_by=User.country,
        order_by=post_count.c.post_count.desc()
    ).label('rank')
).join(post_count, User.id == post_count.c.user_id)

---

3. N+1 Query Problem Solutions

The Problem

# BAD - N+1 query problem
users = session.exec(select(User)).all()

for user in users:
    # Each iteration triggers a new query!
    posts = user.posts  # SELECT * FROM post WHERE user_id = ?
    print(f"{user.username}: {len(posts)} posts")

# This executes 1 + N queries (1 for users, N for each user's posts)

Solution 1: Eager Loading with selectinload

from sqlalchemy.orm import selectinload

# GOOD - Only 2 queries total
statement = select(User).options(selectinload(User.posts))
users = session.exec(statement).all()

for user in users:
    # No additional query! Data already loaded
    posts = user.posts
    print(f"{user.username}: {len(posts)} posts")

Solution 2: Joined Load

from sqlalchemy.orm import joinedload

# Single query with JOIN
statement = select(User).options(joinedload(User.posts))
users = session.exec(statement).unique().all()

Solution 3: Nested Eager Loading

# Load users with posts and post comments
statement = (
    select(User)
    .options(
        selectinload(User.posts).selectinload(Post.comments)
    )
)
users = session.exec(statement).all()

Solution 4: Manual Join and Group

# For simple cases, manual join can be more efficient
statement = (
    select(
        User.id,
        User.username,
        func.count(Post.id).label('post_count')
    )
    .outerjoin(Post)
    .group_by(User.id, User.username)
)
results = session.exec(statement).all()

---

4. Query Optimization Techniques

Use Indexes

# Add indexes to frequently queried columns
class User(SQLModel, table=True):
    __tablename__ = "users"

    id: int = Field(primary_key=True)
    email: str = Field(index=True, unique=True)  # Single column index
    username: str = Field(index=True)

    # Composite index for common query pattern
    __table_args__ = (
        Index('idx_active_created', 'is_active', 'created_at'),
    )

Select Only Needed Columns

# BAD - loads entire object
users = session.exec(select(User)).all()

# GOOD - select only needed columns
statement = select(User.id, User.username, User.email)
results = session.exec(statement).all()

Use Exists for Checking

from sqlalchemy import exists

# BAD - loads all data just to check
statement = select(User).where(User.email == email)
user = session.exec(statement).first()
if user:
    # exists

# GOOD - only checks existence
statement = select(exists(select(User).where(User.email == email)))
exists_result = session.exec(statement).one()

Batch Queries

# BAD - multiple individual queries
for user_id in user_ids:
    user = session.get(User, user_id)
    process(user)

# GOOD - single batch query
statement = select(User).where(User.id.in_(user_ids))
users = session.exec(statement).all()
for user in users:
    process(user)

---

5. Bulk Operations

Bulk Insert

# Create multiple records efficiently
users = [
    User(username=f"user{i}", email=f"user{i}@example.com")
    for i in range(1000)
]

# Add all at once
session.add_all(users)
session.commit()

# Or use bulk_insert_mappings (faster, bypasses ORM)
user_dicts = [
    {"username": f"user{i}", "email": f"user{i}@example.com"}
    for i in range(1000)
]
session.bulk_insert_mappings(User, user_dicts)
session.commit()

Bulk Update

# Update multiple records
statement = (
    update(User)
    .where(User.is_active == False)
    .values(deleted_at=datetime.utcnow())
)
session.exec(statement)
session.commit()

# Bulk update with mappings
user_updates = [
    {"id": 1, "last_login": datetime.utcnow()},
    {"id": 2, "last_login": datetime.utcnow()},
]
session.bulk_update_mappings(User, user_updates)
session.commit()

Bulk Delete

# Delete multiple records
statement = delete(User).where(User.deleted_at.isnot(None))
session.exec(statement)
session.commit()

# Batch delete (for large sets)
batch_size = 1000
while True:
    statement = (
        delete(User)
        .where(User.is_active == False)
        .limit(batch_size)
    )
    result = session.exec(statement)
    session.commit()

    if result.rowcount < batch_size:
        break

---

6. Raw SQL and Performance

Execute Raw SQL

from sqlalchemy import text

# Read query
statement = text("SELECT * FROM users WHERE is_active = :is_active")
results = session.exec(statement, {"is_active": True}).all()

# Write query
statement = text("""
    UPDATE users
    SET last_login = :timestamp
    WHERE id = :user_id
""")
session.exec(statement, {"timestamp": datetime.utcnow(), "user_id": 1})
session.commit()

Use Raw SQL for Complex Operations

# Complex analytical query
statement = text("""
    WITH monthly_stats AS (
        SELECT
            DATE_TRUNC('month', created_at) as month,
            COUNT(*) as user_count,
            COUNT(CASE WHEN is_active THEN 1 END) as active_count
        FROM users
        GROUP BY DATE_TRUNC('month', created_at)
    )
    SELECT
        month,
        user_count,
        active_count,
        ROUND(100.0 * active_count / user_count, 2) as active_percentage
    FROM monthly_stats
    ORDER BY month DESC
""")
results = session.exec(statement).all()

---

7. Testing and Profiling

Query Logging

import logging

# Enable SQLAlchemy query logging
logging.basicConfig()
logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)

# Now all queries are logged
statement = select(User).where(User.is_active == True)
users = session.exec(statement).all()

# Output:
# SELECT users.id, users.username, users.email, users.is_active
# FROM users WHERE users.is_active = true

Query Profiling

from sqlalchemy import event
from time import time

# Profile slow queries
@event.listens_for(engine, "before_cursor_execute")
def before_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    conn.info.setdefault('query_start_time', []).append(time())

@event.listens_for(engine, "after_cursor_execute")
def after_cursor_execute(conn, cursor, statement, parameters, context, executemany):
    total = time() - conn.info['query_start_time'].pop(-1)
    if total > 0.1:  # Log queries slower than 100ms
        print(f"Slow query ({total:.2f}s): {statement}")

Explain Analyze

# PostgreSQL EXPLAIN ANALYZE
statement = select(User).where(User.is_active == True)
explain = session.exec(
    text(f"EXPLAIN ANALYZE {str(statement.compile(engine))}")
).all()

for row in explain:
    print(row)

Query Count Testing

import pytest
from sqlalchemy import event

@pytest.fixture
def query_counter(session):
    """Count queries executed in test"""
    queries = []

    def receive_after_cursor_execute(conn, cursor, statement, *args):
        queries.append(statement)

    event.listen(engine, "after_cursor_execute", receive_after_cursor_execute)

    yield queries

    event.remove(engine, "after_cursor_execute", receive_after_cursor_execute)

def test_no_n_plus_one(session, query_counter):
    # Load users with posts (should be 2 queries)
    statement = select(User).options(selectinload(User.posts))
    users = session.exec(statement).all()

    # Access posts (should not trigger additional queries)
    for user in users:
        _ = user.posts

    # Assert only 2 queries were executed
    assert len(query_counter) == 2

---

Best Practices Summary

1. Always use indexes on foreign keys and frequently queried columns 2. Use eager loading (selectinload/joinedload) to prevent N+1 queries 3. Select only needed columns when possible 4. Use pagination for large result sets 5. Batch operations instead of loops 6. Profile queries in development 7. Use connection pooling in production 8. Monitor slow queries with logging 9. Cache expensive queries when appropriate 10. Test query counts in integration tests