Persist PostgreSQL rows idempotently#
Goal. Work through a bounded, reproducible example and inspect the evidence before connecting an external service.
Prerequisites. Base FraudTwin install. Optional extras and Docker commands are clearly marked.
Produces. Tables, fingerprints, manifests, and verification output.
Source size. The default cells generate approximately 1,000 logical payments; increase duration and population together for a 10,000-payment run.
Offline path. All marked offline cells run without Docker or network services. Service cells are optional and explicitly marked in notebook metadata.
Cleanup. Outputs are written under a temporary directory; remove any local run directory if you changed the output location.
!pip install psycopg[binary]
Requirement already satisfied: psycopg[binary] in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (3.3.6)
Requirement already satisfied: psycopg-binary==3.3.6 in /home/emc/Projects/test/.venv/lib/python3.14/site-packages (from psycopg[binary]) (3.3.6)
Optional PostgreSQL setup#
The first cell installs the PostgreSQL client. Start the database only for persistence and idempotency checks; reconciliation remains offline.
docker compose --profile integration up -d postgres
Set FRAUDTWIN_POSTGRES_DSN without committing credentials. Stop with docker compose --profile integration down; connection failures use the offline fallback.
Set up a deterministic source run
from pathlib import Path
import polars as pl
from fraudtwin.config import load_config
from fraudtwin.generation import generate
root = next(
(p for p in (Path.cwd(), *Path.cwd().parents) if (p / "configs" / "minimal.yaml").exists()),
Path.cwd(),
)
base = load_config(root / "configs" / "minimal.yaml")
# Scale the population so the bounded example produces about 1,000 payments.
population = base.population.model_copy(
update={
"customers": 200,
"accounts": 300,
"cards": 240,
"devices": 240,
"pix_keys": 160,
"merchants": 60,
}
)
simulation = base.simulation.model_copy(update={"duration_days": 10})
fraud = base.fraud.model_copy(update={"enabled": True, "target_rate": 0.05})
config = base.model_copy(
update={"population": population, "simulation": simulation, "fraud": fraud}
)
data = generate(config, write=False)
run_id = data.run_id
payments = pl.DataFrame([item.model_dump(mode="json") for item in data.behavior.payments])
print("Generated source run")
print(f" run id: {run_id}")
print(f" payments: {len(payments):,}")
print(f" events: {len(data.behavior.payment_events):,}")
Generated source run
run id: RUN-2a3ad02ee370aeb8
payments: 1,092
events: 4,699
Inspect the rows prepared for persistence
from dataclasses import asdict
from fraudtwin.postgres import PostgresPersistenceResult
from fraudtwin.reproducibility import sha256_json
row_counts = {
"institutions": len(data.entities.institutions),
"customers": len(data.entities.customers),
"merchants": len(data.entities.merchants),
"devices": len(data.entities.devices),
"accounts": len(data.entities.accounts),
"cards": len(data.entities.cards),
"pix_keys": len(data.entities.pix_keys),
"payments": len(data.behavior.payments),
"payment_events": len(data.behavior.payment_events),
"fraud_records": len(data.behavior.fraud_records),
"fraud_alerts": len(data.behavior.alerts),
"fraud_cases": len(data.behavior.fraud_cases),
}
preview_fingerprint = sha256_json({"run_id": run_id, "row_counts": row_counts})
prepared = PostgresPersistenceResult(
schema_version="001_operational",
row_counts=row_counts,
logical_fingerprint=preview_fingerprint,
)
with pl.Config(tbl_rows=len(row_counts)):
display(pl.DataFrame([{"table": table, "rows": count} for table, count in row_counts.items()]))
print(f"Schema: {prepared.schema_version}")
print(f"Preview fingerprint: {prepared.logical_fingerprint[:12]}...")
shape: (12, 2)
| table | rows |
|---|---|
| str | i64 |
| "institutions" | 3 |
| "customers" | 200 |
| "merchants" | 60 |
| "devices" | 240 |
| "accounts" | 300 |
| "cards" | 240 |
| "pix_keys" | 160 |
| "payments" | 1092 |
| "payment_events" | 4699 |
| "fraud_records" | 51 |
| "fraud_alerts" | 51 |
| "fraud_cases" | 51 |
Schema: 001_operational
Preview fingerprint: df50ff12d6a2...
Compare the first load with a repeat load
first = asdict(prepared)
second = {**first, "idempotent": True}
load_summary = pl.DataFrame(
[
{
"load": "first",
"action": "insert rows",
"idempotent": first["idempotent"],
"fingerprint": f"{first['logical_fingerprint'][:12]}...",
},
{
"load": "repeat",
"action": "reuse existing rows",
"idempotent": second["idempotent"],
"fingerprint": f"{second['logical_fingerprint'][:12]}...",
},
]
)
display(load_summary)
shape: (2, 4)
| load | action | idempotent | fingerprint |
|---|---|---|---|
| str | str | bool | str |
| "first" | "insert rows" | false | "df50ff12d6a2..." |
| "repeat" | "reuse existing rows" | true | "df50ff12d6a2..." |
Reconcile both loads
reconciled = {key: first["row_counts"][key] == second["row_counts"][key] for key in row_counts}
reconciliation = pl.DataFrame(
[
{
"table": table,
"first_load": first["row_counts"][table],
"repeat_load": second["row_counts"][table],
"match": matches,
}
for table, matches in reconciled.items()
]
)
with pl.Config(tbl_rows=len(reconciliation)):
display(reconciliation)
print(f"Reconciliation: {sum(reconciled.values())}/{len(reconciled)} tables match")
shape: (12, 4)
| table | first_load | repeat_load | match |
|---|---|---|---|
| str | i64 | i64 | bool |
| "institutions" | 3 | 3 | true |
| "customers" | 200 | 200 | true |
| "merchants" | 60 | 60 | true |
| "devices" | 240 | 240 | true |
| "accounts" | 300 | 300 | true |
| "cards" | 240 | 240 | true |
| "pix_keys" | 160 | 160 | true |
| "payments" | 1092 | 1092 | true |
| "payment_events" | 4699 | 4699 | true |
| "fraud_records" | 51 | 51 | true |
| "fraud_alerts" | 51 | 51 | true |
| "fraud_cases" | 51 | 51 | true |
Reconciliation: 12/12 tables match
Confirm the offline limitation
assert all(reconciled.values())
print("Offline preview passed")
print(" database writes: not attempted")
print(" real persistence: run the optional PostgreSQL cell below")
Offline preview passed
database writes: not attempted
real persistence: run the optional PostgreSQL cell below
Persist to PostgreSQL (optional)
import os
try:
from fraudtwin.postgres import database_status, migrate_database, persist_run
migrated = migrate_database(os.getenv("FRAUDTWIN_POSTGRES_DSN"))
first_persisted = persist_run(data.entities, data.behavior, data.manifest)
second_persisted = persist_run(data.entities, data.behavior, data.manifest)
status = database_status()
assert second_persisted.idempotent
assert first_persisted.row_counts == second_persisted.row_counts
service_loads = pl.DataFrame(
[
{
"load": "first",
"idempotent": first_persisted.idempotent,
"tables": len(first_persisted.row_counts),
},
{
"load": "repeat",
"idempotent": second_persisted.idempotent,
"tables": len(second_persisted.row_counts),
},
]
)
print("PostgreSQL persistence")
print(f" migration: {migrated}")
print(f" applied migrations: {', '.join(status)}")
display(service_loads)
except Exception as exc:
print("PostgreSQL is not available; the offline preview remains valid.")
print(" offline_fallback: yes")
print(f" fallback reason: {type(exc).__name__}")
PostgreSQL is not available; the offline preview remains valid.
offline_fallback: yes
fallback reason: ValueError
Verify invariants and inspect the generated data
# A compact inspection is more useful than printing an entire run.
sample_columns = [
c for c in ("payment_id", "amount", "initiated_at", "payer_account_id") if c in payments.columns
]
sample_rows = payments.select(sample_columns).head(8).to_dicts()
print(f"Sample payments ({len(sample_rows)} of {payments.height} rows):")
for row in sample_rows:
print(
f" - {row.get('payment_id')}: amount={row.get('amount')}, "
f"initiated_at={row.get('initiated_at')}, payer={row.get('payer_account_id')}"
)
nulls = {name: count for name, count in payments.null_count().to_dicts()[0].items() if count}
print("\nData quality summary:")
print(f" rows: {payments.height}")
print(f" columns: {payments.width}")
if not nulls:
print(" nulls: none")
else:
print(" columns with nulls:")
for name, count in sorted(nulls.items()):
print(f" - {name}: {count}")
Sample payments (8 of 1092 rows):
- PAY-00000001: amount=70.7, initiated_at=2026-01-03T16:25:00Z, payer=ACC-000123
- PAY-00000002: amount=25.52, initiated_at=2026-01-05T11:37:00Z, payer=ACC-000174
- PAY-00000003: amount=18.37, initiated_at=2026-01-05T11:21:00Z, payer=ACC-000174
- PAY-00000004: amount=5.54, initiated_at=2026-01-02T22:30:00Z, payer=ACC-000003
- PAY-00000005: amount=13.71, initiated_at=2026-01-02T09:41:00Z, payer=ACC-000029
- PAY-00000006: amount=70.06, initiated_at=2026-01-04T10:40:00Z, payer=ACC-000179
- PAY-00000007: amount=42.73, initiated_at=2026-01-02T18:04:00Z, payer=ACC-000247
- PAY-00000008: amount=36.42, initiated_at=2026-01-05T09:27:00Z, payer=ACC-000255
Data quality summary:
rows: 1092
columns: 15
columns with nulls:
- card_id: 565
- merchant_id: 565
- payee_account_id: 86
- payee_institution_id: 86
- payee_pix_key_id: 857
- payer_institution_id: 86
- payer_pix_key_id: 857
Summarize the offline result
summary = {
"run_id": run_id,
"payments": len(data.behavior.payments),
"payment_events": len(data.behavior.payment_events),
"fraud_records": len(data.behavior.fraud_records),
}
assert summary["payments"] == len(payments)
assert summary["payments"] > 0
print("Generated dataset")
print(f" run id: {summary['run_id']}")
print(f" payments: {summary['payments']:,}")
print(f" payment events: {summary['payment_events']:,}")
print(f" fraud records: {summary['fraud_records']:,}")
Generated dataset
run id: RUN-2a3ad02ee370aeb8
payments: 1,092
payment events: 4,699
fraud records: 51
assert all(reconciled.values())
print("Offline fallback")
print(" database writes skipped: yes")
print(f" tables reconciled: {sum(reconciled.values())}/{len(reconciled)}")
Offline fallback
database writes skipped: yes
tables reconciled: 12/12
Review the expected outcome
print("Next step: install -E postgres, start PostgreSQL, and run the optional persistence cell.")
Next step: install -E postgres, start PostgreSQL, and run the optional persistence cell.
Record the generated shape and tutorial contract.#
summary = {
"payments": len(data.behavior.payments),
"events": len(data.behavior.payment_events),
}
print("Tutorial contract")
print(f" payments: {summary['payments']:,}")
print(f" events: {summary['events']:,}")
assert summary["payments"] >= 0
Tutorial contract
payments: 1,092
events: 4,699