feat(cli): project hook sqlpush.py — dsn/metadata/chain-dir defaults with flag > hook > env precedence

CHANGELOG.md

@@ -6,6 +6,24 @@ the project follows [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Added
+
+- Project hook `sqlpush.py`: any verb run with a `sqlpush.py` in the
+  CWD discovers it (the alembic `env.py` / pytest `conftest.py`
+  pattern; loaded by path, never by module name) and uses it for
+  defaults — `get_dsn()` for `--dsn`/`--ref-dsn`, `get_metadata()`
+  for the `module:attribute` positional (now optional on
+  `diff`/`check`/`push`/`revision`), `CHAIN_DIR` for `--dir`.
+  Precedence is explicit flag > hook > `$DATABASE_URL`/current
+  default; a hook that is missing a needed member, or whose
+  `get_dsn()`/`get_metadata()` raises, fails as a typed
+  `SqlpushError` naming the file and the member (exit 1, no
+  traceback). The CWD is appended — never prepended — to `sys.path`,
+  so the hook file can never shadow the installed `sqlpush` package.
+  Without a hook every verb behaves exactly as before: `uv run
+  sqlpush revision -m "change"` from the project root now needs no
+  `--dsn`, no `module:attribute` and no PYTHONPATH.
+
 ## [0.5.1] - 2026-09-02
 
 ### Added

README.md

@@ -167,6 +167,51 @@ run under `psql`. Schema change and data backfill ship as one file.
 The [chain guide](docs/the-chain.md) covers the format, the gates and
 the workflows.
 
+## Project hook
+
+Drop a `sqlpush.py` in your repo root (the alembic `env.py` /
+pytest `conftest.py` pattern) and the CLI stops needing flags,
+specs and env vars:
+
+```python
+# sqlpush.py — in your repo root
+def get_metadata():  # REQUIRED for diff/check/push/revision
+    from myapp.models import metadata
+
+    return metadata  # -> a populated MetaData
+
+
+def get_dsn() -> str:  # REQUIRED for every verb
+    from myapp.settings import DATABASE_URL
+
+    return DATABASE_URL  # -> a full psycopg DSN
+
+
+CHAIN_DIR = "migrations/chain"  # OPTIONAL — default for --dir
+```
+
+With that file in place, `uv run sqlpush revision -m "change"` just
+works: no `--dsn`, no `module:attribute`, no credentials on the
+command line, no PYTHONPATH. Inputs resolve with a fixed precedence:
+
+| input | explicit flag | `sqlpush.py` | fallback |
+| --- | --- | --- | --- |
+| `--dsn` / `--ref-dsn` | wins | `get_dsn()` | `$DATABASE_URL`, only without a hook and never for `--ref-dsn` |
+| `module:attribute` (diff/check/push/revision) | wins | `get_metadata()` | usage error |
+| `--dir` (revision/migrate/stamp) | wins | `CHAIN_DIR` | `migrations/versions` |
+
+A hook that is missing a member a verb needs — or whose
+`get_dsn()`/`get_metadata()` raises — fails with a typed error naming
+the file and the member (`sqlpush.py: missing get_dsn()`), never a
+traceback. Without a `sqlpush.py`, every verb behaves exactly as
+before.
+
+One deliberate detail: on discovering the hook, sqlpush **appends**
+your CWD to `sys.path` instead of prepending it. A file named
+`sqlpush.py` would otherwise shadow the installed package the moment
+you ran the CLI from your repo root; appending guarantees the real
+package always wins, and the hook is loaded by path only.
+
 ## Exit codes
 
 | verb | 0 | 1 | 2 | 3 |
@@ -194,13 +239,14 @@ The knobs, per verb:
 | verb | flags |
 | --- | --- |
 | `push` | `--allow-destructive` `--safe-only` `--no-lock` `--lock-timeout` `--advisory-wait` `--no-concurrently` `--statement-timeout` |
-| `revision` | `--ref-dsn` (required) `-m/--message` `--no-concurrently` `--dir` |
+| `revision` | `--ref-dsn` (required without a hook) `-m/--message` `--no-concurrently` `--dir` |
 | `migrate` | `--allow-destructive` `--advisory-wait` `--lock-timeout` `--statement-timeout` `--dir` |
 | `stamp` | `--force` `--dir` |
 
-Every verb except `revision` takes `--dsn` (or `$DATABASE_URL`).
-`revision` requires `--ref-dsn`, with no env fallback: the reference
-DB is a different database from the push target. `diff`, `check`,
+Every verb except `revision` takes `--dsn` (or `$DATABASE_URL`, or
+the project hook's `get_dsn()`). `revision` takes `--ref-dsn` — or
+the hook — with no env fallback: the reference DB is a different
+database from the push target. `diff`, `check`,
 `push` and `revision` also take repeatable `--schema` / `--exclude`.
 Timeouts are seconds; a `lock_timeout` bounds how long a statement
 waits on a lock before failing, `statement_timeout` bounds each

src/sqlpush/cli.py

@@ -24,6 +24,7 @@
 
 from sqlpush import api
 from sqlpush.core.render import render
+from sqlpush.hook import hook_chain_dir, hook_dsn, hook_metadata, load_project_hook
 from sqlpush.types import (
     CheckResult,
     MetadataImportError,
@@ -38,8 +39,10 @@
 SchemaOpt = Annotated[list[str] | None, typer.Option("--schema")]
 ExcludeOpt = Annotated[list[str] | None, typer.Option("--exclude")]
 # Path-typed options need the Annotated form: B008 (call in default) only
-# exempts typer.Option for non-Path annotations
-DirOpt = Annotated[Path, typer.Option("--dir")]
+# exempts typer.Option for non-Path annotations. None default: lets the
+# verb tell "user passed --dir" apart from "use hook/default" resolution.
+DirOpt = Annotated[Path | None, typer.Option("--dir")]
+DEFAULT_CHAIN_DIR = Path("migrations/versions")
 
 
 def _load_metadata(spec: str):
@@ -53,7 +56,28 @@ def _load_metadata(spec: str):
         raise MetadataImportError(f"cannot import {spec!r}: {exc}") from exc
 
 
-def _engine(dsn: str | None):
+def _metadata(metadata_spec: str | None, hook):
+    """Resolve the metadata source: flag > hook.get_metadata() > usage error.
+
+    The hook path returns the OBJECT itself (no module:attribute import
+    happens); the positional path is untouched.
+    """
+    if metadata_spec is not None:
+        return _load_metadata(metadata_spec)
+    if hook is not None:
+        return hook_metadata(hook)
+    typer.secho(
+        "metadata spec required: pass module:attribute or add get_metadata() to ./sqlpush.py",
+        fg="red",
+        err=True,
+    )
+    raise typer.Exit(code=2)
+
+
+def _engine(dsn: str | None, hook=None):
+    # precedence: explicit flag > hook.get_dsn() > $DATABASE_URL > error
+    if dsn is None and hook is not None:
+        dsn = hook_dsn(hook)
     dsn = dsn or os.environ.get("DATABASE_URL")
     if not dsn:
         typer.secho("no --dsn and no DATABASE_URL set", fg="red", err=True)
@@ -66,6 +90,13 @@ def _engine(dsn: str | None):
         raise typer.Exit(code=1) from exc
 
 
+def _resolve_dir(out_dir: Path | None, hook) -> Path:
+    """Resolve --dir: explicit flag > hook.CHAIN_DIR > current default."""
+    if out_dir is not None:
+        return out_dir
+    return Path(hook_chain_dir(hook, str(DEFAULT_CHAIN_DIR)))
+
+
 def _emit_json(plan) -> None:
     typer.echo(json.dumps(plan.to_json_dict(), indent=2))
 
@@ -81,16 +112,17 @@ def _risk_summary(p) -> None:
 
 @app.command()
 def diff(
-    metadata_spec: str = typer.Argument(..., help="module:metadata"),
+    metadata_spec: str | None = typer.Argument(None, help="module:metadata"),
     dsn: str | None = typer.Option(None),
     json_output: bool = typer.Option(False, "--json"),
     verbose: bool = typer.Option(False, "--verbose"),
     quiet: bool = typer.Option(False, "--quiet"),
     schema: SchemaOpt = None,
     exclude: ExcludeOpt = None,
 ):
-    md = _load_metadata(metadata_spec)
-    engine = _engine(dsn)
+    hook = load_project_hook()
+    md = _metadata(metadata_spec, hook)
+    engine = _engine(dsn, hook)
     try:
         p = api.plan(md, engine, schemas=schema, exclude=exclude or ())
         if json_output:
@@ -106,16 +138,17 @@ def diff(
 
 @app.command()
 def check(
-    metadata_spec: str = typer.Argument(...),
+    metadata_spec: str | None = typer.Argument(None),
     dsn: str | None = typer.Option(None),
     json_output: bool = typer.Option(False, "--json"),
     verbose: bool = typer.Option(False, "--verbose"),
     quiet: bool = typer.Option(False, "--quiet"),
     schema: SchemaOpt = None,
     exclude: ExcludeOpt = None,
 ):
-    md = _load_metadata(metadata_spec)
-    engine = _engine(dsn)
+    hook = load_project_hook()
+    md = _metadata(metadata_spec, hook)
+    engine = _engine(dsn, hook)
     try:
         # plan ONCE and derive everything from that single object: a
         # second plan could observe a different DB state than the one
@@ -140,7 +173,7 @@ def check(
 
 @app.command()
 def push(
-    metadata_spec: str = typer.Argument(...),
+    metadata_spec: str | None = typer.Argument(None),
     dsn: str | None = typer.Option(None),
     json_output: bool = typer.Option(False, "--json"),
     allow_destructive: bool = typer.Option(False, "--allow-destructive"),
@@ -155,8 +188,9 @@ def push(
     schema: SchemaOpt = None,
     exclude: ExcludeOpt = None,
 ):
-    md = _load_metadata(metadata_spec)
-    engine = _engine(dsn)
+    hook = load_project_hook()
+    md = _metadata(metadata_spec, hook)
+    engine = _engine(dsn, hook)
     try:
         try:
             report = api.push(
@@ -223,25 +257,35 @@ def push(
 
 @app.command()
 def revision(
-    metadata_spec: str = typer.Argument(..., help="module:metadata"),
-    ref_dsn: str = typer.Option(..., "--ref-dsn"),
+    metadata_spec: str | None = typer.Argument(None, help="module:metadata"),
+    ref_dsn: str | None = typer.Option(None, "--ref-dsn"),
     message: str | None = typer.Option(None, "--message", "-m"),
-    out_dir: DirOpt = Path("migrations/versions"),
+    out_dir: DirOpt = None,
     no_concurrently: bool = typer.Option(False, "--no-concurrently"),
     schema: SchemaOpt = None,
     exclude: ExcludeOpt = None,
 ):
     """Generate the next migration file from models vs the reference DB."""
-    md = _load_metadata(metadata_spec)
-    # required --ref-dsn (no DATABASE_URL fallback): the reference DB is a
-    # different database from the push target — conflating them silently
-    # would chain against the wrong head
+    hook = load_project_hook()
+    md = _metadata(metadata_spec, hook)
+    # --ref-dsn: flag > hook.get_dsn(); NO DATABASE_URL fallback (the
+    # reference DB is a different database from the push target —
+    # conflating them silently would chain against the wrong head)
+    if ref_dsn is None and hook is not None:
+        ref_dsn = hook_dsn(hook)
+    if ref_dsn is None:
+        typer.secho(
+            "--ref-dsn is required (or add get_dsn() to ./sqlpush.py)",
+            fg="red",
+            err=True,
+        )
+        raise typer.Exit(code=2)
     engine = _engine(ref_dsn)
     try:
         path = api.revision(
             md,
             engine,
-            out_dir=out_dir,
+            out_dir=_resolve_dir(out_dir, hook),
             message=message,
             concurrently=not no_concurrently,
             schemas=schema,
@@ -260,14 +304,15 @@ def migrate(
     advisory_wait: float = typer.Option(30.0, "--advisory-wait"),
     lock_timeout: float = typer.Option(5.0, "--lock-timeout"),
     statement_timeout: float | None = typer.Option(None, "--statement-timeout"),
-    out_dir: DirOpt = Path("migrations/versions"),
+    out_dir: DirOpt = None,
 ):
     """Replay pending migration files (gates + checksum bookkeeping)."""
-    engine = _engine(dsn)
+    hook = load_project_hook()
+    engine = _engine(dsn, hook)
     try:
         report = api.migrate(
             engine,
-            chain_dir=out_dir,
+            chain_dir=_resolve_dir(out_dir, hook),
             allow_destructive=allow_destructive,
             advisory_wait=advisory_wait,
             lock_timeout=lock_timeout,
@@ -290,12 +335,13 @@ def migrate(
 def stamp(
     dsn: str | None = typer.Option(None),
     force: bool = typer.Option(False, "--force"),
-    out_dir: DirOpt = Path("migrations/versions"),
+    out_dir: DirOpt = None,
 ):
     """Adopt an existing DB: register chain files without executing SQL."""
-    engine = _engine(dsn)
+    hook = load_project_hook()
+    engine = _engine(dsn, hook)
     try:
-        report = api.stamp(engine, chain_dir=out_dir, force=force)
+        report = api.stamp(engine, chain_dir=_resolve_dir(out_dir, hook), force=force)
     finally:
         engine.dispose()
     typer.echo(

src/sqlpush/hook.py

@@ -0,0 +1,84 @@
+# src/sqlpush/hook.py
+"""Project hook: ``sqlpush.py`` in the CWD.
+
+The alembic ``env.py`` / pytest ``conftest.py`` pattern: when a
+``sqlpush.py`` exists next to where the user runs the CLI, it becomes
+the source of defaults for metadata, DSN and chain dir. Explicit flags
+always win (flag > hook > env/default); without the hook every verb
+behaves exactly as before.
+
+Loading is BY PATH (``spec_from_file_location``), never by module
+name: a file named ``sqlpush.py`` in the CWD would shadow the
+installed package if the CWD came first on ``sys.path`` — so the CWD
+is APPENDED (never ``insert(0)``), guaranteeing the package wins for
+normal imports while the hook itself is only ever loaded explicitly.
+
+Only typed :class:`SqlpushError`-family errors escape this module,
+always naming the file and the member involved.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+from pathlib import Path
+from types import ModuleType
+
+from sqlpush.types import SqlpushError
+
+HOOK_FILENAME = "sqlpush.py"
+
+
+class HookError(SqlpushError):
+    """The project hook is broken or incomplete — names file + member."""
+
+
+def load_project_hook() -> ModuleType | None:
+    """Load ``./sqlpush.py`` by path if present; ``None`` when absent.
+
+    On success the CWD is appended to ``sys.path`` (never inserted at
+    the front — see the module docstring for the shadowing rationale).
+    Any import-time failure is re-typed as :class:`HookError`.
+    """
+    path = Path.cwd() / HOOK_FILENAME
+    if not path.is_file():
+        return None
+    sys.path.append(os.getcwd())
+    spec = importlib.util.spec_from_file_location("sqlpush_project_hook", path)
+    if spec is None or spec.loader is None:
+        raise HookError(f"{HOOK_FILENAME}: cannot load (invalid module spec)")
+    module = importlib.util.module_from_spec(spec)
+    try:
+        spec.loader.exec_module(module)
+    except Exception as exc:
+        raise HookError(f"{HOOK_FILENAME}: import raised: {exc}") from exc
+    return module
+
+
+def hook_dsn(hook: ModuleType) -> str:
+    """Call ``get_dsn()`` LAZILY — only when a verb actually needs it."""
+    getter = getattr(hook, "get_dsn", None)
+    if getter is None:
+        raise HookError(f"{HOOK_FILENAME}: missing get_dsn()")
+    try:
+        return getter()
+    except Exception as exc:
+        raise HookError(f"{HOOK_FILENAME}: get_dsn() raised: {exc}") from exc
+
+
+def hook_metadata(hook: ModuleType):
+    """Call ``get_metadata()`` LAZILY — returns the populated MetaData object."""
+    getter = getattr(hook, "get_metadata", None)
+    if getter is None:
+        raise HookError(f"{HOOK_FILENAME}: missing get_metadata()")
+    try:
+        return getter()
+    except Exception as exc:
+        raise HookError(f"{HOOK_FILENAME}: get_metadata() raised: {exc}") from exc
+
+
+def hook_chain_dir(hook: ModuleType | None, default: str) -> str:
+    """Read ``CHAIN_DIR`` as a module attribute; ``default`` when unset."""
+    value = getattr(hook, "CHAIN_DIR", None) if hook is not None else None
+    return default if value is None else value