Compile Alembic Revisions Into a Migration Ledger; Sign Data-Loss Ops by Hand

Schema-change documentation fails when generated prose describes rollback steps that nobody timed, tested, or agreed to own. The practical split is a compiled Alembic ledger plus a human-signed sheet for data loss, locks, and downgrade policy. A model may draft intent paragraphs from that ledger; it must not fill duration, blast radius, or online-safety claims. The rest of this tutorial is a parser, a JSON contract, a signature checklist, and a gate that fails unsigned destructive operations.

The failure mode in migration docs

Many teams still paste upgrade() bodies into a chat window and request a friendly changelog paragraph for the release notes. The paragraph often sounds complete, yet it still hides DROP TABLE calls, raw op.execute usage, and missing downgrade paths. Readers then treat marketing language as a runbook, which is how production freeze windows get skipped. Generated text is not itself the defect; treating unsigned impact language as an operational runbook is the defect.

Alembic revisions are already structured enough to compile without executing them against a live database. Each file typically exposes a revision string, a down_revision pointer, and functions named upgrade and downgrade. Operation calls such as op.add_column or op.drop_constraint are ordinary Python attributes visible to an ast walk. Importing the module is unsafe in CI when a revision still contains leftover op.execute fragments or local side-effect imports.

Ownership matrix

Document cells should be typed by origin so a compiler cannot overwrite a signature, and a model cannot mint a lock window. The table below is a working contract for the renderer, not a claim about any vendor's database engine.

Cell	Source of truth	Who may draft	Who must sign
revision, down_revision	Module-level string assigns	Compiler	Nobody; fail the build on parse error
op.* names and string args	ast of upgrade() / downgrade()	Compiler	Nobody
Intent paragraph	One ledger row, no SQL bodies	Model, optional	Reviewer confirms tables match
Irreversible / data-loss flag	Heuristic plus owner	Heuristic proposes	Migration owner
Lock or rewrite window	Staging restore, EXPLAIN, load	Human	Migration owner
Prod downgrade allowed	Release policy	Human	Migration owner
Backfill identity and batch size	Runbook	Human	Migration owner
Connection strings, passwords	Never present	Nobody	Reject the draft

Keep connection strings and database passwords out of both the compiled ledger and the optional prose prompt. The compiler should read revision files from disk; it should not connect to Postgres to enrich the story. Cardinality and bloat are staging measurements, and they belong on the signed sheet after a restore test.

Step 1: Compile the ledger from revision files

Place the compiler next to alembic/versions and emit one JSON object per revision file on stdout. The example below is a proposal, not a production scanner, and it does not execute migrations or import them.

#!/usr/bin/env python3
"""alembic_ledger.py — compile op calls; do not import revision modules."""
from __future__ import annotations

import ast
import json
import sys
from pathlib import Path

DESTRUCTIVE = {
    "drop_table",
    "drop_column",
    "drop_index",
    "drop_constraint",
    "rename_table",
    "execute",
    "bulk_insert",
}

class OpVisitor(ast.NodeVisitor):
    def __init__(self) -> None:
        self.ops: list[dict] = []

    def visit_Call(self, node: ast.Call) -> None:
        name = None
        if isinstance(node.func, ast.Attribute) and isinstance(
            node.func.value, ast.Name
        ):
            if node.func.value.id == "op":
                name = node.func.attr
        if name:
            tables: list[str] = []
            for arg in node.args:
                if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
                    tables.append(arg.value)
            self.ops.append({"op": name, "string_args": tables[:4]})
        self.generic_visit(node)


def module_str(tree: ast.AST, ident: str) -> str | None:
    for node in tree.body:
        if isinstance(node, ast.Assign):
            for target in node.targets:
                if isinstance(target, ast.Name) and target.id == ident:
                    val = node.value
                    if isinstance(val, ast.Constant) and isinstance(val.value, str):
                        return val.value
                    if isinstance(val, ast.Constant) and val.value is None:
                        return None
    return None


def function_ops(tree: ast.AST, fname: str) -> list[dict]:
    for node in tree.body:
        if isinstance(node, ast.FunctionDef) and node.name == fname:
            visitor = OpVisitor()
            visitor.visit(node)
            return visitor.ops
    return []


def compile_revision(path: Path) -> dict:
    tree = ast.parse(path.read_text(encoding="utf-8"))
    upgrade_ops = function_ops(tree, "upgrade")
    downgrade_ops = function_ops(tree, "downgrade")
    return {
        "file": path.name,
        "revision": module_str(tree, "revision"),
        "down_revision": module_str(tree, "down_revision"),
        "upgrade_ops": upgrade_ops,
        "downgrade_ops": downgrade_ops,
        "destructive_ops": [o for o in upgrade_ops if o["op"] in DESTRUCTIVE],
        "has_downgrade": bool(downgrade_ops),
    }


def main() -> None:
    root = Path(sys.argv[1] if len(sys.argv) > 1 else "alembic/versions")
    rows = [
        compile_revision(path)
        for path in sorted(root.glob("*.py"))
        if path.name != "__init__.py"
    ]
    json.dump({"migrations": rows}, sys.stdout, indent=2)
    print()


if __name__ == "__main__":
    main()

Invoke the compiler as a deterministic command so the ledger is a build artifact rather than a chat transcript.

python3 alembic_ledger.py alembic/versions > migration_ledger.json

A compact ledger fragment looks like the following, which is synthetic sample data rather than output from a private repository.

{
  "migrations": [
    {
      "file": "20260918_drop_legacy_email.py",
      "revision": "a1b2c3d4e5f6",
      "down_revision": "908070605040",
      "upgrade_ops": [
        {"op": "drop_index", "string_args": ["ix_users_email_legacy"]},
        {"op": "drop_column", "string_args": ["users", "email_legacy"]}
      ],
      "downgrade_ops": [],
      "destructive_ops": [
        {"op": "drop_index", "string_args": ["ix_users_email_legacy"]},
        {"op": "drop_column", "string_args": ["users", "email_legacy"]}
      ],
      "has_downgrade": false
    }
  ]
}

Step 2: Render a skeleton sheet with empty signed cells

The renderer must copy only compiler fields into Markdown and leave owner cells as the sentinel UNSIGNED. Review tools then show a diff on signatures instead of a wall of regenerated adjectives around the same DROP TABLE. The function below is unexecuted example code for that skeleton.

def render_sheet(ledger: dict) -> str:
    parts = ["# Migration impact sheet", ""]
    for row in ledger["migrations"]:
        dest = ", ".join(sorted({o["op"] for o in row["destructive_ops"]})) or "none"
        ops = ", ".join(o["op"] for o in row["upgrade_ops"]) or "(empty)"
        parts += [
            f"## {row['revision']}",
            f"- file: `{row['file']}`",
            f"- down_revision: `{row['down_revision']}`",
            f"- upgrade ops: {ops}",
            f"- destructive ops (compiled): {dest}",
            f"- downgrade present: {row['has_downgrade']}",
            "- intent (model draft, reviewer confirms tables): UNSIGNED",
            "- data_loss (owner): UNSIGNED",
            "- lock_or_rewrite_window (owner): UNSIGNED",
            "- prod_downgrade_allowed (owner): UNSIGNED",
            "- backfill_batch_policy (owner): UNSIGNED",
            "",
        ]
    return "\n".join(parts)

Step 3: Optional prose lane from the ledger, not from SQL

If you want a readable intent sentence, feed the model one JSON row: revision, op names, and string arguments. Do not paste op.execute bodies, because those strings sometimes contain hostnames or redacted fragments that still leak environment shape. Do not ask the model whether the migration is safe to run online; that question belongs on the signed sheet.

Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode's free model access and free server option can host this split: the compiler runs on the free server against a checkout, and a free model drafts intent lines from ledger JSON. Neither lane should receive credentials, and signed cells stay empty until a human fills them.

A prompt that stays inside the prose lane looks like the following proposal.

Using only this JSON object, write 2-3 sentences that describe
the schema operations. Do not estimate runtime. Do not claim
the change is reversible. Do not invent indexes that are absent.
JSON:
{...one migration row...}

If the model adds a lock estimate or a run-in-production-freely line, discard the draft and keep UNSIGNED. Table names in the paragraph still need a reviewer check against string_args, because truncated JSON is a common way models drop a table.

Step 4: Fail CI when destructive ops lack signatures

A documentation pipeline that never fails CI is indistinguishable from a blog post about process. The check below reads the rendered sheet and the ledger together and exits nonzero on UNSIGNED destructive rows. Label this gate as something you must wire into your own workflow file; it does not prove the signed text is correct. It only proves the owner cells are no longer the sentinel, which is a necessary and incomplete control.

OWNER_FIELDS = (
    "data_loss",
    "lock_or_rewrite_window",
    "prod_downgrade_allowed",
)


def signed_value(block: str, field: str) -> str | None:
    prefix = f"- {field}: "
    for line in block.splitlines():
        if line.startswith(prefix):
            return line[len(prefix):].strip()
    return None


def assert_destructive_signed(ledger: dict, sheet: str) -> None:
    blocks: dict[str, str] = {}
    for block in sheet.split("## ")[1:]:
        title, _, rest = block.partition("\n")
        blocks[title.strip()] = rest
    for row in ledger["migrations"]:
        if not row["destructive_ops"]:
            continue
        body = blocks.get(row["revision"] or "", "")
        for field in OWNER_FIELDS:
            value = signed_value(body, field)
            if not value or value == "UNSIGNED":
                raise SystemExit(f"{row['revision']}: {field} is unsigned")

A minimal command sequence for local verification, still unlabeled as a full CI product, is the following.

python3 alembic_ledger.py alembic/versions > migration_ledger.json
python3 render_migration_sheet.py migration_ledger.json > MIGRATION_IMPACT.md
python3 check_migration_signatures.py migration_ledger.json MIGRATION_IMPACT.md

Step 5: Decision rules for the human owner

1. If drop_table or drop_column appears, data_loss must be yes or a named exception with a backup locator, not a vibe.
2. If execute appears, the owner pastes a reviewed SQL digest into the signed sheet, never into the model prompt.
3. If downgrade_ops is empty, prod_downgrade_allowed is no, and the release notes must say so in operator language.
4. If an index is created on an existing large table, lock_or_rewrite_window needs a measured staging number, not a generated guess.
5. If bulk_insert appears, backfill_batch_policy needs batch size, pause, and a stop condition written by the owner.

These rules are policy, not model weights. Changing them in a prompt does not change production risk, and the compiler should not pretend otherwise. Expand/contract still sits outside this pipeline: dual writes, leftover columns, and feature-gated reads remain design work.

Limitations

The ast compiler does not evaluate op.f(), helper functions, or loops that build operations at runtime. Revisions that wrap DDL in custom Python will under-report destructive work, which is a false-negative class you should treat as unsigned by default. The compiler never talks to the database, so it cannot see table cardinality, bloat, or replica lag. Lock windows therefore cannot come from this pipeline, and neither can claims about online-safe rewrites.

Alembic branches, merge revisions, and multiple bases need extra graph logic that this example does not include. Intent drafts can still mis-order operations or drop a table name when the JSON is truncated. That is why the reviewer confirms tables against the ledger rather than against the paragraph. This workflow also does not replace a backup verification drill or a rollback rehearsal on restored data.

Who should not use this approach

Do not use this split if your migrations exist only inside a hosted console you cannot parse as files. Do not use it as permission to let a model approve production DDL, because the prose lane has no authority over catalogs. Teams without a named migration owner will fill UNSIGNED with plausible sentences and ship anyway; the gate only helps when someone is accountable. If you need exact lock timing, measure on a restored snapshot rather than asking a language model for milliseconds.

Skip the prose lane entirely when the ledger already lists three or fewer operations. A sentence that restates add_column users email adds noise, not safety, and it trains reviewers to skim signatures.

Compiled ledgers make schema history cheap to refresh after every revision lands on the default branch. Signed impact cells are the part that still costs a person, because data loss, lock time, and downgrade policy are not properties of English. Keep the model on intent, keep the compiler on op names, and keep production authority on a named owner who can refuse the release.