Build an AI Database Migration Tool with Python

pip install anthropic psycopg2-binary click

mkdir -p ai-migrate/migrations
cd ai-migrate
touch migrate.py

import psycopg2
import json

def introspect_schema(dsn: str) -> dict:
    """Pull full schema from PostgreSQL into a dict."""
    conn = psycopg2.connect(dsn)
    cur = conn.cursor()

    # Get all tables and columns
    cur.execute("""
        SELECT table_name, column_name, data_type,
               is_nullable, column_default
        FROM information_schema.columns
        WHERE table_schema = 'public'
        ORDER BY table_name, ordinal_position
    """)

    tables = {}
    for table, col, dtype, nullable, default in cur.fetchall():
        if table not in tables:
            tables[table] = {"columns": [], "indexes": [], "fkeys": []}
        tables[table]["columns"].append({
            "name": col, "type": dtype,
            "nullable": nullable == "YES",
            "default": default
        })

    # Get indexes
    cur.execute("""
        SELECT tablename, indexname, indexdef
        FROM pg_indexes WHERE schemaname = 'public'
    """)
    for table, name, definition in cur.fetchall():
        if table in tables:
            tables[table]["indexes"].append({
                "name": name, "definition": definition
            })

    # Get foreign keys
    cur.execute("""
        SELECT tc.table_name, kcu.column_name,
               ccu.table_name AS ref_table,
               ccu.column_name AS ref_column
        FROM information_schema.table_constraints tc
        JOIN information_schema.key_column_usage kcu
          ON tc.constraint_name = kcu.constraint_name
        JOIN information_schema.constraint_column_usage ccu
          ON tc.constraint_name = ccu.constraint_name
        WHERE tc.constraint_type = 'FOREIGN KEY'
    """)
    for table, col, ref_table, ref_col in cur.fetchall():
        if table in tables:
            tables[table]["fkeys"].append({
                "column": col, "references": f"{ref_table}.{ref_col}"
            })

    conn.close()
    return tables

import anthropic
import json
from datetime import datetime

client = anthropic.Anthropic(
    api_key="sk-your-ezai-key",
    base_url="https://ezaiapi.com"
)

SYSTEM_PROMPT = """You are a PostgreSQL migration expert. Given a database
schema and a requested change, generate:

1. An UP migration (apply the change)
2. A DOWN migration (revert the change)
3. A risk assessment

Rules:
- Use CREATE INDEX CONCURRENTLY for indexes on tables with data
- Never use ALTER TABLE ... ADD COLUMN ... NOT NULL without a DEFAULT
- Wrap destructive operations in transactions
- Flag any operation that locks the table for writes
- Output valid PostgreSQL SQL only

Return JSON with keys: up_sql, down_sql, risks (array of strings),
estimated_lock_time (string like "none", "<1s", "minutes")."""

def generate_migration(schema: dict, request: str) -> dict:
    message = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=4096,
        system=SYSTEM_PROMPT,
        messages=[{
            "role": "user",
            "content": f"""Current schema:
{json.dumps(schema, indent=2)}

Requested change: {request}"""
        }]
    )

    # Parse the JSON response
    text = message.content[0].text
    # Strip markdown code fences if present
    if "```json" in text:
        text = text.split("```json")[1].split("```")[0]
    elif "```" in text:
        text = text.split("```")[1].split("```")[0]

    return json.loads(text)

import click
import os

@click.group()
def cli():
    """AI-powered database migration tool."""
    pass

@cli.command()
@click.option("--dsn", envvar="DATABASE_URL", required=True)
@click.argument("description")
def create(dsn, description):
    """Generate a migration from a plain-English description."""
    click.echo("🔍 Reading schema...")
    schema = introspect_schema(dsn)
    click.echo(f"   Found {len(schema)} tables")

    click.echo("🧠 Generating migration with Claude...")
    result = generate_migration(schema, description)

    # Create versioned migration files
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    slug = description[:40].lower().replace(" ", "_")
    base = f"migrations/{ts}_{slug}"

    with open(f"{base}.up.sql", "w") as f:
        f.write(result["up_sql"])
    with open(f"{base}.down.sql", "w") as f:
        f.write(result["down_sql"])

    # Show risk assessment
    click.echo(f"\n✅ Migration created: {base}")
    click.echo(f"   Lock time: {result['estimated_lock_time']}")

    if result["risks"]:
        click.echo("\n⚠️  Risks:")
        for risk in result["risks"]:
            click.echo(f"   • {risk}")

@cli.command()
@click.option("--dsn", envvar="DATABASE_URL", required=True)
def validate(dsn):
    """Validate pending migrations against the current schema."""
    schema = introspect_schema(dsn)
    pending = sorted(
        [f for f in os.listdir("migrations")
         if f.endswith(".up.sql")]
    )

    for migration in pending:
        sql = open(f"migrations/{migration}").read()
        click.echo(f"Validating {migration}...")

        result = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=2048,
            messages=[{
                "role": "user",
                "content": f"""Review this PostgreSQL migration against the schema.
Flag: syntax errors, missing indexes for new FKs, locking risks,
data loss potential. Be specific.

Schema: {json.dumps(schema)}
SQL: {sql}"""
            }]
        )
        click.echo(result.content[0].text)

if __name__ == "__main__":
    cli()

# Generate a migration
python migrate.py create "add email_verified boolean to users table"

# Validate before applying
python migrate.py validate

# Apply with psql
psql $DATABASE_URL -f migrations/20260318_142000_add_email_verified.up.sql

def filter_relevant_tables(schema: dict, request: str) -> dict:
    """Use Claude to pick which tables are relevant."""
    table_names = list(schema.keys())

    resp = client.messages.create(
        model="claude-haiku-3-5",  # Cheap model for filtering
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"""Which tables are relevant to this migration?
Tables: {table_names}
Request: {request}
Return JSON array of table names only."""
        }]
    )

    relevant = json.loads(resp.content[0].text)
    return {t: schema[t] for t in relevant if t in schema}