Alembic Migrations Checkpoint
25 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)Fix the mistakes
The slips that cost marks in this module. Spot each one, then check the fix.
A required column is added to a customer table that already has people in it. The change is split into four small migrations. This checkpoint runs them, moves back and forth, and tests the whole module.
Split the change into small migrations
Each migration does one thing that cannot fail halfway for a reason you cannot predict. Add the column, backfill it, then tighten it. A problem then points to one small file.
revisions = ["0001 create customer", "0002 add status (nullable)", "0003 backfill status", "0004 status NOT NULL"]Each revision has its own downgrade(). The backfill has nothing to undo, so its downgrade does nothing.
Know where the database is
The version table holds one revision. After upgrade head it is the newest, and after downgrade -1 it is the one before. Always read it before and after a change.
command.upgrade(cfg, "head")print(MigrationContext.configure(conn).get_current_revision())command.downgrade(cfg, "-1")Example 1 · Down and up again
import os, tempfileimport sqlalchemy as safrom alembic import commandfrom alembic.config import Configfrom alembic.migration import MigrationContextfrom alembic.script import ScriptDirectory
ENV = ( "from alembic import context\n" "context.configure(connection=context.config.attributes['connection'])\n" "with context.begin_transaction():\n" " context.run_migrations()\n")
def project(revisions): root = tempfile.mkdtemp() os.makedirs(os.path.join(root, "versions")) open(os.path.join(root, "env.py"), "w").write(ENV) for rev, (down, up, dn) in revisions.items(): lines = [f'revision = "{rev}"', f"down_revision = {down!r}", "from alembic import op", "import sqlalchemy as sa", "", "def upgrade():"] lines += [" " + s for s in up] + ["", "def downgrade():"] + [" " + s for s in dn] open(os.path.join(root, "versions", rev + ".py"), "w").write("\n".join(lines) + "\n") cfg = Config() cfg.set_main_option("script_location", root) return cfg
engine = sa.create_engine("sqlite://")conn = engine.connect()
def state(): names = sa.inspect(conn).get_table_names() columns = [c["name"] for c in sa.inspect(conn).get_columns("product")] if "product" in names else [] return MigrationContext.configure(conn).get_current_revision(), columns
def run(cfg, action, target): cfg.attributes["connection"] = conn action(cfg, target) conn.commit() return state()
REVS = { "0001": (None, ['op.create_table("customer", sa.Column("id", sa.Integer, primary_key=True), sa.Column("full_name", sa.String(60), nullable=False))'], ['op.drop_table("customer")']), "0002": ("0001", ['op.add_column("customer", sa.Column("status", sa.String(10), nullable=True))'], ['with op.batch_alter_table("customer") as batch:', ' batch.drop_column("status")']), "0003": ("0002", ['op.execute("update customer set status = \'active\' where status is null")'], ['pass']), "0004": ("0003", ['with op.batch_alter_table("customer") as batch:', ' batch.alter_column("status", existing_type=sa.String(10), nullable=False)'], ['with op.batch_alter_table("customer") as batch:', ' batch.alter_column("status", existing_type=sa.String(10), nullable=True)']),}
def status_info(): cols = {c["name"]: c["nullable"] for c in sa.inspect(conn).get_columns("customer")} nulls = conn.execute(sa.text("select count(*) from customer where status is null")).scalar_one() if "status" in cols else None return MigrationContext.configure(conn).get_current_revision(), cols.get("status"), nulls
cfg = project(REVS)run(cfg, command.upgrade, "0001")conn.execute(sa.text("insert into customer (full_name) values ('Asha')"))conn.commit()print(run(cfg, command.upgrade, "head")[0])print(run(cfg, command.downgrade, "0001")[0])print(conn.execute(sa.text("select * from customer")).all())print(run(cfg, command.upgrade, "head")[0])print(conn.execute(sa.text("select * from customer")).all())Output
0004 0001 [(1, 'Asha')] 0004 [(1, 'Asha', 'active')]
What this code does
- Going down to 0001 dropped the
statuscolumn, so its values are gone. - Going up again re-runs the backfill, so the customer gets
activeagain. The backfill makes this path safe.
upgrade and downgrade change the structure. The data stays unless a step drops a column.
Choose the right tool for the change
Same value for all rows: a server default. Value computed from other data: add nullable, backfill, tighten. A rename or move: expand and contract over several releases.
op.add_column("customer", sa.Column("tier", sa.String(10), nullable=True))op.execute("update customer set tier = 'basic'")Example 2 · Four small migrations on live data
import os, tempfileimport sqlalchemy as safrom alembic import commandfrom alembic.config import Configfrom alembic.migration import MigrationContextfrom alembic.script import ScriptDirectory
ENV = ( "from alembic import context\n" "context.configure(connection=context.config.attributes['connection'])\n" "with context.begin_transaction():\n" " context.run_migrations()\n")
def project(revisions): root = tempfile.mkdtemp() os.makedirs(os.path.join(root, "versions")) open(os.path.join(root, "env.py"), "w").write(ENV) for rev, (down, up, dn) in revisions.items(): lines = [f'revision = "{rev}"', f"down_revision = {down!r}", "from alembic import op", "import sqlalchemy as sa", "", "def upgrade():"] lines += [" " + s for s in up] + ["", "def downgrade():"] + [" " + s for s in dn] open(os.path.join(root, "versions", rev + ".py"), "w").write("\n".join(lines) + "\n") cfg = Config() cfg.set_main_option("script_location", root) return cfg
engine = sa.create_engine("sqlite://")conn = engine.connect()
def state(): names = sa.inspect(conn).get_table_names() columns = [c["name"] for c in sa.inspect(conn).get_columns("product")] if "product" in names else [] return MigrationContext.configure(conn).get_current_revision(), columns
def run(cfg, action, target): cfg.attributes["connection"] = conn action(cfg, target) conn.commit() return state()
REVS = { "0001": (None, ['op.create_table("customer", sa.Column("id", sa.Integer, primary_key=True), sa.Column("full_name", sa.String(60), nullable=False))'], ['op.drop_table("customer")']), "0002": ("0001", ['op.add_column("customer", sa.Column("status", sa.String(10), nullable=True))'], ['with op.batch_alter_table("customer") as batch:', ' batch.drop_column("status")']), "0003": ("0002", ['op.execute("update customer set status = \'active\' where status is null")'], ['pass']), "0004": ("0003", ['with op.batch_alter_table("customer") as batch:', ' batch.alter_column("status", existing_type=sa.String(10), nullable=False)'], ['with op.batch_alter_table("customer") as batch:', ' batch.alter_column("status", existing_type=sa.String(10), nullable=True)']),}
def status_info(): cols = {c["name"]: c["nullable"] for c in sa.inspect(conn).get_columns("customer")} nulls = conn.execute(sa.text("select count(*) from customer where status is null")).scalar_one() if "status" in cols else None return MigrationContext.configure(conn).get_current_revision(), cols.get("status"), nulls
cfg = project(REVS)run(cfg, command.upgrade, "0001")conn.execute(sa.text("insert into customer (full_name) values ('Asha'), ('Ravi'), ('Meena')"))conn.commit()for step in ["0002", "0003", "0004"]: run(cfg, command.upgrade, step) print(step, status_info()[1:])print(conn.execute(sa.text("select full_name, status from customer order by id")).all())Output
0002 (True, 3)
0003 (True, 0)
0004 (False, 0)
[('Asha', 'active'), ('Ravi', 'active'), ('Meena', 'active')]What this code does
- After 0002 the column exists and is nullable, and all three rows are empty.
- After 0003 nothing is empty. After 0004 the column is NOT NULL, so
nullableisFalse.
Review each file before you run it, and never edit a migration that has already been applied elsewhere.
Finish up
Mark this lesson complete
This is the last lesson in this section.