db-migrations

name	db-migrations
description	Alembic database migration workflows for FastAPI/SQLModel applications. Use when creating migrations, running schema changes, handling failed migrations, switching database engines, or resetting local state. Triggers on alembic, migration, schema change, database, upgrade, downgrade.

Database Migrations

Alembic database migration workflows for the Vibe4Vets backend application.

Core Setup

Working directory: backend/ Migration tool: Alembic with SQLModel Database: PostgreSQL (production), SQLite (optional local)

Primary Workflows

Creating a Migration

cd backend

# 1. Make changes to SQLModel models in app/models/

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

# 3. Review the generated migration in alembic/versions/
# ALWAYS review autogenerated migrations!

# 4. Apply migration
alembic upgrade head

Running Migrations

cd backend

# Apply all pending migrations
alembic upgrade head

# Upgrade to specific revision
alembic upgrade abc123

# Downgrade one revision
alembic downgrade -1

# Downgrade to specific revision
alembic downgrade abc123

# Check current revision
alembic current

# View migration history
alembic history

Common Migration Operations

Add Column

def upgrade() -> None:
    op.add_column('resource', sa.Column('phone', sa.String(), nullable=True))

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

Add Column with Default

def upgrade() -> None:
    op.add_column('resource', sa.Column('trust_score', sa.Float(), nullable=False, server_default='0.0'))
    # Remove default after column is populated
    op.alter_column('resource', 'trust_score', server_default=None)

def downgrade() -> None:
    op.drop_column('resource', 'trust_score')

Rename Column

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

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

Add Index

def upgrade() -> None:
    op.create_index('ix_resource_category', 'resource', ['category'])

def downgrade() -> None:
    op.drop_index('ix_resource_category', table_name='resource')

Add Foreign Key

def upgrade() -> None:
    op.add_column('resource', sa.Column('source_id', sa.Integer(), nullable=True))
    op.create_foreign_key(
        'fk_resource_source',
        'resource', 'source',
        ['source_id'], ['id']
    )

def downgrade() -> None:
    op.drop_constraint('fk_resource_source', 'resource', type_='foreignkey')
    op.drop_column('resource', 'source_id')

Create Table

def upgrade() -> None:
    op.create_table(
        'source',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('name', sa.String(), nullable=False),
        sa.Column('url', sa.String(), nullable=True),
        sa.Column('tier', sa.Integer(), nullable=False, default=4),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
    )
    op.create_index('ix_source_name', 'source', ['name'], unique=True)

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

Data Migrations

For complex data transformations, create manual migrations:

from alembic import op
import sqlalchemy as sa

def upgrade() -> None:
    # Get connection for raw SQL
    conn = op.get_bind()

    # Backfill data
    conn.execute(sa.text("""
        UPDATE resource
        SET trust_score = source.reliability_score * 0.9
        FROM source
        WHERE resource.source_id = source.id
        AND resource.trust_score IS NULL
    """))

def downgrade() -> None:
    conn = op.get_bind()
    conn.execute(sa.text("UPDATE resource SET trust_score = NULL"))

Troubleshooting

Alembic Doesn't Detect Changes

# Ensure all models are imported in alembic/env.py
from app.models.resource import Resource
from app.models.organization import Organization
from app.models.source import Source
# ... import ALL models

target_metadata = SQLModel.metadata

Migration Targeting Error

# Run upgrade first if you see targeting errors
alembic upgrade head

# Then generate new migration
alembic revision --autogenerate -m "description"

Failed Migration

# Check current state
alembic current

# Option 1: Fix issue and retry
# Edit the migration file, then:
alembic upgrade head

# Option 2: Downgrade and retry
alembic downgrade -1
# Fix migration
alembic upgrade head

# Option 3: Stamp if manually fixed
alembic stamp head

SQLite Limitations

SQLite doesn't support all ALTER operations. If you hit constraints:

# Switch to PostgreSQL for development
export DATABASE_URL="postgresql://user:pass@localhost:5432/vibe4vets"
alembic upgrade head

Reset Local Database

# Option 1: Drop and recreate
dropdb vibe4vets
createdb vibe4vets
alembic upgrade head

# Option 2: Downgrade to base
alembic downgrade base
alembic upgrade head

Production Workflow

Pre-Deploy Checklist

• Migration tested locally
• Both upgrade and downgrade work
• No data loss in downgrade
• Migration is reversible
• Backup taken before deploy

Deploy Steps

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

# 2. Deploy new code (but don't restart yet)

# 3. Run migrations
alembic upgrade head

# 4. Verify migration succeeded
alembic current

# 5. Restart application

# 6. Monitor for issues

Rollback Steps

# 1. Stop application

# 2. Downgrade migration
alembic downgrade -1

# 3. Verify
alembic current

# 4. Deploy previous code version

# 5. Restart application

Zero-Downtime Migration Strategy

For large tables or production systems:

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

# Phase 2: Backfill data (separate migration)
def upgrade():
    conn = op.get_bind()
    # Batch update to avoid locks
    conn.execute(sa.text("""
        UPDATE resource SET new_field = old_field
        WHERE new_field IS NULL
        LIMIT 10000
    """))

# Phase 3: Make non-nullable (after app handles both)
def upgrade():
    op.alter_column('resource', 'new_field', nullable=False)

# Phase 4: Drop old column (after app migrated)
def upgrade():
    op.drop_column('resource', 'old_field')

Best Practices

DO

• Always review autogenerated migrations
• Test migrations locally before deploying
• Use descriptive migration names
• One logical change per migration
• Include downgrade logic
• Backup before production migrations

DON'T

• Edit already-applied migrations in shared environments
• Skip reviewing autogenerated code
• Run migrations without backup
• Make migrations irreversible without good reason
• Mix schema and data migrations in one file