Blocking destructive operations in autogenerated migrations

Add a process_revision_directives hook to env.py that scans the generated operations for DropColumnOp and DropTableOp and raises unless the revision explicitly acknowledges them — autogenerate emits a drop for every attribute someone removes or renames, and a drop is the one migration no downgrade can undo. This guide belongs to autogenerating and reviewing migration scripts.

Quick Answer

Autogenerate compares the model with the database and proposes removing anything the model no longer has. That is correct behaviour and a data-loss risk, because a renamed attribute looks exactly like a deleted one.

Before — a rename generating a drop and an add:

# The developer renamed Customer.fax to Customer.fax_number.
# $ alembic revision --autogenerate -m "rename fax"
import sqlalchemy as sa
from alembic import op


def upgrade() -> None:
    op.add_column("customers", sa.Column("fax_number", sa.String(length=32), nullable=True))
    op.drop_column("customers", "fax")     # every fax number, deleted

After — a hook that refuses unless the revision says so:

# alembic/env.py (excerpt)
import os

from alembic import context
from alembic.operations import ops

from shop.models import Base

DESTRUCTIVE = (ops.DropColumnOp, ops.DropTableOp)


def process_revision_directives(migration_context, revision, directives) -> None:
    script = directives[0]

    # Nothing to do: do not add an empty revision to the history.
    if script.upgrade_ops.is_empty():
        directives[:] = []
        print("No schema changes detected; no revision generated.")
        return

    found = [op_ for op_ in script.upgrade_ops.as_diffs() if op_[0] in
             ("remove_column", "remove_table")]
    if found and not os.environ.get("ALEMBIC_ALLOW_DESTRUCTIVE"):
        details = "\n  ".join(str(item) for item in found)
        raise RuntimeError(
            "autogenerate produced destructive operations:\n  "
            f"{details}\n"
            "If this is intended, re-run with ALEMBIC_ALLOW_DESTRUCTIVE=1 and put the drop "
            "in a revision of its own."
        )


def do_run_migrations(connection) -> None:
    context.configure(
        connection=connection,
        target_metadata=Base.metadata,
        compare_type=True,
        process_revision_directives=process_revision_directives,
        transaction_per_migration=True,
    )
    with context.begin_transaction():
        context.run_migrations()

The rename above now fails at generation time with the column named in the message, which is the moment the developer can still say "that is not what I meant".

Execution Context & Async Workflow Integration

process_revision_directives is called once per alembic revision invocation, after autogenerate has finished comparing and before any file is written. It receives the MigrationScript directive — the in-memory representation of the revision — and can inspect it, modify it, or remove it entirely by emptying the directives list.

Two views of the same information are available. script.upgrade_ops.ops is the tree of operation objects (ops.DropColumnOp, ops.AddColumnOp, ops.ModifyTableOps containing others), which is precise but needs recursion because table-level operations nest. script.upgrade_ops.as_diffs() is the flat tuple form autogenerate's comparison produced — ("remove_column", None, "customers", Column("fax", ...)) — which is easier to filter and is what the example above uses.

Because the hook runs at generation time, it costs nothing at migration time and cannot break a deploy. It also runs on the developer's machine, in the same process as env.py, so it can read environment variables, print guidance, and raise with a message that names exactly what it objected to. That last point matters: a guard that fails with "destructive operation detected" teaches nothing, while one that prints ("remove_column", None, "customers", Column("fax", String(32))) tells the developer which change caused it.

The is_empty() check earns its place separately. Running autogenerate when nothing has changed produces a revision whose upgrade() contains only pass, and that file is permanent: it has to be reviewed, it lengthens every replay from base, and it makes alembic history harder to read. Clearing the directives means no file is written at all.

Under an async env.py the hook is unchanged. Autogenerate runs inside connection.run_sync(), so the comparison happens on a synchronous facade over asyncpg and the directives arrive as ordinary Python objects — the setup described in setting up Alembic env.py for asyncpg.

One limitation is worth stating: the hook only sees operations autogenerate generated. A hand-written op.drop_column() added to a revision afterwards bypasses it completely, which is why the CI check in the last section complements it rather than duplicating it.

Resolving Warnings, Errors & Common Mistakes

The unmapped-table case is the one that has caused real incidents. If target_metadata does not describe a table — a Django app's tables in a shared database, a third-party extension's tables, a reporting table created by hand — then from autogenerate's point of view that table exists in the database and not in the model, so it proposes dropping it. The guard catches it; the real fix is to tell autogenerate the table is not its business:

def include_object(object_, name, type_, reflected, compare_to) -> bool:
    if type_ == "table" and name.startswith(("django_", "metabase_", "pg_stat_")):
        return False
    return True

That, and the include_name variant for whole schemas, is covered in excluding tables and schemas from Alembic autogenerate.

Renames deserve one more note, because the guard turns them from a silent data loss into a decision. On a small table, the right answer is usually a hand-written op.alter_column(..., new_column_name=...), which preserves the data. On a table serving live traffic, a rename breaks the previous release, and the safe sequence is expand and contract — as in renaming a column without downtime.

Advanced: Rewriting Operations Instead of Refusing Them

A hook can do more than raise. Because it holds the operation tree before anything is written, it can also rewrite what autogenerate proposed — which is useful for enforcing house rules rather than blocking mistakes.

The most valuable rewrite is making index builds concurrent by default on existing tables, so the reviewer does not have to remember. Alembic's Rewriter makes this declarative:

# alembic/env.py (excerpt)
from alembic.autogenerate import rewriter
from alembic.operations import ops

writer = rewriter.Rewriter()


@writer.rewrites(ops.CreateIndexOp)
def _concurrent_index(context, revision, op_):
    # A table created in the same revision has no traffic; everything else does.
    created_here = {
        o.table_name for o in context.opts["_autogen_ops"]  # populated below
    }
    if op_.table_name in created_here:
        return op_
    op_.kw["postgresql_concurrently"] = True
    return op_


def process_revision_directives(migration_context, revision, directives) -> None:
    if not directives:
        return
    script = directives[0]
    if script.upgrade_ops.is_empty():
        directives[:] = []
        return
    migration_context.opts["_autogen_ops"] = list(script.upgrade_ops.ops)
    writer(migration_context, revision, directives)

The generated revision then carries postgresql_concurrently=True, and the reviewer's job becomes checking that it is wrapped in an autocommit block — the requirement explained in creating indexes concurrently in Alembic migrations.

Two other rewrites pay for themselves in review time. Splitting destructive operations into their own revision, rather than refusing them, keeps a mixed change reviewable:

def process_revision_directives(migration_context, revision, directives) -> None:
    script = directives[0]
    destructive = [op_ for op_ in script.upgrade_ops.ops
                   if isinstance(op_, (ops.DropTableOp,))]
    if destructive and len(script.upgrade_ops.ops) > len(destructive):
        raise RuntimeError(
            "this change mixes drops with other operations; generate them separately "
            "so the drop can be reviewed and deployed on its own"
        )

And annotating every generated revision with its provenance — who generated it, against which database — costs one line and answers a question that comes up during every incident:

    script.message = f"{script.message} [autogenerated against {migration_context.dialect.name}]"

The line between rewriting and refusing is worth drawing deliberately. Rewrite mechanical things a reviewer would always want the same way: concurrency flags, naming, imports. Refuse anything where the right answer depends on intent — drops, renames, type changes — because a hook that silently "fixes" those has made a decision nobody recorded.

A CI Check for Hand-Written Drops

The hook cannot see a op.drop_column() someone typed into a revision after generating it, and that is a realistic path: a developer hits the guard, decides the drop is fine, and edits the file. A check over the revision files closes the gap and, unlike the hook, runs on every commit including ones nobody generated.

import ast
import pathlib
import re
import sys

DESTRUCTIVE = {"drop_column", "drop_table", "drop_constraint", "drop_index"}
ACKNOWLEDGED = re.compile(r"^#\s*destructive:\s*\S+", re.M)


def destructive_calls(path: pathlib.Path) -> list[str]:
    tree = ast.parse(path.read_text(), filename=str(path))
    found = []
    for node in ast.walk(tree):
        if (
            isinstance(node, ast.Call)
            and isinstance(node.func, ast.Attribute)
            and node.func.attr in DESTRUCTIVE
            and isinstance(node.func.value, ast.Name)
            and node.func.value.id == "op"
        ):
            found.append(f"{path.name}:{node.lineno} op.{node.func.attr}")
    return found


problems: list[str] = []
for path in sorted(pathlib.Path("alembic/versions").glob("*.py")):
    calls = destructive_calls(path)
    if calls and not ACKNOWLEDGED.search(path.read_text()):
        problems.extend(calls)

if problems:
    print("destructive migration operations without an acknowledgement comment:")
    for problem in problems:
        print(f"  {problem}")
    print('\nAdd a comment such as "# destructive: fax removed in release 41, archived 2026-09-10"')
    sys.exit(1)

Parsing with ast rather than grepping avoids two classes of false positive: the words appearing in a docstring or a comment, and drop_index inside a downgrade() that merely reverses a create. Restricting the match to calls on the op name keeps it to Alembic operations.

Requiring a comment rather than forbidding the call is the point. The comment is where the reviewer learns why — which release stopped using the column, whether the data was archived, and when. That is the information the next person needs when the same column is missed by a report three months later, and it is information no schema diff can carry.

Two more practices make the whole arrangement hold. Keep downgrade() honest: for a drop, it can recreate the column but not its values, so it should say so in a comment rather than implying a reversal it cannot perform. And pair the check with alembic check in the same job, so the build fails both for a migration that should not exist and for a model change whose migration is missing — the CI shape described in running Alembic migrations in CI/CD pipelines.

Frequently Asked Questions

Related

• Autogenerating and Reviewing Migration Scripts — The parent guide: what autogenerate compares and how to review it.
• Excluding tables and schemas from Alembic autogenerate — Stopping unmapped tables from being proposed for deletion.
• Dropping a column safely during a rolling deploy — The release sequence a drop belongs in.
• Running Alembic migrations in CI/CD pipelines — Where these checks run.