Fix migrate apply --ask silently skipping every accepted/already-done migration

pgdevkit/cli.py

@@ -371,24 +371,36 @@ def worker() -> None:
             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_step()
-                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_step()
-                continue
+            already_done = migrate.already_fully_applied(conninfo, path)
+            if already_done:
+                bar_write(f"Auto: {path.name} is already fully present in the database — marking as already done")
+            else:
+                # Hold the lock across the whole show-file+prompt step (not just each write) so
+                # the worker's bar_write() calls block until the question is answered instead
+                # of interleaving with the prompt or the migration text while it's being read.
+                with bar_lock:
+                    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_step()
+                    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_step()
+                    continue
 
         work_q.put((path, already_done))
 

pgdevkit/migrate.py

@@ -197,6 +197,83 @@ def _created_table_names(stmts: list[str]) -> list[str]:
     return names
 
 
+_ADD_COLUMN_RE = re.compile(
+    rf"ALTER\s+TABLE\s+(?:IF\s+EXISTS\s+)?([\w.\"]+)\s+ADD\s+COLUMN\s+"
+    rf"(?:IF\s+NOT\s+EXISTS\s+)?\"?({_IDENTIFIER})\"?",
+    re.IGNORECASE,
+)
+_CREATE_RELATION_RE = re.compile(
+    r"CREATE\s+(?:UNIQUE\s+)?INDEX\s+(?:CONCURRENTLY\s+)?(?:IF\s+NOT\s+EXISTS\s+)?([\w.\"]+)"
+    r"|CREATE\s+SEQUENCE\s+(?:IF\s+NOT\s+EXISTS\s+)?([\w.\"]+)"
+    r"|CREATE\s+(?:MATERIALIZED\s+)?VIEW\s+(?:IF\s+NOT\s+EXISTS\s+)?([\w.\"]+)",
+    re.IGNORECASE,
+)
+_CREATE_SCHEMA_RE = re.compile(r"CREATE\s+SCHEMA\s+(?:IF\s+NOT\s+EXISTS\s+)?([\w\"]+)", re.IGNORECASE)
+
+
+def _idempotent_target(stmt: str) -> tuple[str, ...] | None:
+    """Classify a single statement as one of the create-if-missing DDL shapes this module
+    can check for "already applied" with no ambiguity: ("relation", name) for a table,
+    index, sequence or view; ("schema", name); or ("column", table, column) for an ADD
+    COLUMN. None if the statement isn't one of these shapes — including any CREATE OR
+    REPLACE, which is never safe to treat as a no-op just because the object exists, since
+    the migration could be replacing it with different content."""
+    stripped = _strip_line_comments(stmt).strip()
+    if re.search(r"\bOR\s+REPLACE\b", stripped, re.IGNORECASE):
+        return None
+
+    if re.match(r"CREATE\s+TABLE\b", stripped, re.IGNORECASE):
+        names = _created_table_names([stmt])
+        return ("relation", names[0]) if names else None
+
+    m = _CREATE_RELATION_RE.match(stripped)
+    if m:
+        name = next(g for g in m.groups() if g is not None)
+        return ("relation", name)
+
+    m = _CREATE_SCHEMA_RE.match(stripped)
+    if m:
+        return ("schema", m.group(1))
+
+    m = _ADD_COLUMN_RE.match(stripped)
+    if m:
+        return ("column", m.group(1), m.group(2))
+
+    return None
+
+
+def _target_exists(con: psycopg.Connection, target: tuple[str, ...]) -> bool:
+    kind = target[0]
+    if kind == "relation":
+        row = con.execute("select to_regclass(%s)", (target[1],)).fetchone()
+    elif kind == "schema":
+        row = con.execute("select to_regnamespace(%s)", (target[1],)).fetchone()
+    else:  # column
+        row = con.execute(
+            "select 1 from pg_attribute where attrelid = to_regclass(%s) "
+            "and attname = %s and not attisdropped",
+            (target[1], target[2]),
+        ).fetchone()
+    return bool(row and row[0])
+
+
+def already_fully_applied(conninfo: str, path: Path) -> bool:
+    """Whether every statement in this migration is a recognized create-if-missing shape
+    (table/index/sequence/view/schema, or add-column) AND its target already exists in the
+    database — i.e. re-running the migration would do nothing. Used by `--ask` to
+    auto-answer "already done" without prompting, so migrations that are trivially no-ops
+    don't interrupt review. A single unrecognized or not-yet-applied statement means
+    False — this never guesses."""
+    stmts = _split_sql(path.read_text(encoding="utf-8"))
+    if not stmts:
+        return False
+    targets = [_idempotent_target(s) for s in stmts]
+    if any(t is None for t in targets):
+        return False
+    with psycopg.connect(conninfo) as con:
+        return all(_target_exists(con, cast(tuple[str, ...], t)) for t in targets)
+
+
 def list_migration_files(migrations_dir: Path) -> list[Path]:
     return sorted(migrations_dir.glob("*.sql"))
 

pyproject.toml

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

tests/test_migrate.py

@@ -1,6 +1,12 @@
 from __future__ import annotations
 
-from pgdevkit.migrate import _created_table_names, _split_sql, _strip_line_comments, default_tracking_table
+from pgdevkit.migrate import (
+    _created_table_names,
+    _idempotent_target,
+    _split_sql,
+    _strip_line_comments,
+    default_tracking_table,
+)
 
 
 def test_created_table_names_ignores_create_table_mentioned_in_a_comment():
@@ -74,3 +80,55 @@ def test_default_tracking_table_searches_parent_directories(tmp_path):
 
 def test_default_tracking_table_falls_back_with_no_pyproject_at_all(tmp_path):
     assert default_tracking_table(tmp_path / "nonexistent") == "public.schema_migrations"
+
+
+def test_idempotent_target_recognizes_create_table_if_not_exists():
+    assert _idempotent_target("CREATE TABLE IF NOT EXISTS app.widgets (id int)") == ("relation", "app.widgets")
+
+
+def test_idempotent_target_recognizes_create_index():
+    assert _idempotent_target(
+        "CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS widgets_name_idx ON app.widgets (name)"
+    ) == ("relation", "widgets_name_idx")
+
+
+def test_idempotent_target_recognizes_create_sequence():
+    assert _idempotent_target("CREATE SEQUENCE IF NOT EXISTS app.widgets_seq") == ("relation", "app.widgets_seq")
+
+
+def test_idempotent_target_recognizes_create_view():
+    assert _idempotent_target("CREATE VIEW IF NOT EXISTS app.widgets_v AS SELECT * FROM app.widgets") == (
+        "relation",
+        "app.widgets_v",
+    )
+
+
+def test_idempotent_target_recognizes_create_schema():
+    assert _idempotent_target("CREATE SCHEMA IF NOT EXISTS app") == ("schema", "app")
+
+
+def test_idempotent_target_recognizes_add_column():
+    assert _idempotent_target("ALTER TABLE app.widgets ADD COLUMN IF NOT EXISTS name text") == (
+        "column",
+        "app.widgets",
+        "name",
+    )
+
+
+def test_idempotent_target_recognizes_add_column_without_if_not_exists_guard():
+    assert _idempotent_target("ALTER TABLE app.widgets ADD COLUMN name text") == (
+        "column",
+        "app.widgets",
+        "name",
+    )
+
+
+def test_idempotent_target_ignores_create_or_replace():
+    # A view/function could be replaced with different content, so existence alone never
+    # means "already applied" for these.
+    assert _idempotent_target("CREATE OR REPLACE VIEW app.widgets_v AS SELECT 1") is None
+
+
+def test_idempotent_target_ignores_unrecognized_statements():
+    assert _idempotent_target("ALTER TABLE app.widgets DROP COLUMN name") is None
+    assert _idempotent_target("INSERT INTO app.widgets (id) VALUES (1)") is None