"""Atelier Nymphar.ai : remplacer une partition, simuler deux pannes, rejouer. Python 3.10+, bibliothèque standard uniquement. Données de démonstration fictives. Ce programme teste le contrat d'écriture SQLite, pas Airflow ou Dagster. """ import argparse from datetime import date, datetime, timezone import hashlib import json from pathlib import Path import sqlite3 import tempfile import uuid CODE_VERSION = "atelier-1" def digest(value): encoded = json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")) return hashlib.sha256(encoded.encode()).hexdigest() def connect(path): conn = sqlite3.connect(path, isolation_level=None, timeout=5) conn.executescript(""" CREATE TABLE IF NOT EXISTS ventes ( jour TEXT NOT NULL, commande TEXT NOT NULL, centimes INTEGER NOT NULL CHECK (centimes >= 0), PRIMARY KEY (jour, commande) ); CREATE TABLE IF NOT EXISTS partitions ( jour TEXT PRIMARY KEY, source_hash TEXT NOT NULL, code_version TEXT NOT NULL, lignes INTEGER NOT NULL, total_centimes INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS tentatives ( id TEXT PRIMARY KEY, jour TEXT NOT NULL, source_hash TEXT NOT NULL, code_version TEXT NOT NULL, statut TEXT NOT NULL, debut TEXT NOT NULL, erreur TEXT ); """) return conn def validate(jour, rows): if date.fromisoformat(jour).isoformat() != jour: raise ValueError("Date attendue au format AAAA-MM-JJ") if not isinstance(rows, list) or not rows: raise ValueError("Snapshot complet non vide requis dans cet atelier") seen = set() for row in rows: if not isinstance(row, dict) or set(row) != {"jour", "commande", "centimes"}: raise ValueError("Colonnes attendues : jour, commande, centimes") if row["jour"] != jour: raise ValueError("La source contient une autre partition") if not isinstance(row["commande"], str) or not row["commande"].strip(): raise ValueError("Identifiant de commande requis") if row["commande"] in seen: raise ValueError("Commande dupliquée dans la source") if type(row["centimes"]) is not int or row["centimes"] < 0: raise ValueError("Montant entier positif ou nul requis") seen.add(row["commande"]) def snapshot(conn, jour): return [list(row) for row in conn.execute( "SELECT commande, centimes FROM ventes WHERE jour = ? ORDER BY commande", (jour,) )] def load(conn, jour, rows, fail=None): """Source = snapshot COMPLET de la journée. Aucune ingestion de delta.""" attempt = uuid.uuid4().hex source_hash = digest(rows) conn.execute("INSERT INTO tentatives VALUES (?, ?, ?, ?, ?, ?, ?)", ( attempt, jour, source_hash, CODE_VERSION, "started", datetime.now(timezone.utc).isoformat(), None, )) committed = False try: validate(jour, rows) conn.execute("BEGIN IMMEDIATE") conn.execute("DELETE FROM ventes WHERE jour = ?", (jour,)) if fail == "before_commit": raise RuntimeError("Panne simulée après DELETE, avant COMMIT") conn.executemany("INSERT INTO ventes VALUES (?, ?, ?)", [ (jour, row["commande"], row["centimes"]) for row in rows ]) conn.execute("""INSERT INTO partitions VALUES (?, ?, ?, ?, ?) ON CONFLICT(jour) DO UPDATE SET source_hash=excluded.source_hash, code_version=excluded.code_version, lignes=excluded.lignes, total_centimes=excluded.total_centimes""", ( jour, source_hash, CODE_VERSION, len(rows), sum(r["centimes"] for r in rows) )) conn.execute("UPDATE tentatives SET statut = 'committed' WHERE id = ?", (attempt,)) conn.execute("COMMIT") committed = True if fail == "after_commit": raise RuntimeError("Panne simulée après COMMIT, avant accusé de réception") return {"attempt": attempt, "source_hash": source_hash, "rows": snapshot(conn, jour)} except Exception as exc: if conn.in_transaction: conn.execute("ROLLBACK") conn.execute("UPDATE tentatives SET statut = ?, erreur = ? WHERE id = ?", ( "committed_ack_lost" if committed else "failed", str(exc), attempt, )) raise def demo(): day1, day2 = "2026-10-01", "2026-10-02" initial = [{"jour": day1, "commande": "A01", "centimes": 1000}, {"jour": day1, "commande": "A02", "centimes": 2500}] corrected = [{"jour": day1, "commande": "A01", "centimes": 1200}] second = [{"jour": day2, "commande": "B01", "centimes": 900}] checks = [] def verify(name, actual, expected): if actual != expected: raise AssertionError(f"{name}: {actual!r} != {expected!r}") checks.append({"test": name, "status": "pass", "observed": actual}) def expected_failure(fn, fragment): try: fn() except (ValueError, RuntimeError) as exc: if fragment not in str(exc): raise else: raise AssertionError("La panne attendue n'a pas été déclenchée") with tempfile.TemporaryDirectory(prefix="nymphar-reprise-") as tmp: conn = connect(Path(tmp) / "demo.sqlite") try: load(conn, day1, initial) baseline = snapshot(conn, day1) verify("premier_chargement", baseline, [["A01", 1000], ["A02", 2500]]) load(conn, day1, initial) verify("rejeu_identique", snapshot(conn, day1), baseline) load(conn, day2, second) expected_failure(lambda: load(conn, day1, corrected, "before_commit"), "avant COMMIT") verify("rollback_conserve_ancienne_partition", snapshot(conn, day1), baseline) verify("rollback_conserve_metadonnees", conn.execute( "SELECT total_centimes FROM partitions WHERE jour=?", (day1,)).fetchone()[0], 3500) expected_failure(lambda: load(conn, day1, corrected, "after_commit"), "après COMMIT") verify("ecriture_validee_malgre_accuse_perdu", snapshot(conn, day1), [["A01", 1200]]) load(conn, day1, corrected) verify("reprise_apres_commit_sans_doublon", snapshot(conn, day1), [["A01", 1200]]) verify("partition_voisine_preservee", snapshot(conn, day2), [["B01", 900]]) second_corrected = [{"jour": day2, "commande": "B01", "centimes": 950}] for jour, rows in [(day1, corrected), (day2, second_corrected)]: load(conn, jour, rows) verify("backfill_total_deux_jours", conn.execute( "SELECT SUM(centimes) FROM ventes").fetchone()[0], 2150) baseline2 = snapshot(conn, day1) invalid_inputs = [(corrected * 2, "dupliquée"), (second, "autre partition"), ([], "non vide"), ([{"jour": day1, "commande": "A01", "centimes": -1}], "Montant")] for rows, fragment in invalid_inputs: expected_failure(lambda: load(conn, day1, rows), fragment) verify("source_refusee_" + fragment, snapshot(conn, day1), baseline2) verify("hash_source_materialisee", conn.execute( "SELECT source_hash FROM partitions WHERE jour=?", (day1,)).fetchone()[0], digest(corrected)) history = dict(conn.execute("SELECT statut, COUNT(*) FROM tentatives GROUP BY statut")) verify("historique_conserve", history, {"committed": 6, "committed_ack_lost": 1, "failed": 5}) return {"synthetic": True, "code_version": CODE_VERSION, "sqlite_version": sqlite3.sqlite_version, "tested_at": datetime.now(timezone.utc).isoformat(), "checks": checks, "checks_passed": len(checks), "attempts": history, "final_rows": list(conn.execute("SELECT * FROM ventes ORDER BY jour, commande")), "scope": "SQLite local, exécution séquentielle, deux jours fictifs. Aucun benchmark d'orchestrateur."} finally: conn.close() if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--demo", action="store_true") parser.add_argument("--output", type=Path) parser.add_argument("--db", type=Path) parser.add_argument("--date") parser.add_argument("--source", type=Path) parser.add_argument("--fail", choices=["before_commit", "after_commit"]) args = parser.parse_args() if args.demo: result = demo() else: if not all([args.db, args.date, args.source]): parser.error("Utiliser --demo ou fournir --db, --date et --source") rows = json.loads(args.source.read_text(encoding="utf-8")) conn = connect(args.db) try: result = load(conn, args.date, rows, args.fail) finally: conn.close() rendered = json.dumps(result, ensure_ascii=False, indent=2) if args.output: args.output.write_text(rendered + "\n", encoding="utf-8") print(rendered)