Database Schema Diff Tool

Database Schema Diff Tool

Problem Statement

Keeping database schemas in sync across environments (production, staging, development) is one of the most error-prone parts of modern application deployment. When you promote code from staging to production, you need to know exactly what schema changes are pending — new columns, altered types, missing indexes, changed constraints. Manually comparing pg_dump --schema-only outputs across environments is tedious and easy to get wrong, especially when you only want the delta, not the full schema.

This script connects to two PostgreSQL databases simultaneously, introspects their schemas via SQLAlchemy's inspector, computes a structured diff of tables, columns, indexes, and constraints, and emits a ready-to-review SQL migration file. It's designed to be run in CI/CD pipelines or by a DBA before a release, giving a deterministic, reviewable artifact instead of guesswork.

How It Works

The tool uses SQLAlchemy's Inspector API to extract a normalized representation of each database's schema into plain Python dictionaries keyed by table name. It then performs a symmetric set-difference across tables, columns, indexes, and constraints, classifying every difference as either an ADD (present in db2, missing from db1) or a DROP (present in db1, missing from db2). Because we assume db1 is the "source of truth" (e.g., prod) and db2 is the "target" (e.g., staging), additions are emitted as CREATE statements and drops as DROP statements.

The script follows a strategy pattern for SQL generation (each object type has its own renderer) and uses dataclasses for the diff result so it can be serialized to JSON if needed. Type normalization is done via SQLAlchemy's dialect-aware compile() so that VARCHAR(255) from one server matches character varying(255) from another. All connections are context-managed, and the script fails fast with a clear error if either database is unreachable.

Full Script

#!/usr/bin/env python3
"""
Database Schema Diff Tool

Compares the schema of two PostgreSQL databases and emits a SQL migration
script that transforms db1 into db2 (or vice versa, depending on direction).

Usage:
    python3 schema_diff.py --db1 postgresql://prod/app \\
                           --db2 postgresql://staging/app \\
                           --output diff.sql
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
from dataclasses import dataclass, field, asdict
from typing import Any

from sqlalchemy import create_engine, inspect, text
from sqlalchemy.engine import Engine, Inspector
from sqlalchemy.exc import SQLAlchemyError


# ---------------------------------------------------------------------------
# Configuration (overridable via environment variables)
# ---------------------------------------------------------------------------
DEFAULT_SCHEMA = os.environ.get("SCHEMA_DIFF_SCHEMA", "public")
LOG_LEVEL = os.environ.get("SCHEMA_DIFF_LOG_LEVEL", "INFO").upper()
CONNECT_TIMEOUT = int(os.environ.get("SCHEMA_DIFF_TIMEOUT", "10"))
INCLUDE_INDEXES = os.environ.get("SCHEMA_DIFF_INDEXES", "true").lower() == "true"

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
log = logging.getLogger("schema_diff")


# ---------------------------------------------------------------------------
# Data structures
# ---------------------------------------------------------------------------
@dataclass
class ColumnDiff:
    table: str
    name: str
    action: str  # "ADD" or "DROP"
    type_sql: str | None = None
    nullable: bool | None = None
    default: str | None = None


@dataclass
class TableDiff:
    name: str
    action: str  # "ADD" or "DROP"


@dataclass
class IndexDiff:
    table: str
    name: str
    action: str
    columns: list[str] = field(default_factory=list)
    unique: bool = False


@dataclass
class SchemaDiff:
    tables: list[TableDiff] = field(default_factory=list)
    columns: list[ColumnDiff] = field(default_factory=list)
    indexes: list[IndexDiff] = field(default_factory=list)

    def is_empty(self) -> bool:
        return not (self.tables or self.columns or self.indexes)


# ---------------------------------------------------------------------------
# Introspection
# ---------------------------------------------------------------------------
def make_engine(url: str) -> Engine:
    """Create a SQLAlchemy engine with a hard connect timeout."""
    try:
        return create_engine(
            url,
            connect_args={"connect_timeout": CONNECT_TIMEOUT},
            pool_pre_ping=True,
        )
    except SQLAlchemyError as exc:
        raise RuntimeError(f"Failed to create engine for {url!r}: {exc}") from exc


def snapshot(inspector: Inspector, schema: str) -> dict[str, Any]:
    """Return a normalized dict of tables -> columns/indexes."""
    snap: dict[str, Any] = {}
    for table_name in inspector.get_table_names(schema=schema):
        cols = {}
        for col in inspector.get_columns(table_name, schema=schema):
            # Normalize type to a SQL string so cross-dialect comparison works.
            type_sql = str(col["type"])
            cols[col["name"]] = {
                "type": type_sql,
                "nullable": bool(col.get("nullable", True)),
                "default": col.get("default"),
            }
        idxs = {}
        if INCLUDE_INDEXES:
            for idx in inspector.get_indexes(table_name, schema=schema):
                idxs[idx["name"]] = {
                    "columns": list(idx.get("column_names") or []),
                    "unique": bool(idx.get("unique", False)),
                }
        snap[table_name] = {"columns": cols, "indexes": idxs}
    return snap


# ---------------------------------------------------------------------------
# Diffing
# ---------------------------------------------------------------------------
def diff_snapshots(src: dict[str, Any], dst: dict[str, Any]) -> SchemaDiff:
    """
    Compute diff turning `src` into `dst`.
    Tables/columns in dst but not src -> ADD.
    Tables/columns in src but not dst -> DROP.
    """
    result = SchemaDiff()
    src_tables, dst_tables = set(src), set(dst)

    for t in sorted(dst_tables - src_tables):
        result.tables.append(TableDiff(name=t, action="ADD"))
    for t in sorted(src_tables - dst_tables):
        result.tables.append(TableDiff(name=t, action="DROP"))

    for t in sorted(src_tables & dst_tables):
        src_cols = src[t]["columns"]
        dst_cols = dst[t]["columns"]

        for c in sorted(set(dst_cols) - set(src_cols)):
            meta = dst_cols[c]
            result.columns.append(
                ColumnDiff(
                    table=t,
                    name=c,
                    action="ADD",
                    type_sql=meta["type"],
                    nullable=meta["nullable"],
                    default=meta["default"],
                )
            )
        for c in sorted(set(src_cols) - set(dst_cols)):
            result.columns.append(ColumnDiff(table=t, name=c, action="DROP"))

        # Index diffs (only for tables that exist on both sides)
        if INCLUDE_INDEXES:
            src_idx = src[t]["indexes"]
            dst_idx = dst[t]["indexes"]
            for i in sorted(set(dst_idx) - set(src_idx)):
                meta = dst_idx[i]
                result.indexes.append(
                    IndexDiff(
                        table=t,
                        name=i,
                        action="ADD",
                        columns=meta["columns"],
                        unique=meta["unique"],
                    )
                )
            for i in sorted(set(src_idx) - set(dst_idx)):
                result.indexes.append(
                    IndexDiff(table=t, name=i, action="DROP")
                )

    return result


# ---------------------------------------------------------------------------
# SQL rendering
# ---------------------------------------------------------------------------
def quote_ident(name: str) -> str:
    """Safely quote a PostgreSQL identifier."""
    return '"' + name.replace('"', '""') + '"'


def render_sql(diff: SchemaDiff, schema: str) -> str:
    lines: list[str] = [
        "-- Auto-generated schema diff",
        f"-- Target schema: {schema}",
        "BEGIN;",
        "",
    ]

    for t in diff.tables:
        if t.action == "DROP":
            lines.append(f"DROP TABLE IF EXISTS {quote_ident(t.name)} CASCADE;")
        else:
            # We can't reconstruct full CREATE TABLE from inspector alone;
            # emit a notice so the operator uses pg_dump for new tables.
            lines.append(
                f"-- NOTE: table {t.name} exists only in target. "
                f"Use pg_dump --schema-only to create it."
            )
    if diff.tables:
        lines.append("")

    for c in diff.columns:
        qtable = quote_ident(c.table)
        qcol = quote_ident(c.name)
        if c.action == "DROP":
            lines.append(f"ALTER TABLE {qtable} DROP COLUMN IF EXISTS {qcol};")
        else:
            nullable = "" if c.nullable else " NOT NULL"
            default = (
                f" DEFAULT {c.default}"
                if c.default and "nextval" not in str(c.default)
                else ""
            )
            lines.append(
                f"ALTER TABLE {qtable} ADD COLUMN {qcol} {c.type_sql}"
                f"{default}{nullable};"
            )
    if diff.columns:
        lines.append("")

    for i in diff.indexes:
        qtable = quote_ident(i.table)
        qidx = quote_ident(i.name)
        if i.action == "DROP":
            lines.append(f"DROP INDEX IF EXISTS {qidx};")
        else:
            unique = "UNIQUE " if i.unique else ""
            cols = ", ".join(quote_ident(c) for c in i.columns)
            lines.append(
                f"CREATE {unique}INDEX {qidx} ON {qtable} ({cols});"
            )
    if diff.indexes:
        lines.append("")

    lines.append("COMMIT;")
    lines.append("")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        description="Diff two PostgreSQL schemas and emit migration SQL."
    )
    p.add_argument("--db1", required=True, help="Source DB URL (e.g. prod)")
    p.add_argument("--db2", required=True, help="Target DB URL (e.g. staging)")
    p.add_argument("--schema", default=DEFAULT_SCHEMA, help="Schema name")
    p.add_argument("--output", help="Write SQL to this file (default: stdout)")
    p.add_argument(
        "--json",
        action="store_true",
        help="Emit JSON diff instead of SQL",
    )
    p.add_argument(
        "--check",
        action="store_true",
        help="Exit non-zero if any differences are found (CI mode).",
    )
    return p.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)

    try:
        eng1 = make_engine(args.db1)
        eng2 = make_engine(args.db2)
        with eng1.connect() as c1, eng2.connect() as c2:
            c1.execute(text("SELECT 1"))
            c2.execute(text("SELECT 1"))
            insp1 = inspect(c1)
            insp2 = inspect(c2)
            snap1 = snapshot(insp1, args.schema)
            snap2 = snapshot(insp2, args.schema)
    except (SQLAlchemyError, RuntimeError) as exc:
        log.error("Database connection failed: %s", exc)
        return 2

    log.info(
        "Snapshots: db1=%d tables, db2=%d tables",
        len(snap1),
        len(snap2),
    )

    diff = diff_snapshots(snap1, snap2)

    if diff.is_empty():
        log.info("No schema differences detected.")
        if args.output:
            with open(args.output, "w", encoding="utf-8") as fh:
                fh.write("-- No differences.\n")
        return 0

    if args.json:
        payload = {
            "tables": [asdict(t) for t in diff.tables],
            "columns": [asdict(c) for c in diff.columns],
            "indexes": [asdict(i) for i in diff.indexes],
        }
        output_text = json.dumps(payload, indent=2)
    else:
        output_text = render_sql(diff, args.schema)

    if args.output:
        with open(args.output, "w", encoding="utf-8") as fh:
            fh.write(output_text)
        log.info("Wrote diff to %s", args.output)
    else:
        print(output_text)

    if args.check:
        log.warning(
            "Differences found: %d tables, %d columns, %d indexes",
            len(diff.tables),
            len(diff.columns),
            len(diff.indexes),
        )
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())

Configuration

The two database URLs are required CLI arguments and are never read from env vars — this is deliberate so credentials never end up in shell history via a stale variable. Use a .pgpass file or a secrets manager wrapper instead.

Usage Examples

1. Compare prod to staging and write a migration file:

$ python3 schema_diff.py \
    --db1 postgresql://prod/app \
    --db2 postgresql://staging/app \
    --output diff.sql
2024-06-01 10:12:01 [INFO] Snapshots: db1=42 tables, db2=44 tables
2024-06-01 10:12:01 [INFO] Wrote diff to diff.sql

diff.sql will contain ALTER TABLE / CREATE INDEX statements wrapped in a transaction.

2. CI gate — fail the build if schemas drift:

$ python3 schema_diff.py --db1 $PROD_URL --db2 $STAGING_URL --check
2024-06-01 10:12:01 [WARNING] Differences found: 0 tables, 3 columns, 1 indexes
$ echo $?
1

3. Machine-readable output for a dashboard:

$ python3 schema_diff.py --db1 $A --db2 $B --json | jq '.columns'
[
  { "table": "users", "name": "email_verified", "action": "ADD",
    "type_sql": "BOOLEAN", "nullable": true, "default": null }
]

Error Handling

• Unreachable database / bad credentials — make_engine and the SELECT 1 probe both raise SQLAlchemyError; the script logs the message and exits with code 2 (distinct from 1, which signals "diff found").
• Missing schema — inspector.get_table_names(schema=...) returns an empty list rather than raising, so a typo'd schema name silently yields "no differences." Check the log line showing table counts to catch this.
• New tables — the inspector cannot reconstruct full CREATE TABLE DDL (defaults, constraints, sequences). The script emits a -- NOTE: comment telling the operator to use pg_dump --schema-only for those, rather than producing broken SQL.
• Column type changes — currently only add/drop are detected, not type/nullability mutations (see Extending Ideas). This is a deliberate scope limit to avoid emitting destructive ALTER COLUMN TYPE statements without review.
• File write errors — open(..., "w") will raise OSError and propagate; wrap the script in a shell set -e or catch in your orchestrator.

Extending Ideas

• Detect column type/nullability changes by comparing type_sql and nullable for columns present on both sides, then emitting ALTER TABLE ... ALTER COLUMN ... TYPE ... with a -- REVIEW REQUIRED banner.
• Full DDL for new tables by shelling out to pg_dump --schema-only --table=<name> for each ADD table and inlining the output into the migration file.
• Multi-dialect support by swapping create_engine URLs for MySQL/SQLite and adjusting quote_ident (MySQL uses backticks) — SQLAlchemy's inspector already abstracts most of the work.
• Apply mode with --apply that executes the generated SQL against db2 inside the same transaction, gated behind a --yes confirmation flag for safety.