Alembic Cheatsheet 10 — Common Migration Recipes

def upgrade():
    op.add_column("users",
        sa.Column("status", sa.String(20), nullable=False, server_default="active"))

def downgrade():
    op.drop_column("users", "status")

def upgrade():
    op.add_column("users", sa.Column("status", sa.String(20), nullable=True))
    op.execute("UPDATE users SET status = 'active'")
    op.alter_column("users", "status", nullable=False)

def downgrade():
    op.drop_column("users", "status")

def upgrade():
    op.alter_column("users", "name", new_column_name="full_name", existing_type=sa.String(120))

def downgrade():
    op.alter_column("users", "full_name", new_column_name="name", existing_type=sa.String(120))

def upgrade():
    op.create_foreign_key(
        "fk_posts_user", "posts", "users",
        ["user_id"], ["id"],
        ondelete="CASCADE",
    )

def downgrade():
    op.drop_constraint("fk_posts_user", "posts", type_="foreignkey")

def upgrade():
    op.create_foreign_key(
        "fk_posts_user", "posts", "users",
        ["user_id"], ["id"],
        postgresql_not_valid=True,
    )
    op.execute("ALTER TABLE posts VALIDATE CONSTRAINT fk_posts_user")

def upgrade():
    op.execute("CREATE INDEX CONCURRENTLY ix_users_email ON users (email)")

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY ix_users_email")

transaction_per_migration = false

def upgrade():
    op.execute("""
        CREATE UNIQUE INDEX CONCURRENTLY ix_users_email_active
        ON users (email) WHERE deleted_at IS NULL
    """)

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY ix_users_email_active")

def upgrade():
    op.create_check_constraint("ck_orders_amount_positive", "orders", "amount > 0")

def downgrade():
    op.drop_constraint("ck_orders_amount_positive", "orders", type_="check")

def upgrade():
    op.add_column("events", sa.Column("payload", postgresql.JSONB(), nullable=False, server_default="{}"))
    op.execute("""
        CREATE INDEX CONCURRENTLY ix_events_payload_gin ON events
        USING gin (payload jsonb_path_ops)
    """)

def downgrade():
    op.execute("DROP INDEX CONCURRENTLY ix_events_payload_gin")
    op.drop_column("events", "payload")

def upgrade():
    op.execute("CREATE TYPE user_role AS ENUM ('user', 'admin')")
    op.add_column("users",
        sa.Column("role", postgresql.ENUM("user", "admin", name="user_role", create_type=False),
                  nullable=False, server_default="user"))

def downgrade():
    op.drop_column("users", "role")
    op.execute("DROP TYPE user_role")

def upgrade():
    op.execute("ALTER TYPE user_role ADD VALUE 'editor'")

def downgrade():
    # No-op (can't remove enum values)
    pass

def upgrade():
    op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")

def downgrade():
    op.execute("DROP EXTENSION IF EXISTS pg_trgm")

def upgrade():
    op.execute("""
        CREATE OR REPLACE FUNCTION update_modified_column()
        RETURNS TRIGGER AS $$
        BEGIN
            NEW.updated_at = now();
            RETURN NEW;
        END;
        $$ LANGUAGE plpgsql;
    """)
    op.execute("""
        CREATE TRIGGER set_updated_at
        BEFORE UPDATE ON users
        FOR EACH ROW
        EXECUTE FUNCTION update_modified_column();
    """)

def downgrade():
    op.execute("DROP TRIGGER IF EXISTS set_updated_at ON users")
    op.execute("DROP FUNCTION IF EXISTS update_modified_column")

def upgrade():
    op.execute("""
        CREATE VIEW active_users AS
        SELECT * FROM users WHERE deleted_at IS NULL
    """)

def downgrade():
    op.execute("DROP VIEW IF EXISTS active_users")

def upgrade():
    op.execute("""
        CREATE MATERIALIZED VIEW user_stats AS
        SELECT user_id, count(*) AS post_count FROM posts GROUP BY user_id
    """)
    op.execute("CREATE UNIQUE INDEX ON user_stats (user_id)")

def downgrade():
    op.execute("DROP MATERIALIZED VIEW IF EXISTS user_stats")

REFRESH MATERIALIZED VIEW CONCURRENTLY user_stats;

def upgrade():
    op.execute("""
        CREATE TABLE events (
            id bigserial,
            occurred_at timestamptz NOT NULL,
            payload jsonb NOT NULL,
            PRIMARY KEY (id, occurred_at)
        ) PARTITION BY RANGE (occurred_at)
    """)
    op.execute("""
        CREATE TABLE events_2026_01 PARTITION OF events
        FOR VALUES FROM ('2026-01-01') TO ('2026-02-01')
    """)

def downgrade():
    op.execute("DROP TABLE events")

def upgrade():
    op.drop_column("users", "old_field")

def downgrade():
    op.add_column("users", sa.Column("old_field", sa.String, nullable=True))

def upgrade():
    bind = op.get_bind()
    # Don't import ORM models — they evolve
    users = sa.table("users",
        sa.column("id", sa.Integer),
        sa.column("status", sa.String),
    )
    bind.execute(users.update().where(users.c.status == None).values(status="active"))