"""Reclassify cached Alpha checks without changing upstream evidence.""" import sqlalchemy as sa from alembic import op revision = "0018" down_revision = "0017" branch_labels = None depends_on = None def check_type(checks, separate_limits): """Frozen classification for reversible data migration; never import mutable app code.""" checks = checks if isinstance(checks, list) else [] if separate_limits: checks = [c for c in checks if not (isinstance(c, dict) and c.get("name") == "REGULAR_SUBMISSION")] valid = [c for c in checks if isinstance(c, dict)] failures = sum(c.get("result") == "FAIL" for c in valid) if failures: return "FAIL_1" if failures == 1 else "FAIL_2" if not checks or len(valid) != len(checks) or any(c.get("result") != "PASS" for c in valid): return "PENDING" return "PASS" if any(c.get("name") == "PROD_CORRELATION" for c in valid) else "PRE_CHECK" def reclassify(separate_limits): """Update only affected derived columns, in bounded batches; raw snapshots stay intact.""" table = sa.table("alphas", sa.column("id", sa.String()), sa.column("checks", sa.JSON()), sa.column("check_type", sa.String())) connection = op.get_bind() last_id = None while True: query = sa.select(table.c.id, table.c.checks).order_by(table.c.id).limit(500) if last_id is not None: query = query.where(table.c.id > last_id) rows = connection.execute(query).mappings().all() if not rows: break updates = [ {"snapshot_id": row["id"], "classification": check_type(row["checks"], separate_limits)} for row in rows if isinstance(row["checks"], list) and any( isinstance(c, dict) and c.get("name") == "REGULAR_SUBMISSION" for c in row["checks"]) ] if updates: connection.execute(table.update().where(table.c.id == sa.bindparam("snapshot_id")) .values(check_type=sa.bindparam("classification")), updates) last_id = rows[-1]["id"] def upgrade(): reclassify(True) def downgrade(): reclassify(False)