Database Migration

Database Migration Skill

Expert guidance for safe database schema changes using Alembic migrations with SQLAlchemy models.

When This Skill Activates

• Adding new database models
• Modifying existing model fields
• Creating foreign key relationships
• Renaming columns or tables
• Dropping columns or tables
• Troubleshooting migration errors
• Rollback scenarios

Core Principle

NEVER modify database models without creating a corresponding Alembic migration.

Migration Workflow

Step 1: Modify the Model

# backend/app/models/person.py

from sqlalchemy import Column, String, DateTime, Boolean
from app.db.base import Base

class Person(Base):
    __tablename__ = "persons"

    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)

    # NEW FIELD - requires migration
    middle_name = Column(String, nullable=True)  # Added

Step 2: Generate Migration

cd /home/user/Autonomous-Assignment-Program-Manager/backend

# Generate autogenerated migration
alembic revision --autogenerate -m "Add middle_name to Person model"

Step 3: Review Generated Migration

# backend/alembic/versions/xxxx_add_middle_name_to_person_model.py

"""Add middle_name to Person model

Revision ID: xxxx
Revises: yyyy
Create Date: 2025-01-15 10:30:00

"""
from alembic import op
import sqlalchemy as sa

# revision identifiers
revision = 'xxxx'
down_revision = 'yyyy'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # ### commands auto generated by Alembic ###
    op.add_column('persons', sa.Column('middle_name', sa.String(), nullable=True))
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic ###
    op.drop_column('persons', 'middle_name')
    # ### end Alembic commands ###

Step 4: Test Migration

# Apply migration
alembic upgrade head

# Verify migration applied
alembic current

# Test rollback
alembic downgrade -1

# Re-apply
alembic upgrade head

Step 5: Commit Both Files

git add backend/app/models/person.py
git add backend/alembic/versions/xxxx_add_middle_name_to_person_model.py
git commit -m "feat: Add middle_name field to Person model"

Migration Patterns

Adding a Nullable Column (Safe)

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

def downgrade() -> None:
    op.drop_column('table_name', 'new_column')

Adding a Non-Nullable Column (Requires Default)

def upgrade() -> None:
    # Add as nullable first
    op.add_column('table_name',
        sa.Column('new_column', sa.String(), nullable=True)
    )
    # Backfill data
    op.execute("UPDATE table_name SET new_column = 'default_value' WHERE new_column IS NULL")
    # Make non-nullable
    op.alter_column('table_name', 'new_column', nullable=False)

def downgrade() -> None:
    op.drop_column('table_name', 'new_column')

Adding an Index

def upgrade() -> None:
    op.create_index(
        'ix_table_column',
        'table_name',
        ['column_name']
    )

def downgrade() -> None:
    op.drop_index('ix_table_column', 'table_name')

Adding a Foreign Key

def upgrade() -> None:
    op.add_column('child_table',
        sa.Column('parent_id', sa.String(), nullable=True)
    )
    op.create_foreign_key(
        'fk_child_parent',
        'child_table', 'parent_table',
        ['parent_id'], ['id'],
       
    )

def downgrade() -> None:
    op.drop_constraint('fk_child_parent', 'child_table', type_='foreignkey')
    op.drop_column('child_table', 'parent_id')

Renaming a Column

def upgrade() -> None:
    op.alter_column('table_name', 'old_name', new_column_name='new_name')

def downgrade() -> None:
    op.alter_column('table_name', 'new_name', new_column_name='old_name')

Dropping a Column (DANGER)

def upgrade() -> None:
    # CAUTION: Data will be lost!
    op.drop_column('table_name', 'column_to_drop')

def downgrade() -> None:
    # Cannot restore data - only recreate column
    op.add_column('table_name',
        sa.Column('column_to_drop', sa.String(), nullable=True)
    )

Creating a New Table

def upgrade() -> None:
    op.create_table(
        'new_table',
        sa.Column('id', sa.String(), primary_key=True),
        sa.Column('name', sa.String(), nullable=False),
        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()),
        sa.Column('updated_at', sa.DateTime(),
    )
    op.create_index('ix_new_table_name', 'new_table', ['name'])

def downgrade() -> None:
    op.drop_index('ix_new_table_name', 'new_table')
    op.drop_table('new_table')

Migration Safety Checklist

Before Creating Migration

• Model change is complete and tested locally
• Understand impact on existing data
• Have rollback plan ready
• Check for circular dependencies
• Verify foreign key relationships

After Creating Migration

• Review autogenerated code for accuracy
• Test upgrade path: alembic upgrade head
• Test downgrade path: alembic downgrade -1
• Verify data integrity after migration
• Check application still works

Before Merging

• Migration file committed with model change
• Tests pass with new schema
• No breaking changes to existing APIs
• Documentation updated if needed

Common Issues and Fixes

Issue: Autogenerate Missed a Change

# Check what autogenerate sees
alembic revision --autogenerate -m "test" --sql

# If empty, model may not be imported in env.py
# Check: backend/alembic/env.py

Issue: Migration Order Conflict

# Check current heads
alembic heads

# If multiple heads, create merge migration
alembic merge -m "Merge heads" head1 head2

Issue: Migration Failed Partway

# Check current state
alembic current

# May need manual intervention
# Connect to database and check schema state
docker-compose exec db psql -U scheduler -d residency_scheduler
\dt  # List tables
\d table_name  # Describe table

Issue: Foreign Key Constraint Error

# Add column without FK first, then add constraint
def upgrade() -> None:
    # 1. Add column
    op.add_column('table', sa.Column('fk_id', sa.String()))

    # 2. Backfill valid values
    op.execute("""
        UPDATE table
        SET fk_id = (SELECT id FROM parent LIMIT 1)
        WHERE fk_id IS NULL
    """)

    # 3. Add constraint
    op.create_foreign_key('fk_name', 'table', 'parent', ['fk_id'], ['id'])

Alembic Commands Reference

# Show current revision
alembic current

# Show revision history
alembic history

# Upgrade to latest
alembic upgrade head

# Upgrade one step
alembic upgrade +1

# Downgrade one step
alembic downgrade -1

# Downgrade to specific revision
alembic downgrade <revision_id>

# Show SQL without applying
alembic upgrade head --sql

# Create empty migration
alembic revision -m "Description"

# Create autogenerated migration
alembic revision --autogenerate -m "Description"

# Stamp database (mark as migrated without running)
alembic stamp head

Data Migration Examples

Backfilling Data

from sqlalchemy.orm import Session
from app.models.person import Person

def upgrade() -> None:
    # Schema change
    op.add_column('persons', sa.Column('full_name', sa.String()))

    # Data migration using raw SQL (preferred for large datasets)
    op.execute("""
        UPDATE persons
        SET full_name = first_name || ' ' || last_name
    """)

Complex Data Transformation

from alembic import op
from sqlalchemy import orm
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

# Define model as it exists at this migration point
class Person(Base):
    __tablename__ = 'persons'
    id = sa.Column(sa.String, primary_key=True)
    old_field = sa.Column(sa.String)
    new_field = sa.Column(sa.String)

def upgrade() -> None:
    # Add new column
    op.add_column('persons', sa.Column('new_field', sa.String()))

    # Use session for complex logic
    bind = op.get_bind()
    session = orm.Session(bind=bind)

    for person in session.query(Person).all():
        person.new_field = transform(person.old_field)

    session.commit()

Examples

Example 1: Adding Nullable Column with Backfill

Context: Adding middle_name field to Person model

Input:

# Requirement: Add optional middle name field to Person model
# Model: backend/app/models/person.py
class Person(Base):
    __tablename__ = "persons"
    id = Column(String, primary_key=True)
    first_name = Column(String, nullable=False)
    last_name = Column(String, nullable=False)
    # Need to add: middle_name

Process:

1. Modify model to add field:

   middle_name = Column(String, nullable=True)
   

2. Generate migration:

   cd backend
   alembic revision --autogenerate -m "Add middle_name to Person"
   

3. Review generated migration:

   def upgrade() -> None:
       op.add_column('persons', sa.Column('middle_name', sa.String(), nullable=True))
   
   def downgrade() -> None:
       op.drop_column('persons', 'middle_name')
   

4. Test migration:

   alembic upgrade head
   alembic current  # Verify applied
   alembic downgrade -1  # Test rollback
   alembic upgrade head  # Re-apply
   

Output:

Migration created: backend/alembic/versions/abc123_add_middle_name_to_person.py
Status: READY TO COMMIT
Files to commit:
  - backend/app/models/person.py
  - backend/alembic/versions/abc123_add_middle_name_to_person.py

Example 2: Adding Non-Nullable Column (Three-Phase Migration)

Context: Adding required email_verified boolean to User model

Input:

# Requirement: All users must have email_verified status (non-nullable)
# Current state: Field doesn't exist
# Problem: Can't add non-nullable column to table with existing rows

Process:

1. Phase 1: Add as nullable

   # Migration 1: abc123_add_email_verified_nullable.py
   def upgrade() -> None:
       op.add_column('users', sa.Column('email_verified', sa.Boolean(), nullable=True))
   

2. Phase 2: Backfill data

   # Migration 2: abc124_backfill_email_verified.py
   def upgrade() -> None:
       # Set default value for existing rows
       op.execute("UPDATE users SET email_verified = false WHERE email_verified IS NULL")
   

3. Phase 3: Make non-nullable

   # Migration 3: abc125_make_email_verified_required.py
   def upgrade() -> None:
       op.alter_column('users', 'email_verified', nullable=False)
   

Output:

Three-phase migration created:
  1. abc123_add_email_verified_nullable.py - Add column
  2. abc124_backfill_email_verified.py - Set defaults
  3. abc125_make_email_verified_required.py - Enforce constraint

Commit all three together to ensure clean upgrade path.

Example 3: Foreign Key Migration with Data Integrity

Context: Linking Assignment to Rotation via foreign key

Input:

# Requirement: Assignments must reference valid Rotations
# Current: assignment.rotation_name (string, no FK constraint)
# Target: assignment.rotation_id (FK to rotations.id)

Process:

1. Add new column (nullable first)

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

2. Backfill from existing data

   def upgrade() -> None:
       # Map rotation names to IDs
       op.execute("""
           UPDATE assignments a
           SET rotation_id = r.id
           FROM rotations r
           WHERE a.rotation_name = r.name
       """)
   
       # Identify orphaned records
       op.execute("""
           -- Log assignments with invalid rotation names
           SELECT id, rotation_name FROM assignments WHERE rotation_id IS NULL
       """)
   

3. Add foreign key constraint

   def upgrade() -> None:
       # Make non-null
       op.alter_column('assignments', 'rotation_id', nullable=False)
   
       # Add FK constraint
       op.create_foreign_key(
           'fk_assignment_rotation',
           'assignments', 'rotations',
           ['rotation_id'], ['id'],
            # Prevent deletion of rotations in use
       )
   
       # Drop old column after verification
       op.drop_column('assignments', 'rotation_name')
   

Output:

Foreign key migration successful:
  - Added rotation_id column
  - Backfilled 1,247 assignments
  - Created FK constraint
  - Removed deprecated rotation_name column

⚠️  Warning: 3 assignments had invalid rotation names and were not migrated.
    See manual review required: assignment IDs [47, 102, 381]

Common Failure Modes

Integration with Other Skills

With code-review

Coordination: Review migration code before commit

1. database-migration creates migration file
2. code-review validates:
   - upgrade() and downgrade() are inverses
   - No data loss in downgrade path (or documented)
   - SQL is safe (no raw user input)
3. If approved, commit both model and migration

With automated-code-fixer

Coordination: Fix migration issues automatically where safe

1. database-migration detects issue (e.g., missing import)
2. automated-code-fixer applies fix if trivial
3. Re-run migration generation
4. If complex, escalate to human

With test-writer

Coordination: Add tests for migration

# Example migration test
def test_migration_adds_middle_name():
    # Downgrade to before migration
    alembic_downgrade()

    # Verify column doesn't exist
    with pytest.raises(OperationalError):
        db.execute("SELECT middle_name FROM persons LIMIT 1")

    # Apply migration
    alembic_upgrade()

    # Verify column exists
    result = db.execute("SELECT middle_name FROM persons LIMIT 1")
    assert result is not None

With safe-schedule-generation

Coordination: Database changes require schedule regeneration

1. database-migration changes Assignment model
2. safe-schedule-generation must adapt to new schema
3. Run schedule generation test to verify compatibility
4. If breaking change, coordinate deployment

Validation Checklist

After creating a migration, verify:

• Model Change: Model file updated with new field/table
• Migration Generated: Alembic created migration file in versions/
• Upgrade Path: alembic upgrade head succeeds
• Downgrade Path: alembic downgrade -1 succeeds
• Data Integrity: No data lost during upgrade
• Constraints Valid: Foreign keys and indexes applied correctly
• Application Works: Run app and verify functionality
• Tests Pass: All existing tests still pass
• Migration Tests: Added tests for migration if complex
• Rollback Plan: Documented how to recover if migration fails in production
• Both Files Committed: Model and migration in same commit
• No Hardcoded Values: Use parameters, not literal SQL strings
• Idempotent: Running migration twice doesn't break

Escalation Rules

ALWAYS escalate to human when:

1. Dropping tables or columns with data
2. Changing primary keys
3. Large-scale data migration (>100k rows)
4. Production database migration
5. Modifying ACGME-related tables
6. Changes affecting audit trails
7. Unclear rollback path

Can handle automatically:

1. Adding nullable columns
2. Adding indexes
3. Creating new tables
4. Adding foreign keys to empty columns
5. Renaming columns (with verification)

Workflow Diagram

┌─────────────────────────────────────────────────────────────────┐
│              DATABASE MIGRATION WORKFLOW                        │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  STEP 1: Modify Model                                           │
│  ┌────────────────────────────────────────────────────┐         │
│  │ Add/modify/remove columns in SQLAlchemy model      │         │
│  │ Update relationships if needed                     │         │
│  └────────────────────────────────────────────────────┘         │
│                         ↓                                       │
│  STEP 2: Generate Migration                                     │
│  ┌────────────────────────────────────────────────────┐         │
│  │ alembic revision --autogenerate -m "description"   │         │
│  │ Creates new version file in alembic/versions/      │         │
│  └────────────────────────────────────────────────────┘         │
│                         ↓                                       │
│  STEP 3: Review Generated Migration (CRITICAL!)                 │
│  ┌────────────────────────────────────────────────────┐         │
│  │ Check upgrade() logic is correct                   │         │
│  │ Check downgrade() reverses upgrade                 │         │
│  │ Add data migration if needed                       │         │
│  │ ⚠️ Autogenerate can miss changes                   │         │
│  └────────────────────────────────────────────────────┘         │
│                         ↓                                       │
│  STEP 4: Test Migration (Both Directions)                       │
│  ┌────────────────────────────────────────────────────┐         │
│  │ alembic upgrade head    # Apply migration          │         │
│  │ Verify schema changed                              │         │
│  │ alembic downgrade -1    # Test rollback            │         │
│  │ alembic upgrade head    # Re-apply                 │         │
│  └────────────────────────────────────────────────────┘         │
│                         ↓                                       │
│  STEP 5: Verify Application Works                               │
│  ┌────────────────────────────────────────────────────┐         │
│  │ Run application with new schema                    │         │
│  │ Run full test suite                                │         │
│  │ Check for breaking changes                         │         │
│  └────────────────────────────────────────────────────┘         │
│                         ↓                                       │
│  STEP 6: Commit Both Files Together                             │
│  ┌────────────────────────────────────────────────────┐         │
│  │ git add models/*.py alembic/versions/*.py          │         │
│  │ git commit -m "feat: add field X to model Y"       │         │
│  │ ⚠️ NEVER commit model without migration            │         │
│  └────────────────────────────────────────────────────┘         │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

Concrete Usage Example: Adding Credential Tracking

Scenario: Add credential tracking fields to Person model for HIPAA/training compliance.

Complete Migration Walkthrough

Step 1: Modify the Model

# backend/app/models/person.py

from sqlalchemy import Column, String, DateTime, Boolean, JSON
from app.db.base import Base


class Person(Base):
    __tablename__ = "persons"

    id = Column(String, primary_key=True)
    name = Column(String, nullable=False)
    email = Column(String, unique=True, nullable=False)
    role = Column(String, nullable=False)

    # NEW FIELDS for credential tracking
    credentials = Column(JSON, nullable=True)  # {credential_type: expiry_date}
    last_training_date = Column(DateTime, nullable=True)
    hipaa_certified = Column(Boolean, default=False, nullable=False)
    cyber_awareness_date = Column(DateTime, nullable=True)

Step 2: Generate Migration

cd /home/user/Autonomous-Assignment-Program-Manager/backend

# Generate autogenerated migration
alembic revision --autogenerate -m "Add credential tracking to Person model"

# Output:
# INFO  [alembic.autogenerate.compare] Detected added column 'persons.credentials'
# INFO  [alembic.autogenerate.compare] Detected added column 'persons.last_training_date'
# INFO  [alembic.autogenerate.compare] Detected added column 'persons.hipaa_certified'
# INFO  [alembic.autogenerate.compare] Detected added column 'persons.cyber_awareness_date'
# Generating /backend/alembic/versions/a1b2c3d4e5f6_add_credential_tracking.py...done

Step 3: Review and Enhance Generated Migration

# backend/alembic/versions/a1b2c3d4e5f6_add_credential_tracking.py

"""Add credential tracking to Person model

Revision ID: a1b2c3d4e5f6
Revises: 9z8y7x6w5v4u
Create Date: 2026-01-01 10:00:00

"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import postgresql

# revision identifiers
revision = 'a1b2c3d4e5f6'
down_revision = '9z8y7x6w5v4u'
branch_labels = None
depends_on = None


def upgrade() -> None:
    # Add new columns (all nullable initially)
    op.add_column('persons', sa.Column('credentials', postgresql.JSON(astext_type=sa.Text()), nullable=True))
    op.add_column('persons', sa.Column('last_training_date', sa.DateTime(), nullable=True))
    op.add_column('persons', sa.Column('cyber_awareness_date', sa.DateTime(), nullable=True))

    # Add boolean with default - must do in two steps for existing rows
    op.add_column('persons', sa.Column('hipaa_certified', sa.Boolean(), nullable=True))

    # ENHANCEMENT: Backfill existing rows with defaults
    op.execute("""
        UPDATE persons
        SET hipaa_certified = FALSE,
            credentials = '{}'::jsonb
        WHERE hipaa_certified IS NULL
    """)

    # Now make hipaa_certified non-nullable
    op.alter_column('persons', 'hipaa_certified',
                   existing_type=sa.Boolean(),
                   nullable=False,
                   server_default='false')


def downgrade() -> None:
    # Remove columns in reverse order
    op.drop_column('persons', 'cyber_awareness_date')
    op.drop_column('persons', 'last_training_date')
    op.drop_column('persons', 'hipaa_certified')
    op.drop_column('persons', 'credentials')

Step 4: Test Migration (Critical!)

cd /home/user/Autonomous-Assignment-Program-Manager/backend

# Check current state
alembic current
# Output: 9z8y7x6w5v4u (previous migration)

# Apply migration
alembic upgrade head
# Output:
# INFO  [alembic.runtime.migration] Running upgrade 9z8y7x6w5v4u -> a1b2c3d4e5f6, Add credential tracking

# Verify schema changed
docker-compose exec db psql -U scheduler -d residency_scheduler -c "\d persons"
# Should show new columns: credentials, last_training_date, hipaa_certified, cyber_awareness_date

# TEST ROLLBACK (critical to verify downgrade works!)
alembic downgrade -1
# Output:
# INFO  [alembic.runtime.migration] Running downgrade a1b2c3d4e5f6 -> 9z8y7x6w5v4u

# Verify columns removed
docker-compose exec db psql -U scheduler -d residency_scheduler -c "\d persons"
# Should NOT show credential tracking columns

# Re-apply migration
alembic upgrade head
# Output:
# INFO  [alembic.runtime.migration] Running upgrade 9z8y7x6w5v4u -> a1b2c3d4e5f6

Step 5: Verify Application Works

# Start backend
docker-compose up -d backend

# Check health
curl http://localhost:8000/health
# Should return: {"status":"healthy","database":"connected"}

# Run tests
pytest

# Check for any breakage
pytest tests/test_person_model.py -v
pytest tests/api/test_people.py -v

# All should PASS

Step 6: Commit Both Files

# Stage both model and migration
git add backend/app/models/person.py
git add backend/alembic/versions/a1b2c3d4e5f6_add_credential_tracking.py

# Commit together
git commit -m "$(cat <<'EOF'
feat: add credential tracking to Person model

Adds fields for tracking HIPAA certification, training dates, and
general credential expiration tracking.

Fields added:
- credentials (JSON): flexible credential storage
- last_training_date: most recent training completion
- hipaa_certified (boolean): HIPAA compliance status
- cyber_awareness_date: annual cyber training date

Migration includes data backfill for existing records.

Related to slot-type invariant requirements for clinic scheduling.
EOF
)"

Failure Mode Handling

Failure Mode 1: Migration Fails Mid-Execution

Symptom:

$ alembic upgrade head

sqlalchemy.exc.OperationalError: (psycopg2.errors.NotNullViolation)
null value in column "hipaa_certified" violates not-null constraint

Root cause: Added NOT NULL column without default to table with existing rows

Recovery:

# 1. Check current revision (might be partially applied)
alembic current
# Output: might show old or new revision

# 2. Check actual database state
docker-compose exec db psql -U scheduler -d residency_scheduler -c "\d persons"
# See if column was partially added

# 3. Stamp to previous revision (mark as not applied)
alembic downgrade -1
# Or if that fails, manually stamp:
alembic stamp 9z8y7x6w5v4u

# 4. Fix the migration
# Edit migration file to add column as nullable first, then backfill, then make NOT NULL
# (See Step 3 example above)

# 5. Re-run migration
alembic upgrade head

Failure Mode 2: Autogenerate Missed a Change

Symptom:

$ alembic revision --autogenerate -m "Add field"
INFO  [alembic.autogenerate] Detected no changes

But you know you modified the model!

Root cause:

• Model not imported in alembic/env.py
• Model in __pycache__ doesn't match source
• Import cycle preventing model load

Recovery:

# 1. Verify model is imported in env.py
grep -r "from app.models" backend/alembic/env.py

# 2. Clear Python cache
find backend -type d -name __pycache__ -exec rm -rf {} +
find backend -type f -name "*.pyc" -delete

# 3. Restart Python environment
deactivate && source venv/bin/activate  # If using venv

# 4. Try again
alembic revision --autogenerate -m "Add field" --sql
# --sql shows what would be generated without creating file

# 5. If still failing, create manual migration
alembic revision -m "Add field"
# Then manually write upgrade() and downgrade()

Failure Mode 3: Can't Rollback - Data Would Be Lost

Symptom:

$ alembic downgrade -1

WARNING: Downgrading will DROP COLUMN with data.
Proceed? [y/N]

Root cause: Downgrade would delete column with user data

Recovery:

# 1. STOP - do not proceed with destructive downgrade

# 2. If you need to rollback, export data first:
docker-compose exec db psql -U scheduler -d residency_scheduler -c \
  "COPY (SELECT id, field_to_preserve FROM persons) TO STDOUT CSV HEADER" \
  > backup_data.csv

# 3. Then proceed with downgrade if necessary:
alembic downgrade -1

# 4. To restore data (if column is re-added later):
# Import backup_data.csv back into database

Prevention: Always backup before risky migrations

./scripts/backup-db.sh --docker

Failure Mode 4: Multiple Migration Heads (Merge Conflict)

Symptom:

$ alembic upgrade head

FAILED: Multiple heads are present
  - a1b2c3d4e5f6 (from branch A)
  - x9y8z7w6v5u4 (from branch B)

Root cause: Two branches created migrations from same parent

Recovery:

# 1. Check heads
alembic heads
# Output:
# a1b2c3d4e5f6 (head)
# x9y8z7w6v5u4 (head)

# 2. Create merge migration
alembic merge -m "Merge migration heads" a1b2c3d4e5f6 x9y8z7w6v5u4

# Output: Creates merge migration that depends on both heads
# Generating /backend/alembic/versions/m1e2r3g4e5d6_merge_heads.py

# 3. Review merge migration (usually auto-generated correctly)
cat backend/alembic/versions/m1e2r3g4e5d6_merge_heads.py

# 4. Apply merge
alembic upgrade head

Failure Mode 5: Foreign Key Constraint Violation

Symptom:

$ alembic upgrade head

sqlalchemy.exc.IntegrityError: foreign key constraint "fk_assignment_person"
violates foreign key constraint

Root cause: Adding FK to column with orphaned data

Recovery:

# Fix the migration to handle orphaned data:

def upgrade() -> None:
    # 1. Add column as nullable first
    op.add_column('assignments', sa.Column('person_id', sa.String(), nullable=True))

    # 2. Identify and handle orphaned records
    op.execute("""
        -- Option A: Delete orphaned assignments
        DELETE FROM assignments WHERE person_id NOT IN (SELECT id FROM persons);

        -- OR Option B: Set to a default person (if one exists)
        UPDATE assignments
        SET person_id = 'UNKNOWN'
        WHERE person_id NOT IN (SELECT id FROM persons);
    """)

    # 3. Now add the foreign key
    op.create_foreign_key(
        'fk_assignment_person',
        'assignments', 'persons',
        ['person_id'], ['id'],
       
    )

    # 4. Make non-nullable if needed
    op.alter_column('assignments', 'person_id', nullable=False)

Integration with Other Skills

With code-review

Pre-migration review:

[About to generate migration]
[Invoke code-review skill]

→ Reviews model changes for:
  - Breaking changes to existing APIs
  - Missing indexes on new columns
  - Inefficient data types (e.g., Text instead of String(50))
  - Missing constraints (unique, not null)

[code-review approves]
→ Proceed with migration generation

With test-writer

Migration testing workflow:

[Migration created]
[Invoke test-writer skill]

"Generate migration tests:
- Test model after upgrade
- Test model after downgrade
- Test data preservation during upgrade/downgrade
- Test constraints are enforced"

[test-writer creates tests in tests/migrations/]
→ Run tests before committing migration

With pr-reviewer

PR review integration:

[PR with database migration]
[pr-reviewer activated]

→ Detects migration file changes
→ Extra scrutiny applied:
  - Verify model and migration committed together
  - Check upgrade AND downgrade tested
  - Verify no breaking changes
  - Escalate if drops tables/columns

Review comment:
"⚠️ Database migration detected
- [ ] Upgrade tested
- [ ] Downgrade tested
- [ ] Backup plan documented
- [ ] No data loss on rollback"

With safe-schedule-generation

Pre-migration backup:

[About to apply migration to production-like data]
[Invoke safe-schedule-generation skill]

→ Creates mandatory database backup
→ Verifies backup exists and is recent
→ Confirms before proceeding

"Backup verified at 2026-01-01 10:15:23.
Database contains 150 residents, 45 faculty, 12,000 assignments.
Proceed with migration?"

[User confirms]
→ Apply migration
→ Verify success
→ Keep backup for 30 days

With systematic-debugger

Migration failure investigation:

[Migration fails]
[Invoke systematic-debugger skill]

Phase 1: Explore
→ Read migration file
→ Check database state
→ Review error logs

Phase 2: Plan
→ Hypothesis: FK constraint violation due to orphaned data
→ Hypothesis: Data type mismatch
→ Hypothesis: Missing dependency migration

Phase 3: Debug
→ Query database to find orphaned records
→ Test migration in fresh database
→ Isolate root cause

Phase 4: Fix
→ Update migration to handle edge case
→ Test upgrade and downgrade
→ Commit corrected migration

Validation Checklist

Pre-Migration Checklist

• Model changes are complete and tested
• Understand data impact (how many rows affected?)
• Have rollback plan ready
• Database backup exists (if production-like data)
• Know which environments need migration

Migration Generation Checklist

• Generated with meaningful description
• Revision ID is unique
• down_revision points to correct parent
• Both upgrade() and downgrade() present

Migration Review Checklist

• Autogenerated code is correct (not all changes detected!)
• upgrade() matches model changes
• downgrade() reverses upgrade() exactly
• No data loss in downgrade() (or documented)
• Handles existing data correctly (backfills, defaults)
• Foreign keys created with proper constraints
• Indexes added for frequently queried columns

Testing Checklist

• alembic upgrade head succeeds
• Schema matches expected state (inspect with \d)
• alembic downgrade -1 succeeds
• Schema returned to previous state
• alembic upgrade head succeeds again (re-applies cleanly)
• Application starts with new schema
• Full test suite passes
• API tests pass (no breaking changes)

Data Safety Checklist

• Non-nullable columns have defaults or backfills
• Foreign keys validated against existing data
• Unique constraints won't fail on existing data
• Data migrations preserve information
• Downgrade doesn't delete critical data (or export first)

Commit Checklist

• Model file(s) staged
• Migration file(s) staged
• Both committed together
• Commit message describes change
• No orphaned model changes without migrations
• No orphaned migrations without model changes

Escalation Checklist

Escalate to human if ANY of these are true:

• Dropping tables with data
• Dropping columns with data
• Changing primary keys
• Large-scale data migration (>10k rows affected)
• Modifying ACGME compliance tables
• Affects audit trail tables
• Production database migration
• Migration failure with unclear cause
• Multiple conflicting migration heads
• Breaking API changes required

References

• backend/alembic/ - Migration directory
• backend/alembic/env.py - Alembic configuration
• backend/app/db/base.py - Base model configuration
• backend/app/models/ - SQLAlchemy models
• /db-migrate slash command