Add find_orphaned_dbs() / clean_testdb(orphaned=True)

README.md

@@ -200,8 +200,8 @@ def ensure_test_postgres():
         os.environ[k] = v
 ```
 
-CLI: `pgdb testdb up|reset|run-sql|status|shell|clean`. `up`/`reset` accept
-`--env` (default `local_test`) — see "Environment-tagged files" above.
+CLI: `pgdb testdb up|reset|run-sql|status|shell|clean|list-orphaned`. `up`/`reset`
+accept `--env` (default `local_test`) — see "Environment-tagged files" above.
 
 `up`/`reset` accept `--area`/`--exclude-area` and `--schema`/`--exclude-schema`
 (see "Area and schema filtering" above) to scope which `database/` files get
@@ -210,6 +210,27 @@ applied — e.g. `pgdb testdb up --schema billing` for a test DB with only the
 project's schema to apply. `ensure_testdb`/`reset_testdb` take the same
 keyword arguments when called from Python (e.g. from a pytest fixture).
 
+Every git worktree/branch of a project gets its own database, named after
+`project_name` + branch (see `pgdevkit.testdb.naming.workspace_db_name`).
+Removing a worktree (or deleting its directory without `git worktree
+remove`) doesn't drop its database — `pgdb testdb list-orphaned` lists
+this project's databases whose worktree no longer exists, and
+`pgdb testdb clean --orphaned` drops them (as opposed to `--all`, which
+drops every database of this project regardless of whether its worktree is
+still live). The same is available from Python as
+`pgdevkit.testdb.find_orphaned_dbs()` and
+`pgdevkit.testdb.clean_testdb(orphaned=True)`.
+
+If your project's own test setup also creates a sibling database per
+worktree (e.g. `<main_db>_myservice` for a mock service used only by that
+project's tests), add its literal suffix so orphan detection knows it
+belongs to a live worktree too:
+
+```toml
+[tool.pgdevkit]
+extra_db_suffixes = ["_myservice"]
+```
+
 Container connection defaults (`localhost:54322`, `postgres`/`testpwd`) can
 be overridden with `PGDEVKIT_TESTDB_HOST`, `PGDEVKIT_TESTDB_PORT`,
 `PGDEVKIT_TESTDB_USER`, `PGDEVKIT_TESTDB_PASSWORD`. Before touching the

pgdevkit/cli.py

@@ -302,12 +302,33 @@ def testdb_shell() -> None:
 @testdb_app.command("clean")
 def testdb_clean(
     all: bool = typer.Option(False, "--all", help="Drop every database belonging to this project"),
+    orphaned: bool = typer.Option(
+        False,
+        "--orphaned",
+        help="Drop only databases whose git worktree no longer exists (see `testdb list-orphaned`)",
+    ),
 ) -> None:
-    """Drop this workspace's database (or every database of this project with --all)."""
-    testdb.clean_testdb(all=all)
+    """Drop this workspace's database (or every database of this project with --all,
+    or only its orphaned ones with --orphaned)."""
+    if all and orphaned:
+        err_console.print("[red]Error:[/red] pass at most one of --all, --orphaned")
+        raise typer.Exit(2)
+    testdb.clean_testdb(all=all, orphaned=orphaned)
     console.print("[green]Cleaned.[/green]")
 
 
+@testdb_app.command("list-orphaned")
+def testdb_list_orphaned() -> None:
+    """List this project's databases whose git worktree no longer exists,
+    without dropping them."""
+    names = testdb.find_orphaned_dbs()
+    if not names:
+        console.print("No orphaned databases.")
+        return
+    for name in names:
+        console.print(name)
+
+
 @migrate_app.command("check")
 def migrate_check(
     migrations_dir: Path = typer.Argument(..., help="Directory of numbered .sql migration files"),

pgdevkit/testdb/__init__.py

@@ -1,3 +1,21 @@
-from .api import clean_testdb, dsn_for, ensure_testdb, reset_testdb, run_sql, shell_argv, status
+from .api import (
+    clean_testdb,
+    dsn_for,
+    ensure_testdb,
+    find_orphaned_dbs,
+    reset_testdb,
+    run_sql,
+    shell_argv,
+    status,
+)
 
-__all__ = ["clean_testdb", "dsn_for", "ensure_testdb", "reset_testdb", "run_sql", "shell_argv", "status"]
+__all__ = [
+    "clean_testdb",
+    "dsn_for",
+    "ensure_testdb",
+    "find_orphaned_dbs",
+    "reset_testdb",
+    "run_sql",
+    "shell_argv",
+    "status",
+]

pgdevkit/testdb/api.py

@@ -9,7 +9,14 @@
 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 .naming import (
+    current_branch,
+    escape_like_prefix,
+    expected_db_names,
+    live_worktree_branches,
+    slugify,
+    workspace_db_name,
+)
 from .schema import apply_schema
 
 
@@ -68,6 +75,16 @@ async def _drop_database(db_name: str) -> None:
         await con.execute(SQL("DROP DATABASE IF EXISTS {}").format(Identifier(db_name)))
 
 
+async def _dbs_with_prefix(prefix: str) -> list[str]:
+    escaped_prefix = escape_like_prefix(prefix)
+    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}%"},
+        )
+        return [row[0] for row in await result.fetchall()]
+
+
 async def _apply(
     config: ProjectConfig,
     db_name: str,
@@ -154,27 +171,45 @@ def reset_testdb(
     )
 
 
-def clean_testdb(project_root: Path | None = None, all: bool = False) -> None:
+async def _find_orphaned_dbs(config: ProjectConfig) -> list[str]:
+    prefix = f"{slugify(config.name)}_"
+    actual = await _dbs_with_prefix(prefix)
+    expected = expected_db_names(config, live_worktree_branches(config.root))
+    return sorted(set(actual) - expected)
+
+
+def find_orphaned_dbs(project_root: Path | None = None) -> list[str]:
+    """Databases belonging to this project (matched by its name-slug prefix)
+    that don't belong to any currently live git worktree of this repo --
+    i.e. their branch's worktree was removed (or never existed) without
+    also dropping its database."""
+    config, _ = _resolve(project_root)
+    if config.engine == "mssql":
+        return _mssql_api().find_orphaned_dbs(config)
+    return asyncio.run(_find_orphaned_dbs(config))
+
+
+def clean_testdb(project_root: Path | None = None, all: bool = False, orphaned: 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."""
+    every worktree/branch. With orphaned=True, drop only those without a
+    currently live git worktree (see `find_orphaned_dbs`). At most one of
+    all/orphaned may be set."""
+    if all and orphaned:
+        raise ValueError("clean_testdb: pass at most one of all=True, orphaned=True")
+
     config, db_name = _resolve(project_root)
     if config.engine == "mssql":
-        _mssql_api().clean_testdb(config, db_name, all)
+        _mssql_api().clean_testdb(config, db_name, all, orphaned)
         return
 
     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()]
+        if orphaned:
+            names = await _find_orphaned_dbs(config)
+        elif all:
+            names = await _dbs_with_prefix(f"{slugify(config.name)}_")
+        else:
+            names = [db_name]
         for name in names:
             await _drop_database(name)
 

pgdevkit/testdb/config.py

@@ -14,6 +14,7 @@ class ProjectConfig:
     database_dir: str = "database"
     env_prefix: str = ""
     extensions: tuple[str, ...] = ()
+    extra_db_suffixes: tuple[str, ...] = ()
     engine: str = "postgres"
     root: Path = field(default_factory=Path)
 
@@ -48,6 +49,17 @@ def load_config(start: Path | None = None) -> ProjectConfig:
             f"[tool.pgdevkit].extensions in {pyproject} must be a list, got {type(extensions).__name__}"
         )
 
+    # Lets a repo that layers an extra, literally-suffixed sibling database
+    # on top of its main workspace DB (e.g. a mock-service DB used only by
+    # that repo's own test setup) teach `find_orphaned_dbs`/`clean_testdb`
+    # about it, without pgdevkit needing to know why that suffix exists.
+    extra_db_suffixes = section.get("extra_db_suffixes", [])
+    if not isinstance(extra_db_suffixes, list):
+        raise TypeError(
+            f"[tool.pgdevkit].extra_db_suffixes in {pyproject} must be a list, "
+            f"got {type(extra_db_suffixes).__name__}"
+        )
+
     # PGDEVKIT_TESTDB_ENGINE lets CI/ad-hoc runs flip engines without
     # editing pyproject.toml; the toml value is the durable, per-project
     # default (a project's database/ tree is written in one dialect, so
@@ -59,6 +71,7 @@ def load_config(start: Path | None = None) -> ProjectConfig:
         database_dir=section.get("database_dir", "database"),
         env_prefix=section.get("env_prefix", ""),
         extensions=tuple(extensions),
+        extra_db_suffixes=tuple(extra_db_suffixes),
         engine=engine,
         root=root,
     )

pgdevkit/testdb/mssql/api.py

@@ -168,26 +168,46 @@ async def _run() -> None:
     return _env_for(config, db_name)
 
 
-def clean_testdb(config: ProjectConfig, db_name: str, all: bool) -> None:
-    from ..naming import slugify
+async def _dbs_with_prefix(prefix: str) -> list[str]:
+    from ..naming import escape_like_prefix
 
-    async def _run() -> None:
-        if not all:
-            await _drop_database(db_name)
-            return
-        prefix = f"{slugify(config.name)}_"
-        escaped_prefix = prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+    escaped_prefix = escape_like_prefix(prefix)
 
-        def _list_names() -> list[str]:
-            conn = mssql_python.connect(_admin_dsn(), autocommit=True)
-            try:
-                cur = conn.cursor()
-                cur.execute("SELECT name FROM sys.databases WHERE name LIKE ? ESCAPE '\\'", [f"{escaped_prefix}%"])
-                return [row[0] for row in cur.fetchall()]
-            finally:
-                conn.close()
+    def _list_names() -> list[str]:
+        conn = mssql_python.connect(_admin_dsn(), autocommit=True)
+        try:
+            cur = conn.cursor()
+            cur.execute("SELECT name FROM sys.databases WHERE name LIKE ? ESCAPE '\\'", [f"{escaped_prefix}%"])
+            return [row[0] for row in cur.fetchall()]
+        finally:
+            conn.close()
 
-        names = await asyncio.to_thread(_list_names)
+    return await asyncio.to_thread(_list_names)
+
+
+async def _find_orphaned_dbs(config: ProjectConfig) -> list[str]:
+    from ..naming import expected_db_names, live_worktree_branches, slugify
+
+    prefix = f"{slugify(config.name)}_"
+    actual = await _dbs_with_prefix(prefix)
+    expected = expected_db_names(config, live_worktree_branches(config.root))
+    return sorted(set(actual) - expected)
+
+
+def find_orphaned_dbs(config: ProjectConfig) -> list[str]:
+    return asyncio.run(_find_orphaned_dbs(config))
+
+
+def clean_testdb(config: ProjectConfig, db_name: str, all: bool, orphaned: bool = False) -> None:
+    from ..naming import slugify
+
+    async def _run() -> None:
+        if orphaned:
+            names = await _find_orphaned_dbs(config)
+        elif all:
+            names = await _dbs_with_prefix(f"{slugify(config.name)}_")
+        else:
+            names = [db_name]
         for name in names:
             await _drop_database(name)
 

pgdevkit/testdb/naming.py

@@ -4,6 +4,10 @@
 import re
 import subprocess
 from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .config import ProjectConfig
 
 _INVALID_CHARS = re.compile(r"[^a-z0-9_]+")
 _MAX_SLUG_LEN = 30
@@ -39,3 +43,70 @@ def workspace_db_name(project_name: str, branch: str) -> str:
     both inputs are already at the per-component truncation limit."""
     joined = f"{slugify(project_name)}_{slugify(branch)}"
     return slugify(joined)
+
+
+def live_worktree_branches(repo: Path) -> list[str]:
+    """Branches checked out in every currently-live worktree of the repo
+    containing `repo` -- "live" meaning its filesystem path still exists and
+    it isn't a bare checkout. `git worktree list` reports every worktree of
+    a repo regardless of which one it's run from, so this works whether
+    `repo` is the main checkout or a linked worktree.
+
+    A detached-HEAD worktree (no `branch ...` porcelain line) is reported as
+    branch "HEAD" -- matching what `current_branch()` (and so
+    `workspace_db_name()`) computes for that same worktree via `git
+    rev-parse --abbrev-ref HEAD`. Without this, a detached-HEAD worktree
+    (e.g. a CI checkout, which defaults to one) would never appear "live"
+    here, and its database would look orphaned and get dropped out from
+    under it."""
+    result = subprocess.run(
+        ["git", "worktree", "list", "--porcelain"],
+        cwd=repo,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+
+    branches: list[str] = []
+    path: Path | None = None
+    branch: str | None = None
+    bare = False
+
+    def _flush() -> None:
+        if path is not None and not bare and branch is not None and path.exists():
+            branches.append(branch)
+
+    for line in result.stdout.splitlines():
+        if not line:
+            _flush()
+            path, branch, bare = None, None, False
+        elif line.startswith("worktree "):
+            path = Path(line[len("worktree ") :])
+        elif line.startswith("branch "):
+            branch = line[len("branch ") :].removeprefix("refs/heads/")
+        elif line == "detached":
+            branch = "HEAD"
+        elif line == "bare":
+            bare = True
+    _flush()
+
+    return branches
+
+
+def escape_like_prefix(prefix: str) -> str:
+    """Escape a literal string for use as a `LIKE ... ESCAPE '\\'` prefix
+    pattern (with a trailing `%` the caller adds), so a project name/branch
+    containing `%` or `_` can't widen the match to an unrelated database."""
+    return prefix.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+
+
+def expected_db_names(config: "ProjectConfig", branches: list[str]) -> set[str]:
+    """Every DB name a currently live worktree of this project is entitled
+    to own: the main workspace DB per branch, plus one `<main>{suffix}`
+    sibling per `[tool.pgdevkit].extra_db_suffixes` entry."""
+    names: set[str] = set()
+    for branch in branches:
+        main_db = workspace_db_name(config.name, branch)
+        names.add(main_db)
+        names.update(f"{main_db}{suffix}" for suffix in config.extra_db_suffixes)
+    return names

pyproject.toml

@@ -11,7 +11,7 @@ packages = ["pgdevkit"]
 
 [project]
 name = "pgdevkit"
-version = "0.6.2"
+version = "0.7.0"
 description = "A helper for developing with Postgres"
 readme = "README.md"
 requires-python = ">=3.14"