From 6eb793db2a5533f89d53185b108428955a9c1a8a Mon Sep 17 00:00:00 2001 From: Anthony Volk <14987227+anth-volk@users.noreply.github.com> Date: Tue, 20 Oct 2026 16:55:53 +0400 Subject: [PATCH] Enforce SQLAlchemy-only database connections and Alembic-only schema changes Register api.data.sqlalchemy-and-alembic-only as an enforced mechanical rule. tests/unit/data/test_database_access_boundaries.py parses every non-test module under policyengine_api/, scripts/, migrations/ and .github/ (excluding migrations/*/versions/) and fails on: - imports of psycopg, psycopg2, pymysql, MySQLdb, mysql.connector, sqlite3, asyncpg, pg8000, aiosqlite or aiomysql, including through importlib.import_module, and access to the driver connection beneath SQLAlchemy (driver_connection, dbapi_connection, raw_connection()); - create_all, drop_all, create or drop on a table, index or sequence, SQLAlchemy DDL constructs, Alembic operations outside migrations/, and 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. The only allowed findings are in catalog publication staging: the COPY step through the psycopg connection beneath a SQLAlchemy Connection, and STAGING_METADATA.create_all for temporary tables. A separate test checks that every staging table compiles to CREATE TEMPORARY TABLE ... ON COMMIT DROP. The comparison fails both when a finding is undeclared and when a declaration is no longer used. The proof fixtures reproduce two past cases from this repository: the direct psycopg.connect calls in the removed Stage 12 provisioning script and the runtime V1Base.metadata.create_all in the local bootstrap. The rule is explained in docs/engineering/skills/alembic-migrations.md, which also states that Redis is out of scope. Co-Authored-By: Claude Opus 5.5 --- AGENTS.md | 6 +- .../sqlalchemy-alembic-database-rule.added.md | 1 + docs/engineering/skills/alembic-migrations.md | 92 +++- docs/generated/rules.md | 9 +- .../bad_database_access.py | 27 ++ .../good_database_access.py | 23 + rules/registry.toml | 25 + .../data/test_database_access_boundaries.py | 457 ++++++++++++++++++ 8 files changed, 631 insertions(+), 9 deletions(-) create mode 100644 changelog.d/sqlalchemy-alembic-database-rule.added.md create mode 100644 rules/fixtures/api.data.sqlalchemy-and-alembic-only/bad_database_access.py create mode 100644 rules/fixtures/api.data.sqlalchemy-and-alembic-only/good_database_access.py create mode 100644 tests/unit/data/test_database_access_boundaries.py diff --git a/AGENTS.md b/AGENTS.md index 5cb02b98a..28b5f9f2e 100644 --- a/AGENTS.md +++ b/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 diff --git a/changelog.d/sqlalchemy-alembic-database-rule.added.md b/changelog.d/sqlalchemy-alembic-database-rule.added.md new file mode 100644 index 000000000..480ec232c --- /dev/null +++ b/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. diff --git a/docs/engineering/skills/alembic-migrations.md b/docs/engineering/skills/alembic-migrations.md index d4a043d91..f5c72523c 100644 --- a/docs/engineering/skills/alembic-migrations.md +++ b/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 diff --git a/docs/generated/rules.md b/docs/generated/rules.md index dda3ae373..743199ca0 100644 --- a/docs/generated/rules.md +++ b/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 diff --git a/rules/fixtures/api.data.sqlalchemy-and-alembic-only/bad_database_access.py b/rules/fixtures/api.data.sqlalchemy-and-alembic-only/bad_database_access.py new file mode 100644 index 000000000..e69e52c1c --- /dev/null +++ b/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) diff --git a/rules/fixtures/api.data.sqlalchemy-and-alembic-only/good_database_access.py b/rules/fixtures/api.data.sqlalchemy-and-alembic-only/good_database_access.py new file mode 100644 index 000000000..90cd6176e --- /dev/null +++ b/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 diff --git a/rules/registry.toml b/rules/registry.toml index 0c714021f..2ce003d9a 100644 --- a/rules/registry.toml +++ b/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" diff --git a/tests/unit/data/test_database_access_boundaries.py b/tests/unit/data/test_database_access_boundaries.py new file mode 100644 index 000000000..542f44209 --- /dev/null +++ b/tests/unit/data/test_database_access_boundaries.py @@ -0,0 +1,457 @@ +"""Guards for rule api.data.sqlalchemy-and-alembic-only (rules/registry.toml). + +Non-test code reaches relational databases only through SQLAlchemy engines, +connections and sessions, and changes persistent schema only through Alembic +revisions. The two declared exceptions, both in catalog publication staging, +are explained in docs/engineering/skills/alembic-migrations.md under +"Database access through SQLAlchemy and Alembic". + +Like the api.data.runtime-sql-confined guard in test_runtime_sql_boundaries.py, +these tests walk the source tree, collect offenders and compare them with the +expected set. They parse each module with ast so that comments and docstrings +are not mistaken for imports, calls or SQL. +""" + +from __future__ import annotations + +import ast +import re +from pathlib import Path +from typing import NamedTuple + +from sqlalchemy.dialects import postgresql +from sqlalchemy.schema import CreateTable + +from policyengine_api.data.v2.catalog.publication_staging import STAGING_METADATA + +PROJECT_ROOT = Path(__file__).parents[3] +NON_TEST_ROOTS = ("policyengine_api", "scripts", "migrations", ".github") +RULE_DOC = ( + "docs/engineering/skills/alembic-migrations.md" + "#database-access-through-sqlalchemy-and-alembic" +) +PROOF_FIXTURES = "rules/fixtures/api.data.sqlalchemy-and-alembic-only" + +DATABASE_DRIVER_MODULES = ( + "psycopg", + "psycopg2", + "pymysql", + "MySQLdb", + "mysql.connector", + "sqlite3", + "asyncpg", + "pg8000", + "aiosqlite", + "aiomysql", +) +# Attributes and methods that hand out the driver connection beneath a +# SQLAlchemy connection or engine. +DRIVER_CONNECTION_ATTRIBUTES = frozenset({"driver_connection", "dbapi_connection"}) +DRIVER_CONNECTION_METHODS = frozenset({"raw_connection"}) + +SCHEMA_COLLECTION_METHODS = frozenset({"create_all", "drop_all"}) +SCHEMA_OBJECT_METHODS = frozenset({"create", "drop"}) +SCHEMA_OBJECT_CONSTRUCTORS = frozenset({"Table", "Index", "Sequence"}) +SCHEMA_OBJECT_COLLECTIONS = frozenset({"tables", "sorted_tables"}) +DDL_CONSTRUCTORS = frozenset( + { + "DDL", + "CreateTable", + "DropTable", + "CreateIndex", + "DropIndex", + "CreateSchema", + "DropSchema", + "CreateSequence", + "DropSequence", + "AddConstraint", + "DropConstraint", + } +) +ALEMBIC_OPERATION_MODULES = ("alembic.op", "alembic.operations") +ALEMBIC_OPERATION_RECEIVERS = frozenset({"op", "alembic.op"}) +# A schema-changing statement at the start of a string, of a line in it, or +# after a semicolon. Anchoring to a statement start keeps ordinary messages +# such as "failed to create table rows" from matching. +SCHEMA_CHANGING_SQL = re.compile( + r"(?:^|;)\s*(" + r"(?:CREATE(?:\s+OR\s+REPLACE)?" + r"(?:\s+(?:GLOBAL|LOCAL|MATERIALIZED|TEMP|TEMPORARY|UNIQUE|UNLOGGED))*" + r"|ALTER|DROP)" + r"\s+(?:DATABASE|EXTENSION|INDEX|SCHEMA|SEQUENCE|TABLE|TYPE|VIEW)" + r"|TRUNCATE" + r")\b", + re.IGNORECASE | re.MULTILINE, +) + +PUBLICATION_STAGING_MODULE = "policyengine_api/data/v2/catalog/publication_staging.py" +# Catalog publication streams rows with Postgres COPY ... FROM STDIN, which +# SQLAlchemy has no API for. It runs COPY through the psycopg connection beneath +# a SQLAlchemy Connection that SQLAlchemy opened and still manages, and uses +# psycopg's SQL composition and JSONB adapter to do it. +DECLARED_DRIVER_ACCESS = frozenset( + { + (PUBLICATION_STAGING_MODULE, "import psycopg.sql"), + (PUBLICATION_STAGING_MODULE, "import psycopg.types.json.Jsonb"), + (PUBLICATION_STAGING_MODULE, "attribute .driver_connection"), + } +) +# Catalog publication creates its staging tables as TEMPORARY tables that +# Postgres drops at commit, so they never outlive the publishing transaction. +# test_publication_staging_tables_are_temporary_and_dropped_at_commit checks +# every table on STAGING_METADATA. +DECLARED_SCHEMA_CHANGES = frozenset( + {(PUBLICATION_STAGING_MODULE, "call STAGING_METADATA.create_all()")} +) + + +class DatabaseAccessFinding(NamedTuple): + """One construct a check reports, with where it occurs.""" + + module_path: str + line_number: int + construct: str + + @property + def declaration(self) -> tuple[str, str]: + """The (module, construct) pair an allowlist entry declares.""" + return (self.module_path, self.construct) + + +def non_test_python_modules(project_root: Path = PROJECT_ROOT) -> list[str]: + """Return every non-test Python module the rule covers, as relative paths. + + Tests and the generated Alembic revisions under migrations/*/versions/ are + excluded. + """ + module_paths = [] + for root_name in NON_TEST_ROOTS: + for module_file in sorted((project_root / root_name).rglob("*.py")): + relative_path = module_file.relative_to(project_root) + if not _is_test_or_alembic_revision(relative_path): + module_paths.append(relative_path.as_posix()) + assert module_paths, f"no Python modules found under {NON_TEST_ROOTS}" + return module_paths + + +def _is_test_or_alembic_revision(relative_path: Path) -> bool: + directory_names = relative_path.parts[:-1] + if "tests" in directory_names: + return True + return ( + len(directory_names) >= 3 + and directory_names[0] == "migrations" + and directory_names[2] == "versions" + ) + + +def _read_module(module_path: str) -> str: + return (PROJECT_ROOT / module_path).read_text(encoding="utf-8") + + +def _called_name(call: ast.Call) -> str | None: + if isinstance(call.func, ast.Name): + return call.func.id + if isinstance(call.func, ast.Attribute): + return call.func.attr + return None + + +def _is_driver_module(qualified_name: str) -> bool: + return any( + qualified_name == driver_module + or qualified_name.startswith(f"{driver_module}.") + for driver_module in DATABASE_DRIVER_MODULES + ) + + +def _imported_names(node: ast.AST) -> list[str]: + """Fully qualified names an import statement or import call brings in.""" + if isinstance(node, ast.Import): + return [alias.name for alias in node.names] + if isinstance(node, ast.ImportFrom) and node.level == 0 and node.module: + return [f"{node.module}.{alias.name}" for alias in node.names] + if ( + isinstance(node, ast.Call) + and _called_name(node) in {"import_module", "__import__"} + and node.args + and isinstance(node.args[0], ast.Constant) + and isinstance(node.args[0].value, str) + ): + return [node.args[0].value] + return [] + + +def _driver_access_constructs(node: ast.AST) -> list[str]: + constructs = [ + f"import {imported_name}" + for imported_name in _imported_names(node) + if _is_driver_module(imported_name) + ] + if isinstance(node, ast.Attribute) and node.attr in DRIVER_CONNECTION_ATTRIBUTES: + constructs.append(f"attribute .{node.attr}") + if isinstance(node, ast.Call) and _called_name(node) in DRIVER_CONNECTION_METHODS: + constructs.append(f"call .{_called_name(node)}()") + return constructs + + +def find_driver_access( + source_text: str, module_path: str +) -> list[DatabaseAccessFinding]: + """Report every direct database-driver import or driver-connection access. + + Covers import statements, importlib.import_module and __import__ calls + naming a driver module, the driver_connection and dbapi_connection + attributes, and raw_connection() calls. + """ + return [ + DatabaseAccessFinding(module_path, node.lineno, construct) + for node in ast.walk(ast.parse(source_text)) + for construct in _driver_access_constructs(node) + ] + + +def _assigned_names(assignment: ast.Assign | ast.AnnAssign) -> set[str]: + targets = ( + assignment.targets + if isinstance(assignment, ast.Assign) + else [assignment.target] + ) + return {target.id for target in targets if isinstance(target, ast.Name)} + + +def _schema_object_names(tree: ast.Module) -> set[str]: + """Names bound to a Table, Index or Sequence, or to a loop over tables. + + A name counts when it is assigned a Table(...), Index(...) or + Sequence(...) call, assigned a tuple or list of such names, or is the + target of a loop over one of those collections or over a metadata's + tables or sorted_tables. + """ + bindings = [ + (_assigned_names(node), node.value) + for node in ast.walk(tree) + if isinstance(node, (ast.Assign, ast.AnnAssign)) + ] + object_names = { + bound_name + for bound_names, assigned_value in bindings + if isinstance(assigned_value, ast.Call) + and _called_name(assigned_value) in SCHEMA_OBJECT_CONSTRUCTORS + for bound_name in bound_names + } + collection_names = { + bound_name + for bound_names, assigned_value in bindings + if isinstance(assigned_value, (ast.Tuple, ast.List, ast.Set)) + and any( + isinstance(element, ast.Name) and element.id in object_names + for element in assigned_value.elts + ) + for bound_name in bound_names + } + for node in ast.walk(tree): + if not isinstance(node, (ast.For, ast.AsyncFor, ast.comprehension)): + continue + if not isinstance(node.target, ast.Name): + continue + iterates_over_tables = any( + (isinstance(part, ast.Attribute) and part.attr in SCHEMA_OBJECT_COLLECTIONS) + or (isinstance(part, ast.Name) and part.id in collection_names) + for part in ast.walk(node.iter) + ) + if iterates_over_tables: + object_names.add(node.target.id) + return object_names + + +def _targets_schema_object(call: ast.Call, schema_object_names: set[str]) -> bool: + """Whether a create() or drop() call acts on a table, index or sequence.""" + if any(keyword.arg == "checkfirst" for keyword in call.keywords): + return True + if not isinstance(call.func, ast.Attribute): + return False + receiver = call.func.value + if isinstance(receiver, ast.Attribute) and receiver.attr == "__table__": + return True + if ( + isinstance(receiver, ast.Subscript) + and isinstance(receiver.value, ast.Attribute) + and receiver.value.attr in SCHEMA_OBJECT_COLLECTIONS + ): + return True + return isinstance(receiver, ast.Name) and receiver.id in schema_object_names + + +def _schema_call_constructs( + call: ast.Call, + schema_object_names: set[str], + *, + outside_migrations: bool, +) -> list[str]: + called_name = _called_name(call) + if called_name in DDL_CONSTRUCTORS: + return [f"construct {called_name}()"] + if not isinstance(call.func, ast.Attribute): + return [] + receiver = ast.unparse(call.func.value) + if called_name in SCHEMA_COLLECTION_METHODS or ( + called_name in SCHEMA_OBJECT_METHODS + and _targets_schema_object(call, schema_object_names) + ): + return [f"call {receiver}.{called_name}()"] + if outside_migrations and receiver in ALEMBIC_OPERATION_RECEIVERS: + return [f"call {receiver}.{called_name}()"] + return [] + + +def _alembic_operation_imports(node: ast.AST) -> list[str]: + return [ + f"import {imported_name}" + for imported_name in _imported_names(node) + if any( + imported_name == module_name or imported_name.startswith(f"{module_name}.") + for module_name in ALEMBIC_OPERATION_MODULES + ) + ] + + +def _docstring_node_ids(tree: ast.Module) -> set[int]: + docstring_ids = set() + for node in ast.walk(tree): + if not isinstance( + node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef) + ): + continue + first_statement = node.body[0] if node.body else None + if ( + isinstance(first_statement, ast.Expr) + and isinstance(first_statement.value, ast.Constant) + and isinstance(first_statement.value.value, str) + ): + docstring_ids.add(id(first_statement.value)) + return docstring_ids + + +def find_schema_changes( + source_text: str, module_path: str +) -> list[DatabaseAccessFinding]: + """Report every schema change a module makes outside an Alembic revision. + + Covers create_all() and drop_all(); create() and drop() on a table, index + or sequence; SQLAlchemy DDL constructs such as CreateTable; Alembic + operations (op.*) outside migrations/; and string literals, other than + docstrings, that start a CREATE, ALTER or DROP of a table, type, schema, + database, index, sequence, view or extension, or a TRUNCATE. + """ + tree = ast.parse(source_text) + outside_migrations = not module_path.startswith("migrations/") + schema_object_names = _schema_object_names(tree) + docstring_ids = _docstring_node_ids(tree) + findings = [] + for node in ast.walk(tree): + constructs: list[str] = [] + if isinstance(node, ast.Call): + constructs.extend( + _schema_call_constructs( + node, + schema_object_names, + outside_migrations=outside_migrations, + ) + ) + if outside_migrations: + constructs.extend(_alembic_operation_imports(node)) + if ( + isinstance(node, ast.Constant) + and isinstance(node.value, str) + and id(node) not in docstring_ids + ): + constructs.extend( + "SQL text " + " ".join(statement.group(1).upper().split()) + for statement in SCHEMA_CHANGING_SQL.finditer(node.value) + ) + findings.extend( + DatabaseAccessFinding(module_path, node.lineno, construct) + for construct in constructs + ) + return findings + + +def _assert_only_declared( + findings: list[DatabaseAccessFinding], + declared: frozenset[tuple[str, str]], + description: str, +) -> None: + """Fail on any undeclared finding and on any declaration nothing needs.""" + undeclared = sorted( + finding for finding in findings if finding.declaration not in declared + ) + unused_declarations = sorted( + declared - {finding.declaration for finding in findings} + ) + problems = [ + f"{finding.module_path}:{finding.line_number}: {finding.construct}" + for finding in undeclared + ] + problems.extend( + f"declared exception no longer needed, remove it: {module_path}: {construct}" + for module_path, construct in unused_declarations + ) + assert not problems, ( + f"{description} outside the declared exceptions (see {RULE_DOC}):\n" + + "\n".join(problems) + ) + + +def test_non_test_code_reaches_database_drivers_only_where_declared() -> None: + findings = [ + finding + for module_path in non_test_python_modules() + for finding in find_driver_access(_read_module(module_path), module_path) + ] + _assert_only_declared( + findings, DECLARED_DRIVER_ACCESS, "Direct database-driver access" + ) + + +def test_non_test_code_changes_schema_only_through_alembic() -> None: + findings = [ + finding + for module_path in non_test_python_modules() + for finding in find_schema_changes(_read_module(module_path), module_path) + ] + _assert_only_declared( + findings, DECLARED_SCHEMA_CHANGES, "Schema change outside Alembic" + ) + + +def test_publication_staging_tables_are_temporary_and_dropped_at_commit() -> None: + assert STAGING_METADATA.tables, "STAGING_METADATA declares no staging tables" + for table in STAGING_METADATA.tables.values(): + create_statement = str( + CreateTable(table).compile(dialect=postgresql.dialect()) + ).strip() + assert create_statement.startswith(f"CREATE TEMPORARY TABLE {table.name} "), ( + f"staging table {table.name} must be TEMPORARY" + ) + assert create_statement.endswith("ON COMMIT DROP"), ( + f"staging table {table.name} must be dropped at commit" + ) + + +def test_rule_proof_sqlalchemy_and_alembic_only() -> None: + """Proof for rule api.data.sqlalchemy-and-alembic-only (rules/registry.toml).""" + bad_fixture = f"{PROOF_FIXTURES}/bad_database_access.py" + good_fixture = f"{PROOF_FIXTURES}/good_database_access.py" + bad_source = _read_module(bad_fixture) + good_source = _read_module(good_fixture) + + bad_driver_access = find_driver_access(bad_source, bad_fixture) + bad_schema_changes = find_schema_changes(bad_source, bad_fixture) + assert {finding.construct for finding in bad_driver_access} == {"import psycopg"} + assert {finding.construct for finding in bad_schema_changes} == { + "call V1Base.metadata.create_all()" + } + + assert find_driver_access(good_source, good_fixture) == [] + assert find_schema_changes(good_source, good_fixture) == []