Add migrate command

README.md

@@ -125,6 +125,60 @@ bootstraps the `sa` login — additional logins are a known limitation.
 the same category as `psql` for the Postgres path) rather than a Python
 REPL.
 
+## `pgdb migrate`
+
+Applies numbered, forward-only SQL migration files from a directory to a live
+Postgres database, tracking each one in a `schema.table` (default
+`public.schema_migrations`) so repeat runs only apply what's pending. Postgres only —
+not available for `--dialect mssql`.
+
+```bash
+pgdb migrate check  path/to/database/_migration_scripts --url postgresql://user:pass@host:port/db
+pgdb migrate apply  path/to/database/_migration_scripts --url postgresql://user:pass@host:port/db
+```
+
+`--entra-user` works the same as `pgdb compare` (see above). The tracking
+table needs `filename text primary key, applied_at timestamptz not null
+default now(), applied_by text not null default current_user` (a migration
+file that creates it, in the same directory, is the usual way to bootstrap
+it — inserting into a not-yet-existing tracking table is tolerated so that
+migration can still run).
+
+The tracking table defaults to `public.schema_migrations`. Override it per-project in
+`pyproject.toml`:
+
+```toml
+[tool.pgdevkit]
+migrations_table = "myschema.migrations"
+```
+
+or per-invocation with `--tracking-table`, which takes precedence over the
+pyproject.toml value.
+
+`--ask` prints each pending file and asks yes/no/already-done/quit before
+running it. Answering yes queues the file on a background worker and moves
+straight to the next prompt — you can keep reviewing while earlier files are
+still executing, instead of waiting on each one before seeing the next. A
+`tqdm` progress bar tracks the queue; migrations still run one at a time, in
+file order. Without `--ask`, `apply` queues every pending file up front and
+shows the same progress bar. `--file <name>` applies a single file (still
+through the same verify-and-track path) instead of walking all pending ones.
+Pass `--yes` to skip the "about to run migrations against ..." confirmation
+prompt (e.g. in CI).
+
+After each file's DDL is applied, `apply` re-checks that every `CREATE TABLE`
+statement's target actually exists (via `to_regclass`) before recording the
+file as applied — catching a migration that silently rolled back. That check
+parses each statement with `sqlglot` and only falls back to a regex (run
+against comment-stripped SQL) for statements sqlglot's postgres dialect can't
+parse, so a `CREATE TABLE` mentioned only in a `--` comment is never mistaken
+for a real one.
+
+`pgdevkit.migrate` is also usable directly as a library — `list_migration_files`,
+`applied_migrations`, `pending_migrations`, and `apply_migration` are the same
+functions the CLI calls, so a project can script around them without shelling
+out.
+
 ## `pgdevkit.db` — helpers for application code
 
 Install with the `db` extra: `pip install pgdevkit[db]`.

pgdevkit/cli.py

@@ -1,15 +1,18 @@
 from __future__ import annotations
 
 import os
+import queue
+import threading
 from pathlib import Path
 
 import psycopg
 import typer
 from rich.console import Console
 from rich.table import Table
 from rich import box
+from tqdm import tqdm
 
-from . import testdb
+from . import migrate, testdb
 from .backends import get_backend
 from .connection import build_conninfo
 from .diff import DiffKind, compute_diff
@@ -23,6 +26,11 @@
 testdb_app = typer.Typer(name="testdb", help="Manage the shared local Postgres test container")
 app.add_typer(testdb_app, name="testdb")
 
+migrate_app = typer.Typer(
+    name="migrate", help="Apply numbered, forward-only SQL migration files, tracked in a DB table"
+)
+app.add_typer(migrate_app, name="migrate")
+
 
 @app.command()
 def compare(
@@ -224,3 +232,177 @@ def testdb_clean(
     """Drop this workspace's database (or every database of this project with --all)."""
     testdb.clean_testdb(all=all)
     console.print("[green]Cleaned.[/green]")
+
+
+@migrate_app.command("check")
+def migrate_check(
+    migrations_dir: Path = typer.Argument(..., help="Directory of numbered .sql migration files"),
+    url: str = typer.Option(..., "--url", help="PostgreSQL DSN (postgresql://user:pass@host:port/db)"),
+    entra_user: str | None = typer.Option(None, "--entra-user", help="Azure Entra user (triggers token auth)"),
+    tracking_table: str | None = typer.Option(
+        None,
+        "--tracking-table",
+        help="schema.table recording applied migrations "
+        "(default: tool.pgdevkit.migrations_table in pyproject.toml, else public.schema_migrations)",
+    ),
+) -> None:
+    """List which migration files under migrations_dir are applied vs. pending."""
+    if not migrations_dir.is_dir():
+        err_console.print(f"[red]Error:[/red] {migrations_dir} is not a directory")
+        raise typer.Exit(2)
+
+    conninfo = build_conninfo(url, entra_user)
+    tracking_table = tracking_table or migrate.default_tracking_table(migrations_dir)
+    local_files = migrate.list_migration_files(migrations_dir)
+    try:
+        applied = migrate.applied_migrations(conninfo, tracking_table)
+    except migrate.TrackingTableMissing:
+        err_console.print(f"[yellow]⚠[/yellow]  {tracking_table} not found — nothing recorded as applied yet")
+        applied = {}
+
+    table = Table(box=box.SIMPLE, show_header=True, header_style="bold")
+    table.add_column("File")
+    table.add_column("Status")
+    table.add_column("Applied at", style="dim")
+    pending = []
+    for f in local_files:
+        if f.name in applied:
+            applied_at, applied_by = applied[f.name]
+            table.add_row(f.name, "[green]applied[/green]", f"{applied_at:%Y-%m-%d %H:%M} ({applied_by})")
+        else:
+            table.add_row(f.name, "[yellow]pending[/yellow]", "")
+            pending.append(f.name)
+    console.print(table)
+    console.print(f"\n{len(pending)} pending, {len(applied)} applied, {len(local_files)} total")
+
+
+@migrate_app.command("apply")
+def migrate_apply(
+    migrations_dir: Path = typer.Argument(..., help="Directory of numbered .sql migration files"),
+    url: str = typer.Option(..., "--url", help="PostgreSQL DSN (postgresql://user:pass@host:port/db)"),
+    entra_user: str | None = typer.Option(None, "--entra-user", help="Azure Entra user (triggers token auth)"),
+    tracking_table: str | None = typer.Option(
+        None,
+        "--tracking-table",
+        help="schema.table recording applied migrations "
+        "(default: tool.pgdevkit.migrations_table in pyproject.toml, else public.schema_migrations)",
+    ),
+    file: str | None = typer.Option(
+        None, "--file", help="Apply only this one filename (relative to migrations_dir) instead of all pending"
+    ),
+    ask: bool = typer.Option(False, "--ask", help="Show and confirm each migration before running it"),
+    yes: bool = typer.Option(False, "--yes", "-y", help="Skip the confirm-target prompt"),
+) -> None:
+    """Apply pending migration files, in filename order, tracking each in tracking_table."""
+    if not migrations_dir.is_dir():
+        err_console.print(f"[red]Error:[/red] {migrations_dir} is not a directory")
+        raise typer.Exit(2)
+
+    conninfo = build_conninfo(url, entra_user)
+    tracking_table = tracking_table or migrate.default_tracking_table(migrations_dir)
+    target_desc = url.rsplit("@", 1)[-1] if "@" in url else url
+    if not yes:
+        typer.confirm(f"About to run migrations against {target_desc}. Continue?", abort=True)
+
+    if file:
+        targets = [migrations_dir / file]
+    else:
+        try:
+            targets = migrate.pending_migrations(migrations_dir, conninfo, tracking_table)
+        except migrate.TrackingTableMissing:
+            targets = migrate.list_migration_files(migrations_dir)
+
+    if not targets:
+        console.print("No pending migrations.")
+        return
+
+    # A single background worker applies queued migrations in file order (each still waits
+    # for the previous one to land) while --ask keeps prompting for the *next* file, instead
+    # of the review blocking on every execution.
+    work_q: queue.Queue[tuple[Path, bool] | None] = queue.Queue()
+    failure: Exception | None = None
+    stop = threading.Event()
+    outcomes: list[tuple[str, str]] = []
+    bar = tqdm(total=len(targets), unit="migration", desc="Applying")
+
+    def worker() -> None:
+        nonlocal failure
+        for path, already_done in iter(work_q.get, None):
+            if not stop.is_set():
+                try:
+                    result = migrate.apply_migration(conninfo, path, tracking_table, already_done=already_done)
+                except Exception as e:  # noqa: BLE001
+                    failure = e
+                    stop.set()
+                    bar.write(f"FAILED {path.name}: {e}")
+                    outcomes.append((path.name, "failed"))
+                else:
+                    if result.executed:
+                        for tbl in result.verified_tables:
+                            bar.write(f"  table {tbl} exists")
+                        bar.write(f"Applied {path.name}")
+                        outcomes.append((path.name, "applied"))
+                    else:
+                        bar.write(f"Recorded {path.name} as already applied (not executed)")
+                        outcomes.append((path.name, "recorded"))
+            bar.update(1)
+            work_q.task_done()
+
+    thread = threading.Thread(target=worker, daemon=True)
+    thread.start()
+
+    quit_requested = False
+    for path in targets:
+        if stop.is_set():
+            break
+        already_done = False
+        if ask:
+            bar.write(f"\n=== {path.name} ===")
+            bar.write(path.read_text(encoding="utf-8"))
+            answer = typer.prompt("[Y]es execute / [n]o skip / [a]lready done / [q]uit", default="y").strip().lower()
+            if answer in ("q", "quit"):
+                quit_requested = True
+                break
+            if answer in ("n", "no"):
+                bar.write(f"Skipped {path.name}")
+                outcomes.append((path.name, "skipped"))
+                bar.update(1)
+                continue
+            if answer in ("a", "already", "already done"):
+                already_done = True
+            elif answer not in ("", "y", "yes"):
+                bar.write(f"Skipped {path.name}")
+                outcomes.append((path.name, "skipped"))
+                bar.update(1)
+                continue
+
+        work_q.put((path, already_done))
+
+    work_q.put(None)
+    thread.join()
+    bar.close()
+
+    processed = {name for name, _ in outcomes}
+    for path in targets:
+        if stop.is_set() and path.name not in processed:
+            outcomes.append((path.name, "not run (stopped after earlier failure)"))
+
+    table = Table(box=box.SIMPLE, show_header=True, header_style="bold")
+    table.add_column("File")
+    table.add_column("Result")
+    status_style = {
+        "applied": "[green]applied[/green]",
+        "recorded": "[green]recorded (already done)[/green]",
+        "skipped": "[yellow]skipped[/yellow]",
+        "failed": "[red]failed[/red]",
+    }
+    for name, status in outcomes:
+        table.add_row(name, status_style.get(status, status))
+    console.print(table)
+
+    if quit_requested:
+        console.print("Aborted.")
+        raise typer.Exit(1)
+    if failure is not None:
+        err_console.print(f"[red]✗[/red] {failure}")
+        raise typer.Exit(1)