feat: switch from AST parsing to alembic upgrade --sql

Replace extract_sql/AST extraction with generate_sql that calls
alembic upgrade --sql for complete DDL generation. This gives squawk
visibility into ORM operations like op.create_index and op.create_table
that were previously invisible.

Extend autocommit checker to flag op.create_index/op.drop_index with
postgresql_concurrently=True outside autocommit_block.

Author: stephen-kintsugi

README.md

@@ -2,7 +2,9 @@
 
 A [pre-commit](https://pre-commit.com/) hook that lints SQL in [Alembic](https://alembic.sqlalchemy.org/) migrations using [squawk](https://squawkhq.com/), a PostgreSQL migration linter.
 
-Squawk operates on raw SQL files, but Alembic migrations are Python. This hook bridges the gap by extracting SQL from `op.execute()` calls in migration files and passing it to squawk for analysis.
+Squawk operates on raw SQL files, but Alembic migrations are Python. This hook bridges the gap by generating DDL via `alembic upgrade --sql` (offline mode) and passing the complete SQL output to squawk for analysis. This captures all SQL statements a migration produces, including ORM operations like `op.create_index()`, `op.create_table()`, and `op.alter_column()`.
+
+The hook also checks that concurrent index operations (`CONCURRENTLY` in `op.execute()` or `postgresql_concurrently=True` in `op.create_index()` / `op.drop_index()`) are wrapped in `autocommit_block()`.
 
 ## Usage
 
@@ -11,25 +13,24 @@ Add the following to your `.pre-commit-config.yaml`:
 ```yaml
 repos:
     - repo: https://github.com/kintsugi-tax/kintsugi-squawk
-      rev: v0.1.0
+      rev: v0.2.0
       hooks:
           - id: squawk-alembic
 ```
 
-No additional configuration is required. The hook auto-detects your migrations directory by reading `script_location` from `alembic.ini`.
+No additional configuration is required. The hook auto-detects your migrations directory by reading `script_location` from `alembic.ini`. The consumer's `alembic` must be available on PATH (the hook calls it via subprocess).
 
 ## How It Works
 
 When pre-commit runs, the hook:
 
 1. Parses `alembic.ini` to find the migrations `versions/` directory
 2. Filters staged files to only those under that directory
-3. Extracts SQL strings from `op.execute()` calls using Python's AST parser
-4. Pipes the extracted SQL to squawk for linting
-
-The hook handles common patterns including `op.execute("...")`, `op.execute(sa.text("..."))`, triple-quoted strings, and implicit string concatenation.
+3. Checks for concurrent operations outside `autocommit_block()`
+4. Runs `alembic upgrade --sql` to generate the complete DDL for each migration
+5. Pipes the generated SQL to squawk for linting
 
-ORM-level operations like `op.add_column()` and `op.create_table()` are not linted, since they don't contain raw SQL. These produce safe, predictable DDL that squawk is less likely to flag.
+Merge migrations (where `down_revision` is a tuple) are skipped since they produce no DDL.
 
 ## Squawk Configuration
 
@@ -41,6 +42,7 @@ Squawk reads its configuration from `.squawk.toml` in the consumer repo root. Se
 
 * Python (version 3.12)
 * Poetry
+* squawk-cli (`pip install squawk-cli`)
 
 **Steps:**
 

poetry.lock

@@ -13,6 +13,7 @@ python = ">=3.10"
 squawk-cli = "*"
 
 [tool.poetry.group.dev.dependencies]
+alembic = "*"
 pre-commit = "*"
 pytest = "*"
 ruff = "*"

pyproject.toml

@@ -1,7 +1,8 @@
-"""Pre-commit hook that extracts SQL from Alembic migrations and lints with squawk."""
+"""Pre-commit hook that generates DDL via alembic upgrade --sql and lints with squawk."""
 
 import ast
 import configparser
+import os
 import subprocess
 import sys
 import tempfile
@@ -31,50 +32,98 @@ def find_migrations_path():
     return None
 
 
-def extract_sql(filepath):
-    """Parse a migration file and extract SQL strings from op.execute() calls."""
+class RevisionInfo:
+    __slots__ = ("revision", "down_revision", "is_merge")
+
+    def __init__(self, revision, down_revision, is_merge):
+        self.revision = revision
+        self.down_revision = down_revision
+        self.is_merge = is_merge
+
+
+def extract_revision_info(filepath):
+    """Parse a migration file to extract revision and down_revision from module-level assignments."""
     with open(filepath) as f:
         try:
             tree = ast.parse(f.read())
         except SyntaxError:
-            return []
+            return None
 
-    statements = []
+    revision = None
+    down_revision = None
 
-    for node in ast.walk(tree):
-        if not isinstance(node, ast.Call):
+    for node in ast.iter_child_nodes(tree):
+        if not isinstance(node, ast.Assign):
             continue
-
-        if not (
-            isinstance(node.func, ast.Attribute)
-            and node.func.attr == "execute"
-            and isinstance(node.func.value, ast.Name)
-            and node.func.value.id == "op"
-            and node.args
-        ):
+        if len(node.targets) != 1 or not isinstance(node.targets[0], ast.Name):
             continue
 
-        sql = _extract_string(node.args[0])
-        if sql:
-            statements.append(sql)
+        name = node.targets[0].id
+        if name == "revision":
+            if isinstance(node.value, ast.Constant) and isinstance(
+                node.value.value, str
+            ):
+                revision = node.value.value
+        elif name == "down_revision":
+            if isinstance(node.value, ast.Constant):
+                if isinstance(node.value.value, str):
+                    down_revision = node.value.value
+                elif node.value.value is None:
+                    down_revision = None
+            elif isinstance(node.value, ast.Tuple):
+                values = []
+                for elt in node.value.elts:
+                    if isinstance(elt, ast.Constant) and isinstance(elt.value, str):
+                        values.append(elt.value)
+                down_revision = tuple(values)
+
+    if revision is None:
+        return None
 
-    return statements
+    is_merge = isinstance(down_revision, tuple)
+    return RevisionInfo(
+        revision=revision, down_revision=down_revision, is_merge=is_merge
+    )
 
 
-def _extract_string(node):
-    """Extract a string value from an AST node, handling common wrappers."""
-    # Direct string literal: op.execute("SQL")
-    if isinstance(node, ast.Constant) and isinstance(node.value, str):
-        return node.value
+def generate_sql(filepath):
+    """Run alembic upgrade --sql to generate the complete DDL for a migration."""
+    info = extract_revision_info(filepath)
+    if info is None:
+        return None
 
-    # sa.text("SQL") or text("SQL")
-    if isinstance(node, ast.Call) and node.args:
-        if isinstance(node.func, ast.Attribute) and node.func.attr == "text":
-            return _extract_string(node.args[0])
-        if isinstance(node.func, ast.Name) and node.func.id == "text":
-            return _extract_string(node.args[0])
+    if info.is_merge:
+        return None
 
-    return None
+    base = info.down_revision if info.down_revision else "base"
+    target = f"{base}:{info.revision}"
+
+    env = os.environ.copy()
+    if "DATABASE_URL" not in env:
+        env["DATABASE_URL"] = "postgresql://localhost/lint"
+
+    try:
+        result = subprocess.run(
+            ["alembic", "upgrade", target, "--sql"],
+            capture_output=True,
+            text=True,
+            env=env,
+        )
+    except FileNotFoundError:
+        print(
+            "squawk-alembic: alembic not found. Ensure alembic is installed in your environment.",
+            file=sys.stderr,
+        )
+        return None
+
+    if result.returncode != 0:
+        print(
+            f"squawk-alembic: alembic upgrade --sql failed for {filepath}:\n{result.stderr}",
+            file=sys.stderr,
+        )
+        return None
+
+    return result.stdout
 
 
 def check_autocommit_blocks(filepath):
@@ -90,6 +139,18 @@ def check_autocommit_blocks(filepath):
     return checker.warnings
 
 
+def _has_concurrent_kwarg(node):
+    """Check if an AST Call node has postgresql_concurrently=True."""
+    for kw in node.keywords:
+        if (
+            kw.arg == "postgresql_concurrently"
+            and isinstance(kw.value, ast.Constant)
+            and kw.value.value is True
+        ):
+            return True
+    return False
+
+
 class _AutocommitChecker(ast.NodeVisitor):
     def __init__(self):
         self.warnings = []
@@ -113,17 +174,38 @@ def visit_With(self, node):
     def visit_Call(self, node):
         if (
             isinstance(node.func, ast.Attribute)
-            and node.func.attr == "execute"
             and isinstance(node.func.value, ast.Name)
             and node.func.value.id == "op"
-            and node.args
         ):
-            sql = _extract_string(node.args[0])
-            if sql and "concurrently" in sql.lower() and not self._in_autocommit:
-                self.warnings.append(node.lineno)
+            # op.execute("...CONCURRENTLY...")
+            if node.func.attr == "execute" and node.args:
+                sql = _extract_string(node.args[0])
+                if sql and "concurrently" in sql.lower() and not self._in_autocommit:
+                    self.warnings.append(node.lineno)
+
+            # op.create_index(..., postgresql_concurrently=True)
+            # op.drop_index(..., postgresql_concurrently=True)
+            if node.func.attr in ("create_index", "drop_index"):
+                if _has_concurrent_kwarg(node) and not self._in_autocommit:
+                    self.warnings.append(node.lineno)
+
         self.generic_visit(node)
 
 
+def _extract_string(node):
+    """Extract a string value from an AST node, handling common wrappers."""
+    if isinstance(node, ast.Constant) and isinstance(node.value, str):
+        return node.value
+
+    if isinstance(node, ast.Call) and node.args:
+        if isinstance(node.func, ast.Attribute) and node.func.attr == "text":
+            return _extract_string(node.args[0])
+        if isinstance(node.func, ast.Name) and node.func.id == "text":
+            return _extract_string(node.args[0])
+
+    return None
+
+
 def main():
     files = sys.argv[1:]
     if not files:
@@ -152,14 +234,12 @@ def main():
             )
             exit_code = 1
 
-        statements = extract_sql(filepath)
-        if not statements:
+        sql = generate_sql(filepath)
+        if not sql:
             continue
 
-        combined = "\n".join(statements)
-
         with tempfile.NamedTemporaryFile(mode="w", suffix=".sql", delete=False) as tmp:
-            tmp.write(combined)
+            tmp.write(sql)
             tmp_path = tmp.name
 
         try: