Migrations, Seed Data, and Database Changes in Git

CREATE TABLE IF NOT EXISTS schema_migration (    revision_id TEXT PRIMARY KEY,    parent_revision TEXT,    checksum_sha256 TEXT NOT NULL,    description TEXT NOT NULL,    applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,    applied_by TEXT NOT NULL,    duration_ms INTEGER NOT NULL CHECK(duration_ms >= 0)) STRICT;SELECT revision_id, parent_revision, checksum_sha256, applied_atFROM schema_migrationORDER BY applied_at, revision_id;

ALTER TABLE customer ADD COLUMN normalized_email TEXT;CREATE INDEX idx_customer_normalized_email    ON customer(normalized_email);-- Backfill in bounded batches in a separate operational job.UPDATE customerSET normalized_email = lower(trim(email))WHERE normalized_email IS NULL;-- Contract only after every deployed version uses normalized_email.-- ALTER TABLE customer ALTER COLUMN normalized_email SET NOT NULL;-- ALTER TABLE customer DROP COLUMN legacy_email;

from alembic import opimport sqlalchemy as sarevision = "20260805_01"down_revision = "20260730_03"branch_labels = Nonedepends_on = Nonedef upgrade() -> None:    op.add_column(        "customer",        sa.Column("normalized_email", sa.String(320), nullable=True),    )    op.create_index(        "idx_customer_normalized_email",        "customer",        ["normalized_email"],        unique=False,    )def downgrade() -> None:    op.drop_index("idx_customer_normalized_email", table_name="customer")    op.drop_column("customer", "normalized_email")

INSERT INTO order_status(status_code, display_name, terminal)VALUES  ('draft','Draft',0),  ('submitted','Submitted',0),  ('paid','Paid',1),  ('cancelled','Cancelled',1)ON CONFLICT(status_code) DO UPDATE SET  display_name = excluded.display_name,  terminal = excluded.terminal;

db/  migrations/    20260805_01_add_normalized_email.py    20260807_01_backfill_normalized_email.py    20260812_01_enforce_normalized_email.py  seeds/    reference_order_status.sql  checks/    verify_normalized_email.sql  README.md  OWNERS

set -euo pipefail# 1. Build an empty database from the complete revision history.alembic upgrade headalembic current --check-headspytest tests/database# 2. Upgrade a copy representing the previous production release.restore_fixture previous_release.dbalembic upgrade headpython db/checks/verify_schema.pypytest tests/integration# 3. Render SQL for review where supported.alembic upgrade base:head --sql > migration-plan.sqlgit diff --exit-code -- migration-plan.sql || true

import hashlibimport sqlite3from pathlib import PathMIGRATIONS = [    ("001", "create product", "CREATE TABLE product(id INTEGER PRIMARY KEY, sku TEXT UNIQUE)"),    ("002", "add active", "ALTER TABLE product ADD COLUMN active INTEGER NOT NULL DEFAULT 1"),]with sqlite3.connect("academy.db") as db:    db.execute("""CREATE TABLE IF NOT EXISTS schema_migration(        revision_id TEXT PRIMARY KEY,        checksum_sha256 TEXT NOT NULL,        description TEXT NOT NULL,        applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP    )""")    applied = {row[0]: row[1] for row in db.execute(        "SELECT revision_id, checksum_sha256 FROM schema_migration"    )}    for revision, description, sql in MIGRATIONS:        checksum = hashlib.sha256(sql.encode()).hexdigest()        if revision in applied:            if applied[revision] != checksum:                raise RuntimeError(f"edited applied migration: {revision}")            continue        db.execute(sql)        db.execute(            "INSERT INTO schema_migration(revision_id,checksum_sha256,description) VALUES(?,?,?)",            (revision, checksum, description),        )