Alembic Cheatsheet 04 — op API Reference

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

op.create_table(
    "users",
    sa.Column("id", sa.Integer, primary_key=True),
    sa.Column("email", sa.String(255), nullable=False),
    sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now()),
    sa.UniqueConstraint("email"),
)

op.drop_table("users")
op.rename_table("old", "new")

op.add_column("users", sa.Column("status", sa.String(20), nullable=False, server_default="active"))
op.drop_column("users", "old_field")

op.alter_column("users", "name",
    type_=sa.String(200),
    nullable=False,
    new_column_name="full_name",
    server_default=sa.text("'unknown'"),
    existing_type=sa.String(120),
    existing_nullable=True,
)

op.create_index("ix_users_email", "users", ["email"], unique=True)
op.drop_index("ix_users_email", "users")
op.create_index("ix_users_active", "users", ["email"], postgresql_where=sa.text("active = true"))
op.create_index("ix_users_lower_email", "users", [sa.text("lower(email)")])

# Postgres GIN
op.create_index("ix_events_gin", "events", ["payload"],
    postgresql_using="gin",
    postgresql_ops={"payload": "jsonb_path_ops"},
)

# Concurrent (can't be in transaction)
op.execute("CREATE INDEX CONCURRENTLY ix_users_email ON users (email)")

op.create_unique_constraint("uq_users_email", "users", ["email"])
op.create_check_constraint("ck_users_age", "users", "age > 0")
op.drop_constraint("uq_users_email", "users", type_="unique")
op.drop_constraint("ck_users_age", "users", type_="check")

op.create_foreign_key(
    "fk_posts_user", "posts", "users",
    ["user_id"], ["id"],
    ondelete="CASCADE",
    onupdate="NO ACTION",
    deferrable=False,
    initially="IMMEDIATE",
)

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

op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
op.execute("UPDATE users SET status = 'active' WHERE status IS NULL")
op.execute(sa.text("DELETE FROM sessions WHERE expires_at < now()"))

bind = op.get_bind()

# Lightweight tables (don't import ORM models)
users = sa.table("users",
    sa.column("id", sa.Integer),
    sa.column("status", sa.String),
)

bind.execute(users.update().values(status="active"))

op.execute("CREATE TYPE user_role AS ENUM ('user', 'admin')")
op.add_column("users", sa.Column("role", postgresql.ENUM("user", "admin", name="user_role")))

# Add value
op.execute("ALTER TYPE user_role ADD VALUE 'editor'")

# Drop
op.execute("DROP TYPE user_role")

op.execute("CREATE SCHEMA IF NOT EXISTS audit")
op.execute("DROP SCHEMA audit CASCADE")
op.create_table("logs", ..., schema="audit")

with op.batch_alter_table("users") as batch:
    batch.add_column(sa.Column("status", sa.String(20)))
    batch.alter_column("name", new_column_name="full_name")

op.add_column("users",
    sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()")),
)
op.alter_column("users", "created_at",
    server_default=sa.text("now()"),
    existing_type=sa.DateTime(timezone=True),
)

op.create_table("users",
    sa.Column(...),
    comment="App users",
)
op.alter_column("users", "email", comment="User email address")

op.execute("CREATE EXTENSION IF NOT EXISTS pgcrypto")

# Range column
op.add_column("bookings",
    sa.Column("period", postgresql.TSTZRANGE),
)

# Exclusion constraint
op.execute("""
    ALTER TABLE bookings ADD CONSTRAINT no_overlap
    EXCLUDE USING gist (room_id WITH =, period WITH &&)
""")

def upgrade():
    bind = op.get_bind()
    inspector = sa.inspect(bind)
    
    cols = [c["name"] for c in inspector.get_columns("users")]
    if "old_field" in cols:
        op.drop_column("users", "old_field")