Worktree pgdevkit testdb

.gitignore

@@ -11,3 +11,6 @@ wheels/
 
 # Local planning docs (not shipped with the package)
 docs/superpowers/
+
+# Local migration checklist (not shipped with the package)
+/MIGRATION_NOTES.md

README.md

@@ -0,0 +1,42 @@
+# pgdevkit
+
+A helper for developing with Postgres.
+
+## `pgdb compare`
+
+Compare a directory of SQL scripts (see the `database-in-source` layout
+convention) against a live database and report differences:
+
+```bash
+pgdb compare --url postgresql://user:pass@host:port/db path/to/database/
+```
+
+## `pgdb testdb`
+
+Manages a single shared, Podman-backed Postgres container for local tests
+across all your projects — no more one-container-per-project-per-worktree.
+Isolation between projects and worktrees is per-database, inside one
+container.
+
+Add to `pyproject.toml`:
+
+```toml
+[tool.pgdevkit]
+name = "myproject"        # optional; defaults to the repo directory name
+database_dir = "database" # optional; defaults to "database"
+```
+
+Add to `conftest.py`:
+
+```python
+import os
+import pytest
+from pgdevkit.testdb import ensure_testdb
+
+@pytest.fixture(scope="session", autouse=True)
+def ensure_test_postgres():
+    for k, v in ensure_testdb().items():
+        os.environ[k] = v
+```
+
+CLI: `pgdb testdb up|reset|run-sql|status|shell|clean`.

pgdevkit/cli.py

@@ -1,12 +1,14 @@
 from __future__ import annotations
 
+import os
 from pathlib import Path
 
 import typer
 from rich.console import Console
 from rich.table import Table
 from rich import box
 
+from . import testdb
 from .connection import build_conninfo
 from .diff import DiffKind, compute_diff
 from .introspect import introspect_db
@@ -16,10 +18,12 @@
 console = Console()
 err_console = Console(stderr=True)
 
+testdb_app = typer.Typer(name="testdb", help="Manage the shared local Postgres test container")
+app.add_typer(testdb_app, name="testdb")
+
 
 @app.command()
 def compare(
-
     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)"),
     report_extra_db: bool = typer.Option(False, "--report-extra-db", help="Report objects in DB but not in scripts"),
@@ -61,3 +65,73 @@ def compare(
     console.print(table)
     console.print(f"\n[bold red]{len(diffs)} difference(s) found.[/bold red]")
     raise typer.Exit(1)
+
+
+@testdb_app.command("up")
+def testdb_up() -> None:
+    """Ensure the container is running, the workspace DB exists, and schema is applied."""
+    testdb.ensure_testdb()
+    info = testdb.status()
+    console.print(f"[green]Test DB ready:[/green] {info['database']} ({info['dsn']})")
+
+
+@testdb_app.command("reset")
+def testdb_reset() -> None:
+    """Drop and recreate only this workspace's database, then reapply schema + seed data."""
+    testdb.reset_testdb()
+    info = testdb.status()
+    console.print(f"[green]Test DB reset:[/green] {info['database']}")
+
+
+@testdb_app.command("run-sql")
+def testdb_run_sql(
+    file: Path | None = typer.Argument(None, help="Path to a .sql file"),
+    sql: str | None = typer.Option(None, "--sql", help="Inline SQL string"),
+    results: bool = typer.Option(False, "--results", help="Print query results as a table"),
+) -> None:
+    """Run SQL against this workspace's database."""
+    if (file is None) == (sql is None):
+        err_console.print("[red]Error:[/red] pass exactly one of FILE or --sql")
+        raise typer.Exit(2)
+    statement = file.read_text(encoding="utf-8") if file else sql
+    assert statement is not None
+    rows = testdb.run_sql(statement)
+
+    if rows is None:
+        console.print("OK")
+        return
+    if not results:
+        console.print(f"OK — {len(rows)} row(s)")
+        return
+    if not rows:
+        console.print("(0 rows)")
+        return
+    table = Table(box=box.SIMPLE, show_header=True, header_style="bold")
+    for col in rows[0]:
+        table.add_column(col)
+    for row in rows:
+        table.add_row(*(str(v) for v in row.values()))
+    console.print(table)
+    console.print(f"({len(rows)} row(s))")
+
+
+@testdb_app.command("status")
+def testdb_status() -> None:
+    """Show container state, this workspace's database name, and DSN."""
+    for key, value in testdb.status().items():
+        console.print(f"{key}: {value}")
+
+
+@testdb_app.command("shell")
+def testdb_shell() -> None:
+    """Drop into psql against this workspace's database."""
+    os.execvp("psql", ["psql", testdb.dsn_for()])
+
+
+@testdb_app.command("clean")
+def testdb_clean(
+    all: bool = typer.Option(False, "--all", help="Drop every database belonging to this project"),
+) -> None:
+    """Drop this workspace's database (or every database of this project with --all)."""
+    testdb.clean_testdb(all=all)
+    console.print("[green]Cleaned.[/green]")

pgdevkit/diff.py

@@ -204,7 +204,10 @@ def _unwrap_paren(node):
 
 
 def _norm_sql(s: str) -> str:
-    return " ".join(s.lower().split())
+    # Postgres's pg_get_viewdef()/pg_get_indexdef() always append a trailing
+    # ";", while sqlglot's Expression.sql() rendering never does; strip it so
+    # semantically-identical definitions compare equal.
+    return " ".join(s.lower().split()).rstrip(";")
 
 
 def _norm_body(s: str) -> str:

pgdevkit/parser.py

@@ -197,15 +197,25 @@ def _handle_view(expr: exp.Create, db_schema: DatabaseSchema) -> None:
 def _handle_function(expr: exp.Create, raw: str, db_schema: DatabaseSchema, kind: str) -> None:
     # Get name/schema from sqlglot
     func_node = expr.this
-    if hasattr(func_node, "name"):
-        fname = func_node.name
+    fname = func_node.name if hasattr(func_node, "name") else ""
+    if fname:
         db_node = func_node.args.get("db") if hasattr(func_node, "args") else None
         fschema = db_node.name if db_node else "public"
     else:
-        result = _resolve_name(expr)
-        if not result:
-            return
-        fschema, fname = result
+        # sqlglot (30.11.0) parses "CREATE FUNCTION myapp.greet(...)" as a
+        # UserDefinedFunction wrapping a Table (this=Identifier(greet),
+        # db=Identifier(myapp)). UserDefinedFunction.name doesn't unwrap
+        # that nested Table, so pull the name/schema from it directly.
+        inner = func_node.this if hasattr(func_node, "this") else None
+        if isinstance(inner, exp.Table) and inner.name:
+            fname = inner.name
+            db_node = inner.args.get("db")
+            fschema = db_node.name if db_node else "public"
+        else:
+            result = _resolve_name(expr)
+            if not result:
+                return
+            fschema, fname = result
 
     # Extract args, return type, language, body with regex on raw SQL
     args, return_type, language, body = _parse_function_details(raw)
@@ -275,8 +285,15 @@ def _handle_type(expr: exp.Create, db_schema: DatabaseSchema) -> None:
 
 def _handle_schema_create(expr: exp.Create, db_schema: DatabaseSchema) -> None:
     this = expr.this
-    if hasattr(this, "name"):
-        db_schema.schemas.add(this.name)
+    name = this.name if hasattr(this, "name") else ""
+    if not name:
+        # sqlglot (30.11.0) parses "CREATE SCHEMA myapp" as a Table node
+        # whose `db` arg holds the schema name and `.name` (the table
+        # identifier) is empty.
+        db_node = this.args.get("db") if hasattr(this, "args") else None
+        name = db_node.name if db_node else ""
+    if name:
+        db_schema.schemas.add(name)
 
 
 def _handle_index(expr: exp.Create, db_schema: DatabaseSchema) -> None:

pgdevkit/testdb/__init__.py

@@ -0,0 +1,3 @@
+from .api import clean_testdb, dsn_for, ensure_testdb, reset_testdb, run_sql, status
+
+__all__ = ["clean_testdb", "dsn_for", "ensure_testdb", "reset_testdb", "run_sql", "status"]

pgdevkit/testdb/api.py

@@ -0,0 +1,139 @@
+from __future__ import annotations
+
+import asyncio
+import os
+from pathlib import Path
+
+import psycopg
+from psycopg.sql import SQL, Identifier
+
+from . import constants, query
+from .config import ProjectConfig, load_config
+from .container import ensure_container
+from .naming import current_branch, slugify, workspace_db_name
+from .schema import apply_schema
+
+
+def _admin_dsn() -> str:
+    return f"postgresql://{constants.USER}:{constants.PASSWORD}@{constants.HOST}:{constants.PORT}/postgres?connect_timeout=10"
+
+
+def _db_dsn(db_name: str) -> str:
+    return f"postgresql://{constants.USER}:{constants.PASSWORD}@{constants.HOST}:{constants.PORT}/{db_name}?connect_timeout=10"
+
+
+def _resolve(project_root: Path | None) -> tuple[ProjectConfig, str]:
+    config = load_config(project_root)
+    branch = current_branch(config.root)
+    db_name = workspace_db_name(config.name, branch)
+    return config, db_name
+
+
+def _env_for(config: ProjectConfig, db_name: str) -> dict[str, str]:
+    prefix = config.env_prefix
+    return {
+        f"{prefix}POSTGRES_HOST": constants.HOST,
+        f"{prefix}POSTGRES_PORT": str(constants.PORT),
+        f"{prefix}POSTGRES_DB": db_name,
+        f"{prefix}POSTGRES_USER": constants.USER,
+        f"{prefix}POSTGRES_PASSWORD": constants.PASSWORD,
+    }
+
+
+async def _ensure_database(db_name: str) -> None:
+    async with await psycopg.AsyncConnection.connect(_admin_dsn(), autocommit=True) as con:
+        result = await con.execute("SELECT 1 FROM pg_database WHERE datname = %(db)s", {"db": db_name})
+        if await result.fetchone():
+            return
+        try:
+            await con.execute(SQL("CREATE DATABASE {}").format(Identifier(db_name)))
+        except psycopg.errors.DuplicateDatabase:
+            pass
+
+
+async def _drop_database(db_name: str) -> None:
+    async with await psycopg.AsyncConnection.connect(_admin_dsn(), autocommit=True) as con:
+        await con.execute(
+            "SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE datname = %(db)s",
+            {"db": db_name},
+        )
+        await con.execute(SQL("DROP DATABASE IF EXISTS {}").format(Identifier(db_name)))
+
+
+async def _apply(config: ProjectConfig, db_name: str, force_reset: bool) -> None:
+    async with await psycopg.AsyncConnection.connect(_db_dsn(db_name), autocommit=True) as con:
+        await apply_schema(
+            con,
+            config.root / config.database_dir,
+            extensions=config.extensions,
+            force_reset=force_reset,
+        )
+
+
+def ensure_testdb(project_root: Path | None = None, force_reset: bool = False) -> dict[str, str]:
+    """Ensure the shared container is running, this workspace's database
+    exists, and its schema is applied. Returns the {PREFIX}POSTGRES_* env
+    vars for this workspace."""
+    if not os.environ.get("PGDEVKIT_SKIP_CONTAINER"):
+        ensure_container()
+    config, db_name = _resolve(project_root)
+
+    async def _run() -> None:
+        if force_reset:
+            await _drop_database(db_name)
+        await _ensure_database(db_name)
+        await _apply(config, db_name, force_reset)
+
+    asyncio.run(_run())
+    return _env_for(config, db_name)
+
+
+def reset_testdb(project_root: Path | None = None) -> dict[str, str]:
+    """Drop and recreate only this workspace's database, then reapply
+    schema and seed data."""
+    return ensure_testdb(project_root, force_reset=True)
+
+
+def clean_testdb(project_root: Path | None = None, all: bool = False) -> None:
+    """Drop this workspace's database. With all=True, drop every database
+    belonging to this project (matched by its name-slug prefix), across
+    every worktree/branch."""
+    config, db_name = _resolve(project_root)
+
+    async def _run() -> None:
+        if not all:
+            await _drop_database(db_name)
+            return
+        prefix = f"{slugify(config.name)}_"
+        escaped_prefix = prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        async with await psycopg.AsyncConnection.connect(_admin_dsn(), autocommit=True) as con:
+            result = await con.execute(
+                "SELECT datname FROM pg_database WHERE datname LIKE %(pattern)s ESCAPE '\\'",
+                {"pattern": f"{escaped_prefix}%"},
+            )
+            names = [row[0] for row in await result.fetchall()]
+        for name in names:
+            await _drop_database(name)
+
+    asyncio.run(_run())
+
+
+def status(project_root: Path | None = None) -> dict[str, str]:
+    config, db_name = _resolve(project_root)
+    return {
+        "container": constants.CONTAINER_NAME,
+        "host": constants.HOST,
+        "port": str(constants.PORT),
+        "database": db_name,
+        "dsn": _db_dsn(db_name),
+    }
+
+
+def run_sql(sql: str, project_root: Path | None = None) -> list[dict] | None:
+    _, db_name = _resolve(project_root)
+    return asyncio.run(query.execute(_db_dsn(db_name), sql))
+
+
+def dsn_for(project_root: Path | None = None) -> str:
+    _, db_name = _resolve(project_root)
+    return _db_dsn(db_name)