Enforce SQLAlchemy-only database connections and Alembic-only schema changes

AGENTS.md

@@ -18,8 +18,8 @@ cutover plans, generated migration docs, or migration guard scripts, read
 When adding, moving, or reviewing tests, read
 `docs/engineering/skills/testing.md`.
 
-When changing SQLAlchemy models, Alembic configuration, database schemas, or
-migration revisions, read
+When changing SQLAlchemy models, Alembic configuration, database schemas,
+migration revisions, or code that connects to a database, read
 `docs/engineering/skills/alembic-migrations.md`.
 
 When adding or moving API v2 route, service, or database-access modules, read
@@ -83,7 +83,7 @@ To add or change a rule, read `docs/engineering/skills/rules-registry.md`.
 
 ### Runtime persistence and route boundaries
 
-- Checked: `api.data.runtime-sql-confined`, `api.legacy.removed-layers-stay-removed`, `api.routes.flask-blueprints-own-urls`, `api.routes.no-domain-logic-in-http`, `api.routes.no-inline-json-errors`, `api.services.no-direct-queries`, `api.services.session-ownership`
+- Checked: `api.data.runtime-sql-confined`, `api.data.sqlalchemy-and-alembic-only`, `api.legacy.removed-layers-stay-removed`, `api.routes.flask-blueprints-own-urls`, `api.routes.no-domain-logic-in-http`, `api.routes.no-inline-json-errors`, `api.services.no-direct-queries`, `api.services.session-ownership`
 
 ### Import and startup side effects
 

changelog.d/sqlalchemy-alembic-database-rule.added.md

@@ -0,0 +1 @@
+Added an enforced rule (api.data.sqlalchemy-and-alembic-only): code connects to relational databases only through SQLAlchemy engines, connections and sessions, and changes persistent schema only through Alembic revisions. A unit test parses non-test code and fails on direct database-driver imports, driver-connection access, create_all, drop_all, table create or drop calls, Alembic operations outside migrations/, and schema-changing SQL text, allowing only the COPY step and the temporary staging tables in catalog publication staging. Application behaviour is unchanged.

docs/engineering/skills/alembic-migrations.md

@@ -1,7 +1,8 @@
 # Alembic Migrations
 
 Use this skill whenever adding or changing SQLAlchemy models, Alembic
-configuration, database schema, or migration revisions.
+configuration, database schema, migration revisions, or code that connects to
+a database.
 
 ## Mandatory generation rule
 
@@ -40,13 +41,95 @@ Before committing a migration, run these checks against the matching domain:
 2. Upgrade a fresh database to `head`.
 3. Compare the database to the complete ORM metadata before release.
 4. Downgrade one revision and upgrade to `head` again in an isolated database.
-5. Confirm application startup performs no implicit DDL.
+5. Confirm application startup performs no implicit DDL
+   (`api.data.sqlalchemy-and-alembic-only` checks the code for it; see the
+   next section).
 
 Never print database credentials or embed them in Alembic configuration. The
 deployed v1 database is already versioned; never recreate or restamp its
 baseline. Missing or invalid revision history must fail closed for manual
 recovery.
 
+## Database access through SQLAlchemy and Alembic
+
+Rule `api.data.sqlalchemy-and-alembic-only`: code connects to relational
+databases only through SQLAlchemy engines, connections and sessions, never by
+calling a database driver directly, and changes persistent schema only through
+Alembic revisions.
+
+- API v1 builds its Cloud SQL engine in `policyengine_api/data/orm.py`. The
+  Cloud SQL connector is passed to `create_engine` as `creator=`, so SQLAlchemy
+  opens and pools the connections.
+- API v2 builds its SQLModel engine from a `postgresql+psycopg` URL in
+  `policyengine_api/data/v2/database.py`.
+- Take connections and sessions from those engines, never from
+  `psycopg.connect`, `pymysql.connect`, `sqlite3.connect` or another driver.
+- Change schema only with an autogenerated revision (see "Mandatory generation
+  rule"). Application code, scripts and CI never call `create_all`,
+  `drop_all`, `Table.create` or `Table.drop`, build SQLAlchemy DDL constructs,
+  call Alembic operations outside `migrations/`, or run `CREATE`, `ALTER`,
+  `DROP` or `TRUNCATE` SQL.
+
+`tests/unit/data/test_database_access_boundaries.py` checks `policyengine_api/`,
+`scripts/`, `migrations/` and `.github/`, skipping tests and the generated
+revisions under `migrations/*/versions/`. It parses each module and reports:
+
+- imports of `psycopg`, `psycopg2`, `pymysql`, `MySQLdb`, `mysql.connector`,
+  `sqlite3`, `asyncpg`, `pg8000`, `aiosqlite` or `aiomysql`, including through
+  `importlib.import_module`;
+- access to the driver connection beneath SQLAlchemy: the `driver_connection`
+  and `dbapi_connection` attributes and `raw_connection()`;
+- `create_all()`, `drop_all()`, and `create()` or `drop()` on a table, index
+  or sequence; SQLAlchemy DDL constructs such as `CreateTable`; and `op.*`
+  calls or `alembic.op` imports outside `migrations/`;
+- string literals, other than docstrings, that begin a `CREATE`, `ALTER` or
+  `DROP` of a table, type, schema, database, index, sequence, view or
+  extension, or a `TRUNCATE`.
+
+### Declared exceptions
+
+The guard allows exactly two exceptions, both in
+`policyengine_api/data/v2/catalog/publication_staging.py`, which stages a
+metadata catalog before publishing it. The guard fails when either exception
+gains a name or a file, and when either is no longer used.
+
+1. **COPY through the driver connection.** Publication writes rows with
+   Postgres `COPY ... FROM STDIN`, which SQLAlchemy has no API for. `copy_rows`
+   uses the psycopg connection beneath a SQLAlchemy `Connection`
+   (`connection.connection.driver_connection`) that SQLAlchemy opened and
+   still manages, so the transaction stays SQLAlchemy's. It imports
+   `psycopg.sql` to compose the `COPY` statement and `psycopg.types.json.Jsonb`
+   to send JSON values.
+2. **Temporary staging tables.** `create_staging_tables` calls
+   `STAGING_METADATA.create_all` for tables declared `TEMPORARY` with
+   `ON COMMIT DROP`. Postgres drops them when the publishing transaction
+   commits, so they are not persistent schema. The guard checks that every
+   table on `STAGING_METADATA` compiles to `CREATE TEMPORARY TABLE ... ON
+   COMMIT DROP`.
+
+A new exception is a rule change: add it to the guard's declared sets, explain
+here why SQLAlchemy or Alembic can't do the job, and get it reviewed as one.
+
+### Out of scope
+
+Redis (`policyengine_api/runtime_cache/`) is a key-value cache, not a
+relational database. This rule does not cover it.
+
+### Related rules
+
+- `api.data.runtime-sql-confined`: raw SQL stays in the data-access layer.
+- `api.services.no-direct-queries`: services never query or open connections
+  themselves; they call the data-access layer.
+- `api.v2.single-orm-layer`: v2 models use SQLModel, with no second
+  SQLAlchemy ORM layer.
+- `api.migration.alembic-v1-qualified` and `api.migration.alembic-v2-qualified`:
+  CI applies the v1 and v2 revisions to disposable MySQL and Postgres
+  databases, and checks v2 for schema drift.
+- `api.migration.alembic-autogenerated-only` (doc-only): revisions are
+  autogenerated, never hand-written.
+- `api.startup.no-import-side-effects`: importing v2 modules, starting up and
+  serving a request open no network connections.
+
 ## API v1 database targets
 
 Alembic manages the API v1 MySQL schema only. The local SQLite database is a
@@ -212,8 +295,9 @@ policy.
 
 `alembic check` for v2 must detect schema drift.
 Application startup, model import, project provisioning, and external
-infrastructure provisioning must never call `create_all`, create or stamp
-application tables, or mutate versioned application data. The Supabase CLI is
+infrastructure provisioning must never call `create_all`
+(`api.data.sqlalchemy-and-alembic-only`), create or stamp application tables,
+or mutate versioned application data. The Supabase CLI is
 not an application schema or seed migration authority.
 
 Canonical metadata catalogs derived from the exact installed country and

docs/generated/rules.md

@@ -8,11 +8,11 @@ Every rule that governs `policyengine-api` is listed here with how it is enforce
 
 | Tier | enforced | ratcheted | advisory | doc-only | Total |
 | --- | ---: | ---: | ---: | ---: | ---: |
-| mechanical | 54 | 0 | 1 | 13 | 68 |
+| mechanical | 55 | 0 | 1 | 13 | 69 |
 | metric | 0 | 0 | 0 | 0 | 0 |
 | judgment | 0 | 0 | 0 | 1 | 1 |
 
-**Blocking:** 54 of 69 rules.
+**Blocking:** 55 of 70 rules.
 
 ## Gaps
 
@@ -123,6 +123,7 @@ flowchart LR
 | Rule | Statement | Tier | Status | Blocking | Enforced by | CI | Proof |
 | --- | --- | --- | --- | --- | --- | --- | --- |
 | `api.data.runtime-sql-confined` | Raw SQL is confined to the data-access layer; runtime modules and report orchestration use sessions and mapped models. | mechanical | enforced | yes | pytest `tests/unit/data/test_runtime_sql_boundaries.py` | `pr.yml:test` | **unproven** |
+| `api.data.sqlalchemy-and-alembic-only` | Code connects to relational databases only through SQLAlchemy engines, connections and sessions, never by calling a database driver directly, and changes persistent schema only through Alembic revisions. | mechanical | enforced | yes | pytest `tests/unit/data/test_database_access_boundaries.py` | `pr.yml:test` | yes |
 | `api.legacy.removed-layers-stay-removed` | Removed legacy persistence layers, the temporary DAO module and the legacy endpoints package stay removed. | mechanical | enforced | yes | pytest `tests/unit/data/test_runtime_sql_boundaries.py`, `tests/unit/routes/test_flask_route_boundaries.py`, `tests/unit/services/test_orm_service_boundaries.py` | `pr.yml:test` | **unproven** |
 | `api.routes.flask-blueprints-own-urls` | Legacy Flask URLs are owned by blueprints, and the Flask app assembles blueprints without legacy endpoint wiring. | mechanical | enforced | yes | pytest `tests/unit/routes/test_flask_route_boundaries.py` | `pr.yml:test` | **unproven** |
 | `api.routes.no-domain-logic-in-http` | Flask routes delegate domain processing, such as calculation, to services instead of doing it in the HTTP layer. | mechanical | enforced | yes | pytest `tests/unit/routes/test_flask_route_boundaries.py` | `pr.yml:test` | **unproven** |
@@ -135,6 +136,8 @@ flowchart LR
   r_api_data_runtime_sql_confined("api.data.runtime-sql-confined (enforced)")
   e_pytest_tests_unit_data_test_runtime_sql_boundaries_py["pytest tests/unit/data/test_runtime_sql_boundaries.py"]
   j_CI_pr_yml_test["CI pr.yml:test"]
+  r_api_data_sqlalchemy_and_alembic_only("api.data.sqlalchemy-and-alembic-only (enforced)")
+  e_pytest_tests_unit_data_test_database_access_boundaries_py["pytest tests/unit/data/test_database_access_boundaries.py"]
   r_api_legacy_removed_layers_stay_removed("api.legacy.removed-layers-stay-removed (enforced)")
   e_pytest_tests_unit_data_test_runtime_sql_boundaries_py__tests_unit_routes_test_flask_route_boundaries_py__tests_unit_services_test_orm_service_boundaries_py["pytest tests/unit/data/test_runtime_sql_boundaries.py, tests/unit/routes/test_flask_route_boundaries.py, tests/unit/services/test_orm_service_boundaries.py"]
   r_api_routes_flask_blueprints_own_urls("api.routes.flask-blueprints-own-urls (enforced)")
@@ -147,6 +150,8 @@ flowchart LR
   e_pytest_tests_unit_services_test_local_data_service_boundaries_py__tests_unit_services_test_orm_service_boundaries_py["pytest tests/unit/services/test_local_data_service_boundaries.py, tests/unit/services/test_orm_service_boundaries.py"]
   r_api_data_runtime_sql_confined --> e_pytest_tests_unit_data_test_runtime_sql_boundaries_py
   e_pytest_tests_unit_data_test_runtime_sql_boundaries_py --> j_CI_pr_yml_test
+  r_api_data_sqlalchemy_and_alembic_only --> e_pytest_tests_unit_data_test_database_access_boundaries_py
+  e_pytest_tests_unit_data_test_database_access_boundaries_py --> j_CI_pr_yml_test
   r_api_legacy_removed_layers_stay_removed --> e_pytest_tests_unit_data_test_runtime_sql_boundaries_py__tests_unit_routes_test_flask_route_boundaries_py__tests_unit_services_test_orm_service_boundaries_py
   e_pytest_tests_unit_data_test_runtime_sql_boundaries_py__tests_unit_routes_test_flask_route_boundaries_py__tests_unit_services_test_orm_service_boundaries_py --> j_CI_pr_yml_test
   r_api_routes_flask_blueprints_own_urls --> e_pytest_tests_unit_routes_test_flask_route_boundaries_py

rules/fixtures/api.data.sqlalchemy-and-alembic-only/bad_database_access.py

@@ -0,0 +1,27 @@
+"""Proof fixture: a driver connection and a runtime schema change.
+
+The rule must reject both. Both mistakes are taken from this repository's
+history: the Stage 12 provisioning script (removed in d8bd72191) opened
+psycopg connections directly, and the local database bootstrap added in
+6c7804100 created the v1 tables with metadata.create_all at runtime.
+"""
+
+import psycopg
+from sqlalchemy import Engine
+
+from policyengine_api.data.v1_models import V1Base
+
+
+def read_current_role(connection_url: str) -> str:
+    """Return the role the database authenticated, through a driver connection."""
+    with psycopg.connect(connection_url) as connection:
+        row = connection.execute("SELECT current_user").fetchone()
+    if row is None:
+        msg = "the database returned no current role"
+        raise RuntimeError(msg)
+    return str(row[0])
+
+
+def bootstrap_tables(engine: Engine) -> None:
+    """Create every v1 table when the application starts."""
+    V1Base.metadata.create_all(engine)

rules/fixtures/api.data.sqlalchemy-and-alembic-only/good_database_access.py

@@ -0,0 +1,23 @@
+"""Proof fixture: a query through a SQLAlchemy session. The rule must accept it.
+
+The connection comes from a SQLAlchemy engine and session, and the tables come
+from Alembic revisions applied before the application starts, so this module
+neither imports a database driver nor changes the schema.
+"""
+
+from sqlalchemy import func, select, text
+from sqlalchemy.orm import Session, sessionmaker
+
+from policyengine_api.data.v1_models import UserProfile
+
+
+def read_current_role(session_factory: sessionmaker[Session]) -> str:
+    """Return the role the database authenticated, through a SQLAlchemy session."""
+    with session_factory() as session:
+        return str(session.execute(text("SELECT current_user")).scalar_one())
+
+
+def count_user_profiles(session_factory: sessionmaker[Session]) -> int:
+    """Count user profiles in the table an Alembic revision created."""
+    with session_factory() as session:
+        return session.scalar(select(func.count()).select_from(UserProfile)) or 0

rules/registry.toml

@@ -273,6 +273,31 @@ workflow = ".github/workflows/pr.yml"
 job = "test"
 invocation = "make test"
 
+[[rule]]
+id = "api.data.sqlalchemy-and-alembic-only"
+category = "runtime-layers"
+statement = "Code connects to relational databases only through SQLAlchemy engines, connections and sessions, never by calling a database driver directly, and changes persistent schema only through Alembic revisions."
+rationale = "docs/engineering/skills/alembic-migrations.md#database-access-through-sqlalchemy-and-alembic"
+scope = ["policyengine_api/**/*.py", "scripts/**/*.py", "migrations/*/env.py", ".github/**/*.py"]
+tier = "mechanical"
+status = "enforced"
+autofix = false
+[rule.enforcer]
+kind = "pytest"
+nodes = [
+  "tests/unit/data/test_database_access_boundaries.py::test_non_test_code_reaches_database_drivers_only_where_declared",
+  "tests/unit/data/test_database_access_boundaries.py::test_non_test_code_changes_schema_only_through_alembic",
+  "tests/unit/data/test_database_access_boundaries.py::test_publication_staging_tables_are_temporary_and_dropped_at_commit",
+]
+[[rule.ci]]
+workflow = ".github/workflows/pr.yml"
+job = "test"
+invocation = "make test"
+[rule.proof]
+bad_fixture = "rules/fixtures/api.data.sqlalchemy-and-alembic-only/bad_database_access.py"
+good_fixture = "rules/fixtures/api.data.sqlalchemy-and-alembic-only/good_database_access.py"
+test = "tests/unit/data/test_database_access_boundaries.py::test_rule_proof_sqlalchemy_and_alembic_only"
+
 [[rule]]
 id = "api.legacy.removed-layers-stay-removed"
 category = "runtime-layers"