#!/usr/bin/env python3 """Original LockedIn Labs workshop: establish a complete credit extract. Python 3.11+, standard library. All requests call a local fixture, not a network. Keep fde-financial-reconciliation.py beside this file. No financial writes. The scope echo and pinned snapshot are this fictional provider's contract; they are NOT guarantees of pagination generally or a real provider adapter. """ from __future__ import annotations import argparse from copy import deepcopy from dataclasses import asdict import json from pathlib import Path import runpy SCOPE = {"workspace": "SYNTHETIC-WORKSPACE", "case_id": "SYNTHETIC-CASE-104", "account_id": "SYNTHETIC-ACCOUNT-7", "transaction_id": "SYNTHETIC-DEBIT-81"} SNAPSHOT = "SYNTHETIC-SNAPSHOT-7" ENTRY = {**SCOPE, "entry_id": "SYNTHETIC-CREDIT-9", "amount_minor": 12500, "currency": "USD", "kind": "provisional", "status": "posted", "direction": "credit"} def collect(fetch_page, scope: dict, *, max_pages: int = 5, stop_after_first: bool = False, allow_mixed_snapshots: bool = False) -> dict: """Read one scoped snapshot or withhold the entire extract. fetch_page receives (scope, opaque_cursor, pinned_snapshot). Empty cursor starts the collection; None snapshot asks the fictional provider to pin one. Later requests must keep both scope and snapshot. Retries are not automatic. """ trace, observed, seen_cursors = [], [], set() cursor, snapshot = "", None def finish(reason: str, complete: bool = False) -> dict: return {"complete": complete, "reason": reason, "observed_rows": len(observed), "requests": len(trace), "extract": {"entries": deepcopy(observed), "complete": True, "reference": snapshot} if complete else None, "trace": trace} if (type(max_pages) is not int or max_pages < 1 or not isinstance(scope, dict) or set(scope) != set(SCOPE) or any(type(value) is not str or not value for value in scope.values())): return finish("invalid_request") scope = deepcopy(scope) for _ in range(max_pages): if cursor in seen_cursors: return finish("cursor_cycle") seen_cursors.add(cursor) request = {"scope": deepcopy(scope), "cursor": cursor, "snapshot": snapshot} step = {"request": request} trace.append(step) try: response = fetch_page(deepcopy(scope), cursor, snapshot) except (TimeoutError, ConnectionError): step["error"] = "transport_unavailable" return finish("transport_unavailable") step["response"] = deepcopy(response) if not isinstance(response, dict) or type(response.get("status")) is not int: return finish("invalid_response") if response["status"] != 200: return finish("cursor_expired" if response["status"] == 410 else "page_unavailable") if response.get("scope") != scope: return finish("scope_changed") version = response.get("snapshot") if type(version) is not str or not version: return finish("snapshot_missing") if snapshot is None: snapshot = version elif version != snapshot and not allow_mixed_snapshots: return finish("snapshot_changed") rows, next_cursor = response.get("entries"), response.get("next_cursor") if not isinstance(rows, list) or type(next_cursor) is not str: return finish("invalid_page") for row in rows: if (not isinstance(row, dict) or set(row) != set(ENTRY) or type(row["amount_minor"]) is not int or row["amount_minor"] <= 0 or any(type(row[key]) is not str or not row[key] for key in ENTRY if key != "amount_minor")): return finish("invalid_entry") if any(row[key] != scope[key] for key in scope): return finish("entry_scope_changed") observed.extend(deepcopy(rows)) if stop_after_first: return finish("first_page_assumed_complete", True) if next_cursor == "": return finish("end_of_snapshot", True) if next_cursor in seen_cursors: return finish("cursor_cycle") cursor = next_cursor return finish("page_budget_exhausted") def page(entries: list, next_cursor: str = "", **overrides) -> dict: return {"status": 200, "scope": deepcopy(SCOPE), "snapshot": SNAPSHOT, "entries": deepcopy(entries), "next_cursor": next_cursor, **overrides} def scenarios() -> list[dict]: """Each map key is an opaque cursor in this invented provider fixture.""" first = page([ENTRY], "SYNTHETIC-CURSOR-A") return [ {"name": "credit-on-page-two", "pages": { "": page([], "SYNTHETIC-CURSOR-A"), "SYNTHETIC-CURSOR-A": page([ENTRY])}}, {"name": "second-credit-on-page-two", "pages": { "": first, "SYNTHETIC-CURSOR-A": page([{**ENTRY, "entry_id": "SYNTHETIC-CREDIT-10"}])}}, {"name": "terminal-empty", "pages": {"": page([])}}, {"name": "empty-middle-page", "pages": { "": page([], "SYNTHETIC-CURSOR-A"), "SYNTHETIC-CURSOR-A": page([], "SYNTHETIC-CURSOR-B"), "SYNTHETIC-CURSOR-B": page([ENTRY])}}, {"name": "failed-later-page", "pages": { "": first, "SYNTHETIC-CURSOR-A": {"status": 503}}}, {"name": "changed-snapshot", "pages": { "": first, "SYNTHETIC-CURSOR-A": page([], snapshot="SYNTHETIC-SNAPSHOT-8")}}, {"name": "changed-scope", "pages": { "": first, "SYNTHETIC-CURSOR-A": page([], scope={**SCOPE, "account_id": "SYNTHETIC-OTHER"})}}, {"name": "repeated-cursor", "pages": { "": first, "SYNTHETIC-CURSOR-A": page([], "SYNTHETIC-CURSOR-A")}}, {"name": "page-budget", "max_pages": 2, "pages": { "": first, "SYNTHETIC-CURSOR-A": page([], "SYNTHETIC-CURSOR-B"), "SYNTHETIC-CURSOR-B": page([])}}, {"name": "malformed-continuation", "pages": {"": page([ENTRY], None)}}, {"name": "expired-cursor", "pages": { "": first, "SYNTHETIC-CURSOR-A": {"status": 410}}}, {"name": "wrong-account-row", "pages": { "": page([{**ENTRY, "account_id": "SYNTHETIC-OTHER"}])}}, {"name": "invalid-money-shape", "pages": {"": page([{**ENTRY, "amount_minor": True}])}}, ] def run_case(scenario: dict, reconciliation: dict, *, mode: str = "guarded") -> dict: source = deepcopy(scenario["pages"]) def fetch_page(scope, cursor, snapshot): # The collector's request is recorded; this local fixture never sends it. return deepcopy(source.get(cursor, {"status": 404})) gathered = collect(fetch_page, SCOPE, max_pages=scenario.get("max_pages", 5), stop_after_first=mode == "stop_after_first", allow_mixed_snapshots=mode == "allow_mixed_snapshots") matched = None if gathered["complete"]: extract = gathered["extract"] matched = reconciliation["reconcile"]( reconciliation["Case"](), reconciliation["Extract"]( entries=tuple(reconciliation["Entry"](**row) for row in extract["entries"]), complete=True, reference=extract["reference"])) return {"name": scenario["name"], "source_pages": source, "collection": gathered, "reconciliation": matched, "handoff": {"owner": matched["owner"] if matched else "integration_operations", "next_action": matched["next_action"] if matched else "Resolve collection coverage or the provider contract, then start a fresh scoped run.", "investigation_state": reconciliation["Case"]().investigation_state}} def report(mode: str = "guarded") -> dict: companion = Path(__file__).with_name("fde-financial-reconciliation.py") if not companion.is_file(): raise SystemExit("Keep fde-financial-reconciliation.py in the same folder as this lab.") reconciliation = runpy.run_path(str(companion)) return {"synthetic": True, "execution": "local fixture calls; no HTTP requests", "mode": mode, "case": asdict(reconciliation["Case"]()), "cases": [run_case(item, reconciliation, mode=mode) for item in scenarios()]} def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--json", action="store_true") experiments = parser.add_mutually_exclusive_group() experiments.add_argument("--stop-after-first", action="store_true", help="Deliberate defect: publish page one as a complete extract") experiments.add_argument("--allow-mixed-snapshots", action="store_true", help="Deliberate defect: accept a different source snapshot") args = parser.parse_args() mode = "stop_after_first" if args.stop_after_first else "allow_mixed_snapshots" if args.allow_mixed_snapshots else "guarded" output = report(mode) if args.json: print(json.dumps(output, indent=2)) return print(f"Synthetic paginated extract workshop | {mode}") print("Local fixture calls only. Exit 0 means a report was produced, not that a learner passed.\n") for item in output["cases"]: gathered, matched = item["collection"], item["reconciliation"] print(f"{item['name']}: {gathered['requests']} request(s) | {gathered['reason']} | " f"{matched['reason'] if matched else 'reconciliation not called'}") if __name__ == "__main__": main()