"""Compare a live database against the models (DB-001). Why this exists --------------- `db.create_all()` creates missing tables and never ALTERs an existing one. A column added to a model months ago is therefore simply absent from any database that already had the table, and nothing says so: the application starts, and the first query touching that column fails at runtime. The audit called the accumulated result "les dérives" and could not measure it, because measuring it needs the production database. This script measures it. It is **read-only** — it opens a connection, reads the catalogue, prints a report and exits. It issues no DDL and no DML, and takes no locks beyond what reading `information_schema` takes. It is the prerequisite for everything in the DB wave: `DB-002` asks for an initial Alembic migration describing the **real** schema rather than the models', and this is what tells you what the real schema is. Usage ----- # Against whatever DATABASE_URL points at python app/supporting_scripts/schema_report.py # Against a restored copy, which is the safe way to do it first python app/supporting_scripts/schema_report.py \ --url postgresql://user:pass@host:5432/restored_copy # Also look for the seeded admin/password account (SEC-003) python app/supporting_scripts/schema_report.py --check-seed-accounts Exit codes ---------- 0 the live schema matches the models 1 drift found — the report says what 2 could not connect or read the catalogue Reading the output ------------------ Findings are grouped by what they cost you: BLOCKING the application will fail at runtime — a table or column the models use and the database does not have. RISK the database has something the models do not describe. Harmless to the running application, but an Alembic autogenerate would propose to DROP it, which is how a corrective migration deletes a column somebody still needed. DIFFERENCE type, nullability, default or constraint disagreements. Each one needs a human: some are dialect spelling, some are real. """ import argparse import os import sys # Importable as a script from the project root, like the other supporting # scripts: `python app/supporting_scripts/schema_report.py`. sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) from sqlalchemy import create_engine, inspect # noqa: E402 from sqlalchemy.exc import SQLAlchemyError # noqa: E402 BLOCKING = 'BLOCKING' RISK = 'RISK' DIFFERENCE = 'DIFFERENCE' class Finding: """One disagreement between the models and the live database.""" def __init__(self, severity, table, detail, consequence=''): self.severity = severity self.table = table self.detail = detail self.consequence = consequence def __str__(self): line = f' [{self.severity:10}] {self.table}: {self.detail}' if self.consequence: line += f'\n → {self.consequence}' return line def __repr__(self): # pragma: no cover — debugging aid return f'' def model_metadata(): """The schema the models describe. Imports app.models for its side effect: importing the modules is what registers every table on the shared metadata. """ from app.extensions import db from app.models import User # noqa: F401 — registers the whole model package return db.metadata def _type_of(column_type, dialect): """A type as this dialect spells it, so the two sides are comparable. Comparing `String(200)` with `VARCHAR(200)` as strings would report every column as different. Compiling both against the same dialect makes the comparison mean something. """ try: return column_type.compile(dialect=dialect) except Exception: # noqa: BLE001 — an uncompilable type is still reportable return str(column_type) def compare_tables(metadata, inspector): """Tables the models expect against tables the database has.""" findings = [] model_tables = set(metadata.tables) live_tables = set(inspector.get_table_names()) for name in sorted(model_tables - live_tables): findings.append( Finding( BLOCKING, name, 'table is missing from the database', 'every query against this model fails. create_all() would ' 'create it — which is why the absence can survive unnoticed ' 'on a machine where AUTO_CREATE_TABLES is on.', ) ) for name in sorted(live_tables - model_tables): findings.append( Finding( RISK, name, 'table exists in the database and in no model', 'an Alembic autogenerate would propose to DROP it. Decide ' 'before running one: it may be a leftover, or it may be the ' 'only copy of something.', ) ) return findings, sorted(model_tables & live_tables) def compare_columns(metadata, inspector, table_name, dialect): """Column-by-column, for one table.""" findings = [] model_columns = {c.name: c for c in metadata.tables[table_name].columns} live_columns = {c['name']: c for c in inspector.get_columns(table_name)} for name in sorted(set(model_columns) - set(live_columns)): column = model_columns[name] findings.append( Finding( BLOCKING, table_name, f'column "{name}" is in the model and not in the database', 'this is exactly what create_all() cannot fix: it never ' 'ALTERs. Any query selecting or writing this column fails.' + ( '' if column.nullable else ' The column is NOT NULL, so the ' 'corrective migration needs a default or a backfill.' ), ) ) for name in sorted(set(live_columns) - set(model_columns)): findings.append( Finding( RISK, table_name, f'column "{name}" is in the database and not in any model', 'an autogenerated migration would propose to DROP it, taking ' 'its data. Check whether something outside the application ' 'reads it before agreeing.', ) ) for name in sorted(set(model_columns) & set(live_columns)): model_column, live_column = model_columns[name], live_columns[name] model_type = _type_of(model_column.type, dialect) live_type = _type_of(live_column['type'], dialect) if model_type != live_type: findings.append( Finding( DIFFERENCE, table_name, f'column "{name}" type: model says {model_type}, database says {live_type}', 'a narrower column in the database silently truncates or ' 'rejects; a wider one is usually harmless.', ) ) if bool(model_column.nullable) != bool(live_column.get('nullable', True)): findings.append( Finding( DIFFERENCE, table_name, f'column "{name}" nullability: model says ' f'{"NULL" if model_column.nullable else "NOT NULL"}, database says ' f'{"NULL" if live_column.get("nullable", True) else "NOT NULL"}', 'a NOT NULL the database does not enforce is a constraint ' 'the application only believes it has.', ) ) return findings def compare_constraints(metadata, inspector, table_name): """Unique constraints, indexes and foreign keys. Named constraints are compared by the columns they cover rather than by name: the same rule declared under two names is the same rule, and reporting it as a difference would bury the ones that matter. """ findings = [] table = metadata.tables[table_name] def column_sets(entries, key): return {tuple(sorted(entry[key] or [])) for entry in entries} model_unique = { tuple(sorted(c.name for c in constraint.columns)) for constraint in table.constraints if constraint.__class__.__name__ == 'UniqueConstraint' } live_unique = column_sets(inspector.get_unique_constraints(table_name), 'column_names') for columns in sorted(model_unique - live_unique): findings.append( Finding( DIFFERENCE, table_name, f'unique constraint on {list(columns)} is declared and absent from the database', 'the application believes duplicates are impossible here. ' 'They are not, and two concurrent requests will prove it.', ) ) model_fks = { tuple(sorted(fk.parent.name for fk in constraint.elements)) for constraint in table.foreign_key_constraints } live_fks = column_sets(inspector.get_foreign_keys(table_name), 'constrained_columns') for columns in sorted(model_fks - live_fks): findings.append( Finding( DIFFERENCE, table_name, f'foreign key on {list(columns)} is declared and absent from the database', 'orphan rows are possible, and ON DELETE behaviour is not ' 'being enforced by the database at all.', ) ) model_indexes = {tuple(sorted(c.name for c in index.columns)) for index in table.indexes} live_indexes = column_sets(inspector.get_indexes(table_name), 'column_names') for columns in sorted(model_indexes - live_indexes): findings.append( Finding( DIFFERENCE, table_name, f'index on {list(columns)} is declared and absent from the database', 'correctness is unaffected; the queries that rely on it are ' 'doing sequential scans.', ) ) return findings def collect_findings(engine): """Every disagreement between the models and this database.""" metadata = model_metadata() inspector = inspect(engine) findings, shared_tables = compare_tables(metadata, inspector) for table_name in shared_tables: findings.extend(compare_columns(metadata, inspector, table_name, engine.dialect)) findings.extend(compare_constraints(metadata, inspector, table_name)) return findings def find_seed_accounts(engine): """Accounts matching the credentials clear_db.py used to seed (SEC-003). The script was removed from the deployment, but it had already been run: the audit could not tell whether an `admin` account with the password `password` still exists in production, and that question cannot be answered from the repository. Returns: list[tuple]: (username, role, whether the known password matches). """ from sqlalchemy import text from app.extensions import check_password with engine.connect() as connection: rows = connection.execute( text('SELECT username, role, password_hash FROM users WHERE username = :name'), {'name': 'admin'}, ).fetchall() results = [] for username, role, password_hash in rows: try: matches = check_password(password_hash, 'password') except Exception: # noqa: BLE001 — an unreadable hash is not a match matches = False results.append((username, role, matches)) return results def main(argv=None): parser = argparse.ArgumentParser(description=__doc__.split('\n')[0]) parser.add_argument( '--url', default=os.getenv('DATABASE_URL'), help='Database URL. Defaults to DATABASE_URL. Point it at a restored copy the first time.', ) parser.add_argument( '--check-seed-accounts', action='store_true', help='Also look for the admin/password account seeded by clear_db.py (SEC-003).', ) args = parser.parse_args(argv) if not args.url: print('No database URL. Pass --url or set DATABASE_URL.', file=sys.stderr) return 2 from app.app import normalise_database_url try: engine = create_engine(normalise_database_url(args.url)) findings = collect_findings(engine) except SQLAlchemyError as exc: print(f'Could not read the schema: {exc}', file=sys.stderr) return 2 print('=' * 78) print('Schema report — models vs live database (DB-001)') print('=' * 78) if not findings: print('\nNo drift. The live schema matches the models.') for severity in (BLOCKING, RISK, DIFFERENCE): group = [f for f in findings if f.severity == severity] if not group: continue print(f'\n{severity} — {len(group)} finding(s)') for finding in group: print(finding) if args.check_seed_accounts: print('\n' + '=' * 78) print('Seeded accounts (SEC-003)') print('=' * 78) try: accounts = find_seed_accounts(engine) except SQLAlchemyError as exc: print(f'Could not check: {exc}') else: if not accounts: print('No account named "admin".') for username, role, matches in accounts: verdict = ( 'PASSWORD IS STILL "password" — change it now' if matches else 'password has been changed' ) print(f' {username} ({role}): {verdict}') blocking = sum(1 for f in findings if f.severity == BLOCKING) print(f'\n{len(findings)} finding(s), {blocking} blocking.') return 1 if findings else 0 if __name__ == '__main__': # pragma: no cover sys.exit(main())