clubesportsudes/team-tryouts
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(db): mesurer la derive du schema, au lieu de la supposer

Browse Source 
DB-001. create_all() cree les tables manquantes et ne fait jamais d ALTER.
Une colonne ajoutee a un modele il y a six mois est donc absente de toute
base qui possedait deja la table, et rien ne le dit : l application demarre,
et la premiere requete qui touche cette colonne echoue a l execution. C est
la raison d etre de migrations/add_tryout_coaches.py, ecrit a la main pour
rattraper un cas. Personne ne sait combien il y en a d autres.

app/supporting_scripts/schema_report.py compare le catalogue d une base
vivante aux modeles : tables, colonnes, types, nullabilite, contraintes
d unicite, cles etrangeres, index. En **lecture seule** — il ouvre une
connexion, lit, imprime, sort. Aucun DDL, aucun DML.

Les constats sont classes par ce qu ils coutent, pas par ce qu ils sont :

- BLOCKING : les modeles l attendent, la base ne l a pas. C est la derive ;
- RISK : la base l a, aucun modele ne le decrit. Inoffensif tant que rien
  ne bouge — et **un alembic --autogenerate proposera de le supprimer**,
  avec ses donnees. C est la classe qu on lit en entier ;
- DIFFERENCE : types, nullabilite, contraintes. Chacune demande un humain.

Les types sont compares apres compilation vers le meme dialecte : opposer
String(200) a VARCHAR(200) en chaines aurait signale chaque colonne comme
differente, et un rapport qui crie partout ne se lit plus.

--check-seed-accounts repond a la question de SEC-003 a laquelle le depot
ne peut pas repondre : le compte admin/password seme par clear_db.py
existe-t-il encore, et son mot de passe est-il toujours celui-la.

13 tests le pilotent contre des bases SQLite fabriquees pour diverger d une
facon connue. Le cas qui compte le plus est la base propre : un rapport qui
crie sur une base saine ne sera pas lu, et un rapport qui dit « aucun
ecart » sur une base derivee est pire que pas de rapport — c est un feu
vert pour laisser autogenerate ecrire la difference en DROP.

docs/database-schema.md donne la suite, etape par etape, avec le piege de
DB-002 en toutes lettres : la migration initiale doit decrire la base telle
qu elle est, pas telle que les modeles la decrivent. Generer depuis les
modeles puis estampiller revient a declarer que la derive n existe pas.

Alembic n est pas ajoute aux dependances : rien ne l utilise encore, et une
dependance que rien n utilise est exactement ce que l audit reprochait
ailleurs. Le document dit a quelle etape l ajouter.

530 tests.
This commit is contained in:
GGThed
2026-08-11 15:00:18 -04:00
parent 3882b6035f
commit 7b9eee4805
3 changed files with 764 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/supporting_scripts/schema_report.py
+384
View File
@@ -0,0 +1,384 @@
"""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'<Finding {self.severity} {self.table} {self.detail}>'
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())