py-migrate

name	py-migrate
description	Use when creating or modifying database schema in Python projects using Alembic with SQLAlchemy 2.0 and PostgreSQL

Database Migrations (Alembic + SQLAlchemy)

Overview

Safe, reversible PostgreSQL schema changes using Alembic with SQLAlchemy 2.0 async. Models drive migrations via autogenerate.

Core principle: Models are the source of truth. Autogenerate from models, review every migration, test up/down cycle.

When to Use

• Adding or modifying SQLAlchemy models
• Creating new tables, columns, indexes, constraints
• Data migrations (backfills, transforms)
• Enum type changes

SQLAlchemy 2.0 Models

# src/infrastructure/postgres/models.py
from sqlalchemy import String, text
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from datetime import datetime
from uuid import uuid4

class Base(DeclarativeBase):
    pass

class UserModel(Base):
    __tablename__ = "users"

    id: Mapped[uuid.UUID] = mapped_column(UUID, primary_key=True, default=uuid4)
    email: Mapped[str] = mapped_column(String, unique=True, nullable=False)
    name: Mapped[str] = mapped_column(String, nullable=False)
    phone: Mapped[str | None] = mapped_column(String, nullable=True)
    metadata_: Mapped[dict] = mapped_column("metadata", JSONB, server_default=text("'{}'::jsonb"))
    created_at: Mapped[datetime] = mapped_column(server_default=text("NOW()"))
    updated_at: Mapped[datetime] = mapped_column(server_default=text("NOW()"))

Alembic Workflow

# Create migration from model changes
alembic revision --autogenerate -m "add users table"

# Apply migrations
alembic upgrade head

# Rollback one step
alembic downgrade -1

# Show current revision
alembic current

# Show migration history
alembic history

Async env.py Configuration

# alembic/env.py
from sqlalchemy.ext.asyncio import async_engine_from_config
import asyncio

def run_migrations_online() -> None:
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
    )
    async def do_run():
        async with connectable.connect() as connection:
            await connection.run_sync(do_run_migrations)
        await connectable.dispose()

    asyncio.run(do_run())

Makefile Targets

migrate:
	alembic upgrade head

migrate-down:
	alembic downgrade -1

migrate-create:
	alembic revision --autogenerate -m "$(msg)"

migrate-history:
	alembic history

Safe DDL Patterns

Same zero-downtime rules as Go migrations:

Operation	Safe Approach
Add column	nullable=True first → backfill → add constraint
Drop column	Remove from model → deploy → migration to drop
Rename column	Add new → copy → update code → drop old
Add index	op.create_index(..., if_not_exists=True)

Enum Handling

# In migration
import sqlalchemy as sa

user_role = sa.Enum('admin', 'member', 'viewer', name='user_role')

def upgrade() -> None:
    user_role.create(op.get_bind(), checkfirst=True)
    op.add_column('users', sa.Column('role', user_role, nullable=False, server_default='member'))

def downgrade() -> None:
    op.drop_column('users', 'role')
    user_role.drop(op.get_bind(), checkfirst=True)

Data Migrations

def upgrade() -> None:
    # Schema change
    op.add_column('users', sa.Column('full_name', sa.String))
    # Data migration
    op.execute("UPDATE users SET full_name = first_name || ' ' || last_name")
    # Then enforce constraint
    op.alter_column('users', 'full_name', nullable=False)

Verification

alembic upgrade head && alembic downgrade -1 && alembic upgrade head

Common Mistakes

• Not reviewing autogenerated migrations — always read the SQL
• Missing Mapped[] type annotations — breaks autogenerate detection
• Using nullable=False without server_default on existing tables
• Forgetting async engine config in env.py
• Enum types not dropped in downgrade

Chains

• REQUIRED: Update CLAUDE.md if new migration commands are added (claude-md)