"""repository.py — load a whole aggregate in one go; save it whole, idempotently.
An AGGREGATE is the unit a use case loads, decides about, and saves: a quote with its versions,
lines and revenue schedule. The two rules come from an earlier CRM quote repository, shaped by what DataK3 actually does:
1. **One load pulls the whole aggregate.** No lazy loading: a fetch that happens after a write in the
same transaction can come back stale, because there is no read-your-writes inside an open
transaction. One small indexed SELECT per table, every one scoped by the root's key — not one big
JOIN, which multiplies versions by lines by months and has to be de-duplicated in Python.
2. **One save makes storage match memory, and running it twice changes nothing.** No change
tracking: the in-memory aggregate IS the intended state. Rows are written with
`INSERT … ON CONFLICT DO UPDATE`, never `session.merge()` (a SELECT then a write — two round trips
that lean on reading back a row just written). Rows that are no longer in memory are deleted.
Two defects found in the code these rules came from, which this base makes impossible rather than
merely discouraged (verified 2026-10-16 by reading that repository and the platform's db.upsert):
* **Stale rows below the first level survived.** That save deleted removed LINES but never their
revenue-schedule rows, nor schedule months dropped from a line that stayed. DataK3 enforces no
foreign key, so nothing cascades; the rows sat in a table that looks authoritative. Here every
part's key starts with the root's key, so "everything stored under this root" is one indexed
filter, and a stale row is found at any depth — under a removed parent, or trimmed from a kept one.
* **Creation columns were overwritten on every save.** An upsert that refreshes "every non-key
column" rewrites `created_at` / `created_by`, so after the first edit they mean "last saved".
Columns named in `insert_only` are written for a new row only — see upsert() for why that takes
a key lookup on DataK3 rather than a narrower SET clause.
A REPOSITORY takes an open connection; the caller owns the transaction, because a use case often
writes two aggregates atomically and a repository that opened its own connection would forbid it.
RACE, stated rather than hidden: stale rows are found by reading the keys stored under the root at
the start of the save. A row committed by ANOTHER connection in the instant before can be missed
and survive until the next save of the same aggregate. The rows a save writes are never affected.
OPTIMISTIC VERSIONS (opt-in: `Aggregate(..., version="version")`). Without them the last writer wins:
two HR admins open one person, both save, and the first one's change is silently gone. With them, `load`
returns the root's `version` and `save(..., expected_version=v)` writes `v + 1` only if nobody saved in
between; otherwise it raises `Conflict` ("changed since you opened it"), which both doors answer as 409.
How, and why not `UPDATE … WHERE version = :v`. DataK3 has no read-your-writes inside a transaction, so
"read the version, compare, write" in one transaction compares against a snapshot two racers share. What
the engine DOES guarantee, measured on erp-platform-test (2026-10-17, scripts/live_version_check.py):
* a plain INSERT of a primary key that another transaction has committed — or commits first — is
rejected 23505, at the statement or at COMMIT. Ten concurrent transactions inserting one key: one
commits, nine fail 23505; the loser's other writes in the same transaction are rolled back with it.
* `UPDATE … WHERE version = :v` also lets exactly one of ten commit (the rest 40001 at COMMIT), but every
one of the ten was told rowcount 1 before committing — so the statement cannot be the decision, and
40001 is a "retry me" code (erp_platform.retry), not a "you lost" code.
So every save CLAIMS its new version by a plain INSERT into `<root table>_versions` (root key + version,
both in the primary key), in the SAME transaction as the rows it writes. Two saves from version 3 both try
to insert (key, 4); one commits, the other's whole transaction fails 23505 → Conflict. The capability
claim stands on the same measured property. The ordinary stale case is refused before anything is written: against
the load's snapshot (0.9.13: no statement), or, for a save without one, by a SELECT of the committed root.
`version_ddl()` prints the claim table for schema.sql.
"""
import copy
import os
import re
import threading
from collections import defaultdict
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass
from datetime import date, datetime, timezone
from typing import Any, Optional
from sqlalchemy import Column, Integer, Table, TIMESTAMP, and_, delete, event, insert, or_, select
from sqlalchemy.engine import Engine
from sqlalchemy.dialects import postgresql
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.exc import IntegrityError
INSERT_ONLY = ("created_at", "created_by")
VERSIONS_SUFFIX = "_versions"
CONFLICT_REASON = "changed since you opened it — reload it and make your change again"
class Conflict(Exception):
"""Someone else saved this aggregate after it was loaded. Both doors answer 409 with the reason.
`expected` is the version the caller loaded; `stored` the one found, when it is known (a race lost at
COMMIT says only that the version was taken)."""
def __init__(self, table: str = "", key: Optional[dict] = None, expected: Optional[int] = None,
stored: Optional[int] = None):
super().__init__(CONFLICT_REASON)
self.reason, self.table, self.key, self.expected, self.stored = CONFLICT_REASON, table, key, expected, stored
def _chain(exc: BaseException):
seen = set()
while exc is not None and id(exc) not in seen:
seen.add(id(exc))
yield exc
exc = getattr(exc, "orig", None) or exc.__cause__ or exc.__context__
def as_conflict(exc: BaseException) -> Optional[Conflict]:
"""The Conflict an exception means, or None.
A version claim lost to a racer can fail at the INSERT or only at COMMIT (measured: DataK3 checks a
staged key against another transaction's at commit) — and the commit belongs to the caller's
`with engine.begin()`, outside save(). So a duplicate key on a `*_versions` table is recognised
wherever it surfaces: SQLSTATE 23505 (or SQLite's UNIQUE failure) naming a versions table."""
from erp_platform import retry
for e in _chain(exc):
if isinstance(e, Conflict):
return e
text = " ".join(str(e) for e in _chain(exc)).lower()
duplicate = retry.sqlstate(exc) == "23505" or "unique constraint failed" in text or "duplicate key" in text
if duplicate and (VERSIONS_SUFFIX in text or any(t in text for t in CLAIM_TABLES)):
return Conflict()
return None
# Tables a brick claims with a plain INSERT (number series, allocation and guard claims, one-row-per-event keys). On DataK3 a
# racer's duplicate key on one of them surfaces at COMMIT, outside the use case, exactly like a version claim: without this the
# door answers 500 instead of the retry-safe 409 (measured on the pg wire 2026-10-24, a payments allocation race).
CLAIM_TABLES: set = set()
def claim_tables(*names: str) -> None:
"""Declare the brick's claim tables, once at import: `repository.claim_tables("pay_source_claim", "pay_claim")`.
A duplicate key naming one of them, wherever it surfaces (the INSERT or the COMMIT), is a Conflict (409)."""
for n in names:
if not n or not n.replace("_", "").isalnum():
raise ValueError(f"a claim table is a plain table name, got {n!r}")
CLAIM_TABLES.add(n.lower())
@contextmanager
def conflicts():
"""Turn a version claim lost at COMMIT into Conflict. Wrap the unit of work, commit included:
with repository.conflicts(), engine.begin() as conn:
PERSON.save(conn, rows, expected_version=v)
A use case run through a brick's doors needs none of this: the doors do it (erp_platform.bricks)."""
try:
yield
except Conflict:
raise
except Exception as exc: # noqa: BLE001 — re-raised unless it is a lost claim
c = as_conflict(exc)
if c is None:
raise
raise c from None
def retry_on_conflict(fn, *, attempts: int = 10, sleep=None):
"""Run `fn` (a whole unit of work: load, decide, save, COMMIT) again when its version claim lost a race.
For the caller who sent NO version: they asked for "this change, on whatever is there now", so a lost
race is re-run on the latest state — both changes land, none is lost (the CRM audit, 2026-10-17: two
people editing one deal at once, 6 of 6 races lost an edit). For a caller who sent the version they
loaded, never use this: their 409 "changed since you opened it" is the answer, because they decided on
what they saw. Safe to re-run because a lost claim rolls the WHOLE transaction back (measured on DataK3:
the loser's other writes go with it). Backoff is jittered, so racers do not collide again in step."""
import random
import time
wait = sleep or time.sleep
for attempt in range(1, attempts + 1):
try:
return fn()
except Exception as exc: # noqa: BLE001 — re-raised unless a lost version claim
if as_conflict(exc) is None or attempt == attempts:
raise
wait(min(1.0, 0.05 * 2 ** (attempt - 1)) * (0.5 + random.random()))
raise AssertionError("unreachable")
def _table(t: Any) -> Table:
"""Accept a Core Table or an ORM model."""
return getattr(t, "__table__", t)
def _versions_table(root: Table, key: tuple, version: str) -> Table:
"""`<root>_versions`: the root's key + the version, all of it the primary key. In the root's MetaData,
so a brick's `md.create_all` (its tests) creates it with the rest."""
name = root.name + VERSIONS_SUFFIX
if name in root.metadata.tables:
return root.metadata.tables[name]
cols = [Column(k, root.c[k].type, primary_key=True, autoincrement=False, nullable=False) for k in key]
cols += [Column(version, Integer, primary_key=True, autoincrement=False, nullable=False),
Column("saved_at", TIMESTAMP, nullable=False)]
return Table(name, root.metadata, *cols)
def _check_key_types(part_name: str, table: Table, key: tuple, row: dict) -> None:
"""A key value must already be the column's Python type.
Stale rows are found by comparing stored keys with in-memory keys. The database hands back a
DATE as a `date`; if memory holds "2026-01" the two never compare equal, and the save would
delete the very rows it had just written. So a mistyped key fails here, before any write.
"""
for k in key:
try:
expected = table.c[k].type.python_type
except NotImplementedError:
continue
v = row.get(k)
wrong = v is None or not isinstance(v, expected) or (
expected is date and isinstance(v, datetime)) # a datetime is not a DATE
if wrong:
raise TypeError(f"{part_name}.{k} is {v!r} ({type(v).__name__}); the column holds "
f"{expected.__name__}, and a key of another type would never match "
"what is stored — the save would delete the rows it just wrote")
def _existing_keys(conn: Any, table: Table, keys: list, rows: list) -> set:
"""Which of these rows' keys are already stored (committed)."""
wanted = {tuple(r[k] for k in keys) for r in rows}
cond = or_(*(and_(*(table.c[k] == v for k, v in zip(keys, kv))) for kv in sorted(wanted, key=repr)))
return {tuple(r) for r in conn.execute(select(*(table.c[k] for k in keys)).where(cond))}
def upsert(conn: Any, table: Any, rows: list, key: Any, insert_only: tuple = INSERT_ONLY,
existing: Optional[set] = None, *, new: bool = False) -> None:
"""Idempotent keyed write. Each row's outcome depends on that row alone.
Rows are grouped by the columns they name and written one statement per group, so a row only
ever touches the columns IT names. The earlier form padded every row to the batch's column set
with None — which turned "this row says nothing about phone" into "set phone to NULL" whenever a
different row in the same batch did mention phone.
INSERT-ONLY COLUMNS (created_at, created_by) are written when the row is new and left alone
after. On DataK3 that cannot be done in the SET clause: `ON CONFLICT DO UPDATE` writes EVERY
column named in VALUES and ignores the SET list (verified live 2026-10-16 on erp-platform-test —
`SET name = excluded.name` also overwrote phone and created_at; Postgres would keep both). What
the engine does honour is a column ABSENT from VALUES: it is kept. So the keys that already
exist are looked up first (or passed in as `existing`), and those rows are written without their
insert-only columns. The SET clause is still generated correctly, so this stays right on
Postgres and after the engine is fixed.
The lookup sees committed rows only (no read-your-writes inside a transaction): a row inserted
earlier in the SAME transaction counts as new, and its creation columns are written again with
this save's values.
`new=True` (0.9.13): the caller KNOWS these keys are not stored yet (an id minted for this write, a row keyed by a
claim this transaction won): a plain INSERT, no lookup. A key that is stored after all is refused as a Conflict
(409), never silently overwritten.
"""
table = _table(table)
keys = [key] if isinstance(key, str) else list(key)
if not rows:
return
if new:
# A plain INSERT: a key that IS stored is refused (23505), never overwritten — so a wrong "new" cannot rewrite a
# row's creation columns. At the statement it is a Conflict here; on DataK3 a racer's duplicate can surface at
# COMMIT instead, which is a Conflict where the table is declared with claim_tables().
groups_new: dict = defaultdict(list)
for row in rows:
groups_new[tuple(sorted(row))].append(row)
for group in groups_new.values():
try:
conn.execute(insert(table).values(group))
except IntegrityError as exc:
from erp_platform import retry
text_ = " ".join(str(e) for e in _chain(exc)).lower()
if retry.sqlstate(exc) == "23505" or "unique constraint failed" in text_ or "duplicate key" in text_:
raise Conflict(table.name) from None
raise
return
if existing is None and any(c in r for r in rows for c in insert_only):
existing = _existing_keys(conn, table, keys, rows)
existing = existing or set()
groups: dict = defaultdict(list)
for row in rows:
if tuple(row[k] for k in keys) in existing:
row = {c: v for c, v in row.items() if c not in insert_only}
groups[tuple(sorted(row))].append(row)
for cols, group in groups.items():
stmt = pg_insert(table).values(group)
updatable = [c for c in cols if c not in keys and c not in insert_only]
if updatable:
stmt = stmt.on_conflict_do_update(
index_elements=keys, set_={c: getattr(stmt.excluded, c) for c in updatable})
else:
# Only keys (an edge or junction row) or only insert-only columns: nothing to update.
stmt = stmt.on_conflict_do_nothing(index_elements=keys)
conn.execute(stmt)
@dataclass(frozen=True)
class Part:
"""One table of an aggregate, with its FULL primary key."""
table: Any
key: tuple
insert_only: tuple = INSERT_ONLY
@property
def name(self) -> str:
return _table(self.table).name
class Aggregate:
"""The tables of one aggregate, all scoped by the root's key.
QUOTE = Aggregate(
Part(quotes, ("quote_id",)),
Part(quote_lines, ("quote_id", "version", "line_no")),
Part(quote_schedule, ("quote_id", "version", "line_no", "month")))
Every part's key must START with the root's key. That is not decoration: it is what makes
"everything stored under this quote" one indexed filter, and therefore what lets a save find a
stale row at any depth.
"""
def __init__(self, root: Part, *children: Part, version: Optional[str] = None):
self.root, self.children = root, children
self.root_key = tuple(root.key)
for part in children:
if tuple(part.key[:len(self.root_key)]) != self.root_key:
raise ValueError(f"{part.name}: its key {part.key} must start with the root key "
f"{self.root_key} — that is how a save finds stale rows under it")
# Optimistic versions, opt-in: the root's integer column, and the claim table beside it.
self.version = version
self.versions: Optional[Table] = None
if version:
rt = _table(root.table)
if version not in rt.c:
raise ValueError(f"{rt.name} has no column {version!r} to carry the aggregate's version")
if version in self.root_key:
raise ValueError("the version column cannot be part of the root key")
self.versions = _versions_table(rt, self.root_key, version)
def version_ddl(self) -> str:
"""The claim table's CREATE for the brick's schema.sql (and an ALTER for the root's column, if the
root table predates it — add that yourself, at the END of its CREATE as well)."""
if self.versions is None:
raise ValueError("this aggregate is not versioned")
t, dialect = self.versions, postgresql.dialect()
cols = [f" {c.name} {c.type.compile(dialect=dialect)} NOT NULL," for c in t.c if c.name != "saved_at"]
return (f"-- Optimistic versions (erp_platform.repository): a save CLAIMS version n+1 with a plain INSERT here,\n"
f"-- in the same transaction as its rows; a racer from the same version fails 23505 -> 409.\n"
f"CREATE TABLE IF NOT EXISTS {t.name} (\n" + "\n".join(cols)
+ f"\n saved_at {t.c.saved_at.type.compile(dialect=dialect)} NOT NULL,"
+ f"\n PRIMARY KEY ({', '.join(c.name for c in t.primary_key.columns)})\n);\n")
def version_of(self, loaded: Optional[dict]) -> int:
"""The version a load returned (0 for nothing loaded) — what to pass back as expected_version."""
if not loaded or not self.version:
return 0
return int(loaded[self.root.name][0].get(self.version) or 0)
@property
def parts(self) -> tuple:
return (self.root, *self.children)
def _scope(self, part: Part, root_key: dict):
table = _table(part.table)
return and_(*(table.c[k] == root_key[k] for k in self.root_key))
def _select(self, part: Part, root_keys: list):
table = _table(part.table)
if len(root_keys) == 1:
return select(table).where(self._scope(part, root_keys[0]))
return select(table).where(_any_key(table, self.root_key, [tuple(k[c] for c in self.root_key) for k in root_keys]))
def load(self, conn: Any, root_key: dict) -> Optional["Loaded"]:
"""The whole aggregate, or None if the root is not there. One SELECT per table — sent together in ONE
round trip on psycopg 3 (pipeline mode), one after another elsewhere. The answer is a dict as it always
was, carrying `.snapshot`: what was read, to hand back to `save(..., snapshot=...)` so the save writes
only what changed."""
return self.load_many(conn, [root_key])[tuple(root_key[k] for k in self.root_key)]
def load_many(self, conn: Any, root_keys: list, *, chunk: int = 100) -> dict:
"""Many aggregates in one go: one SELECT per table per `chunk` roots (an IN list costs what one key
costs on the pg wire, measured 2026-10-24), all in one round trip on psycopg 3.
Returns {root key tuple: Loaded or None}, in the order the keys were given.
Inside a request (`request_scope`, which both doors open per attempt) an aggregate is read ONCE per key: a
second load answers a copy of the first one's rows and snapshot, until this request saves or removes it,
writes one of its tables, calls a neighbour's command, or rolls back (see the identity map below)."""
keys, seen, ordered = [], set(), []
for k in root_keys:
t = tuple(k[c] for c in self.root_key)
if t not in seen:
seen.add(t)
ordered.append(t)
keys.append({c: k[c] for c in self.root_key})
if not keys:
return {}
tree, who, bind = current_tree(), _who(), _bind_of(conn)
use = tree is not None and who is not None and bind is not None
cached: dict = {}
if use:
for k in keys:
t = tuple(k[c] for c in self.root_key)
hit = tree.entries.get(("agg", id(self), who, bind, t), _MISS)
if hit is not _MISS:
cached[t] = hit
keys = [k for k in keys if tuple(k[c] for c in self.root_key) not in cached]
out = self._read_many(conn, keys, chunk) if keys else {}
if not use:
return out
if not _wrote(conn):
_AGGREGATES[id(self)] = self
names = [p.name for p in self.parts] + ([self.versions.name] if self.versions is not None else [])
for t, got in out.items():
tree.put(("agg", id(self), who, bind, t), _copy_loaded(got), names)
out.update(cached)
# Every answer is a COPY: a use case edits its rows in place, and the next load in the request must still
# answer what is stored, not that edit.
return {t: (out[t] if t not in cached else _copy_loaded(out[t])) for t in ordered}
def _read_many(self, conn: Any, keys: list, chunk: int) -> dict:
chunks = [keys[i:i + chunk] for i in range(0, len(keys), chunk)]
stmts = [(part, self._select(part, ch)) for ch in chunks for part in self.parts]
results = _read_all(conn, [s for _, s in stmts])
grouped: dict = {tuple(k[c] for c in self.root_key): {p.name: [] for p in self.parts} for k in keys}
for (part, _), part_rows in zip(stmts, results):
if len(grouped) == 1: # one root: every row is its own, as load() always read them
next(iter(grouped.values()))[part.name].extend(part_rows)
continue
for row in part_rows:
t = tuple(row[c] for c in self.root_key)
if t in grouped:
grouped[t][part.name].append(row)
out: dict = {}
for t, parts in grouped.items():
if not parts[self.root.name]:
out[t] = None
continue
if self.version:
# A row written before the aggregate opted in has no version: it is version 0.
root = parts[self.root.name][0]
root[self.version] = int(root.get(self.version) or 0)
out[t] = Loaded(parts, Snapshot.of(self, dict(zip(self.root_key, t)), parts))
return out
def nothing_stored(self, root_key: dict) -> "Snapshot":
"""The snapshot of an aggregate that is not stored yet: `save(conn, rows, snapshot=AGG.nothing_stored(key))`
writes a NEW aggregate without first reading what is under its key. Only for a key minted for this save
(a versioned aggregate's claim still refuses a racer)."""
return Snapshot.of(self, {k: root_key[k] for k in self.root_key}, {p.name: [] for p in self.parts})
def save(self, conn: Any, rows: dict, *, now: Optional[datetime] = None,
expected_version: Optional[int] = None, snapshot: Any = None) -> Optional[int]:
"""Make storage match `rows` ({table name: [row, ...]}). Running it twice changes nothing.
`now` is read once and stamped on every row, so one save is one moment rather than a spread
of microseconds that turns a single action into an ordering question.
A VERSIONED aggregate needs `expected_version` — what `load` returned, 0 for a new one — and
returns the version it wrote (expected + 1), or raises Conflict. Running the same save twice is
then a Conflict the second time, which is the point: the second one did not see the first.
`snapshot` (0.9.6) — what `load` returned (or its `.snapshot`, or `nothing_stored(key)` for a new one).
With it the save does not re-read the keys under the root and writes only the DIFF: rows that are new
or whose columns changed, one multi-row upsert per table and column set, and ONE delete per table for
the rows that went. An unchanged unversioned aggregate is zero statements; a versioned one is the
version check, the claim, and the root's new version (3), because the claim IS the race guard and a
use case may lean on "nobody changed this since I read it" to decide something it writes elsewhere.
Rows unchanged since the load are not written, so a concurrent edit to one of them is not overwritten
either. Without a snapshot the save behaves exactly as before.
"""
root_rows = rows.get(self.root.name) or []
if len(root_rows) != 1:
raise ValueError(f"an aggregate has exactly one {self.root.name} row, not {len(root_rows)}")
root_key = {k: root_rows[0][k] for k in self.root_key}
_forget(self, tuple(root_key[k] for k in self.root_key))
now = now or datetime.now(timezone.utc)
if self.version and expected_version is None:
raise TypeError(f"{self.root.name} is versioned: pass expected_version (what load returned; "
"0 for a new one)")
if not self.version and expected_version is not None:
raise TypeError(f"{self.root.name} is not versioned: construct the Aggregate with version=...")
if snapshot is not None:
return self._save_diff(conn, rows, root_key, now, expected_version, _snapshot_of(snapshot))
# 1. What is stored under this root NOW — read before anything is written, so the reads see
# committed state only (there is no read-your-writes inside the transaction).
stored: dict = {}
for part in self.parts:
table = _table(part.table)
result = conn.execute(select(*(table.c[k] for k in part.key))
.where(self._scope(part, root_key)))
stored[part.name] = {tuple(r) for r in result}
# 2. Check every row before writing any, so a bad row leaves storage untouched.
for part in self.parts:
for row in rows.get(part.name) or []:
_check_key_types(part.name, _table(part.table), part.key, row)
for k in self.root_key:
if row.get(k) != root_key[k]:
raise ValueError(f"{part.name}: a row belongs to {k}={row.get(k)!r}, "
f"not to this aggregate's {root_key[k]!r}")
# 2b. Versioned: refuse the ordinary stale save against committed state, then CLAIM expected+1 with a
# plain INSERT in this transaction — the race-safe part (see the module docstring).
written: Optional[int] = None
if self.version:
rt = _table(self.root.table)
found = conn.execute(select(rt.c[self.version]).where(self._scope(self.root, root_key))).first()
current = int(found[0] or 0) if found else 0
if current != int(expected_version):
raise Conflict(rt.name, root_key, int(expected_version), current)
written = int(expected_version) + 1
try:
conn.execute(insert(self.versions).values(**root_key, **{self.version: written}, saved_at=now))
except IntegrityError as exc:
if as_conflict(exc) is None:
raise
raise Conflict(rt.name, root_key, int(expected_version)) from None
root_rows = [{**root_rows[0], self.version: written}]
rows = {**rows, self.root.name: root_rows}
# 3. Write everything in memory.
for part in self.parts:
table = _table(part.table)
stamped = [self._stamp(table, row, now) for row in rows.get(part.name) or []]
if stamped:
upsert(conn, table, stamped, part.key, part.insert_only,
existing=stored[part.name])
# 4. Delete what is stored but no longer in memory — at every depth, by full key.
for part in self.children:
table = _table(part.table)
in_memory = {tuple(r[k] for k in part.key) for r in rows.get(part.name) or []}
for stale in sorted(stored[part.name] - in_memory, key=repr):
conn.execute(delete(table).where(
and_(*(table.c[k] == v for k, v in zip(part.key, stale)))))
return written
def _check_rows(self, rows: dict, root_key: dict) -> None:
for part in self.parts:
for row in rows.get(part.name) or []:
_check_key_types(part.name, _table(part.table), part.key, row)
for k in self.root_key:
if row.get(k) != root_key[k]:
raise ValueError(f"{part.name}: a row belongs to {k}={row.get(k)!r}, "
f"not to this aggregate's {root_key[k]!r}")
def _save_diff(self, conn: Any, rows: dict, root_key: dict, now: datetime,
expected_version: Optional[int], snap: "Snapshot") -> Optional[int]:
if snap.aggregate is not self:
raise ValueError("this snapshot was read by another aggregate")
if snap.root_key != tuple(root_key[k] for k in self.root_key):
raise ValueError(f"this snapshot is of {dict(zip(self.root_key, snap.root_key))}, not of {root_key}")
self._check_rows(rows, root_key)
# What to write: rows that are new, or that differ from what was read in a column they name.
writes: dict = {}
stale: dict = {}
for part in self.parts:
table, stored = _table(part.table), snap.parts.get(part.name, {})
ignore = {"updated_at", *part.insert_only} | ({self.version} if part is self.root and self.version else set())
mine: dict = {}
for row in rows.get(part.name) or []:
mine[tuple(row[k] for k in part.key)] = row
changed = [row for kv, row in mine.items()
if kv not in stored or any(c not in ignore and (c not in stored[kv] or stored[kv][c] != v)
for c, v in row.items())]
writes[part.name] = [self._stamp(table, row, now) for row in changed]
if part is not self.root:
stale[part.name] = sorted(set(stored) - set(mine), key=repr)
written: Optional[int] = None
if self.version:
rt = _table(self.root.table)
# 0.9.13: the stale case is refused against the snapshot first, with no statement. The committed root is
# still checked — a concurrent remove() deletes the root, its children AND every claim, so the claim alone
# would let a save from an older snapshot resurrect a partial aggregate (review of 70fcf3e) — but the check
# is QUEUED in the same psycopg pipeline as the claim and the writes (one round trip), and decided when the
# pipeline returns, before COMMIT: a missing root or another version is a Conflict and the caller's
# transaction rolls back. It reads committed state, so it needs no read-your-writes. Why not a conditional
# upsert (`… ON CONFLICT DO UPDATE … WHERE version = :v`): on the DataK3 pg wire a statement's rowcount
# cannot be the decision (every racer was told rowcount 1 before COMMIT, measured 2026-10-17; data_pg
# reports row_count 0 for any DML), so only a row read or a duplicate key can say "you lost".
current = snap.version
if current != int(expected_version):
raise Conflict(rt.name, root_key, int(expected_version), current)
written = int(expected_version) + 1
if writes[self.root.name]:
writes[self.root.name] = [{**writes[self.root.name][0], self.version: written}]
else:
# Only the version moves, but the row sent is the WHOLE root as read (with the caller's values on
# top, equal by definition): Postgres and SQLite check NOT NULL on the proposed INSERT row before
# ON CONFLICT turns it into an update, so a row of the key and the version alone failed on any
# root with a NOT NULL column (0.9.6). Still one statement, the same upsert as a changed root;
# the claim already makes this transaction the only versioned writer of the root.
stored_root = snap.parts.get(self.root.name, {}).get(snap.root_key) or {}
mine_root = (rows.get(self.root.name) or [{}])[0]
bump = {c: v for c, v in {**stored_root, **mine_root}.items() if c in rt.c}
writes[self.root.name] = [{**self._stamp(rt, bump, now), self.version: written}]
if not self.version and not any(writes.values()) and not any(stale.values()):
return None # nothing changed: nothing to send
check = None
try:
with _write_pipeline(conn, batched=True):
if self.version:
rt = _table(self.root.table)
# The claim first: on SQLite it takes the write lock before the read (a read first would hold a shared
# lock two racers then both try to upgrade — "database is locked" after the busy timeout). The check
# still comes before the root's own write, so it reads the committed root everywhere.
conn.execute(insert(self.versions).values(**root_key, **{self.version: written}, saved_at=now))
check = _deferred_select(conn, select(rt.c[self.version]).where(self._scope(self.root, root_key)))
for part in self.parts:
if writes[part.name]:
upsert(conn, _table(part.table), writes[part.name], part.key, part.insert_only,
existing=set(snap.parts.get(part.name, {})))
for part in self.children:
if stale[part.name]:
_delete_keys(conn, _table(part.table), part.key, self.root_key, root_key, stale[part.name])
except Conflict:
raise
except Exception as exc: # noqa: BLE001 — re-raised unless the version claim lost
if self.version and as_conflict(exc) is not None:
raise Conflict(_table(self.root.table).name, root_key, int(expected_version)) from None
raise
if check is not None:
found = check.rows()
stored = int(found[0][0] or 0) if found else None
if stored != int(expected_version) and not (stored is None and int(expected_version) == 0):
# removed meanwhile (None), or saved by someone the claim missed; a NEW aggregate (0) has no root yet.
# The claim and the writes are already queued in the caller's transaction: roll it back before raising,
# so a caller that catches the Conflict inside its unit of work and commits persists none of them.
_abort(conn)
raise Conflict(_table(self.root.table).name, root_key, int(expected_version), stored)
return written
def remove(self, conn: Any, root_key: dict) -> None:
"""Delete the whole aggregate — every part, scoped to this root. Children first. A versioned
aggregate's claims go too, so the same key can be created again from version 0."""
_forget(self, tuple(root_key[k] for k in self.root_key))
for part in (*reversed(self.children), self.root):
conn.execute(delete(_table(part.table)).where(self._scope(part, root_key)))
if self.versions is not None:
conn.execute(delete(self.versions).where(
and_(*(self.versions.c[k] == root_key[k] for k in self.root_key))))
@staticmethod
def _stamp(table: Table, row: dict, now: datetime) -> dict:
row = dict(row)
if "updated_at" in table.c:
row["updated_at"] = now
if "created_at" in table.c:
row.setdefault("created_at", now) # insert-only, so this lands on INSERT alone
return row
# ── 0.9.6: snapshots, single-round-trip reads, diff writes ─────────────────────────────────────────────────
# Measured live 2026-10-24 (sales-order on DataK3, ~70 ms to pg.uk-lon-1): a statement costs one round trip and
# almost no server time; an IN list of 50-100 keys costs what one key costs; psycopg 3's pipeline mode sent 20
# reads in 210 ms instead of 1,580. So the cost of a use case is its count of round trips, and these helpers
# cut it: a load is one round trip, a save writes only what changed.
class Snapshot:
"""What a load read, per part: {part name: {key tuple: row}}. Copies, so editing the loaded dict in place
does not change what the save compares against."""
def __init__(self, aggregate: "Aggregate", root_key: tuple, parts: dict):
self.aggregate, self.root_key, self.parts = aggregate, root_key, parts
@classmethod
def of(cls, aggregate: "Aggregate", root_key: dict, loaded: dict) -> "Snapshot":
parts = {p.name: {tuple(r[k] for k in p.key): copy.deepcopy(r) for r in loaded.get(p.name) or []}
for p in aggregate.parts}
return cls(aggregate, tuple(root_key[k] for k in aggregate.root_key), parts)
@property
def version(self) -> int:
if not self.aggregate.version:
return 0
root = next(iter(self.parts[self.aggregate.root.name].values()), None)
return int((root or {}).get(self.aggregate.version) or 0)
class Loaded(dict):
"""A load's answer: the same {table name: [row, ...]} dict as always, plus `.snapshot`."""
def __init__(self, parts: dict, snapshot: Snapshot):
super().__init__(parts)
self.snapshot = snapshot
def _snapshot_of(s: Any) -> Snapshot:
if isinstance(s, Snapshot):
return s
if isinstance(s, Loaded):
return s.snapshot
raise TypeError("snapshot= takes what load() returned (or its .snapshot, or nothing_stored(key))")
def _any_key(table: Table, cols: tuple, values: list):
"""Rows whose `cols` equal any of `values`: `c IN (...)` for one column, OR-of-ANDs for several (portable:
DataK3's dialect is not promised to take a tuple IN)."""
if len(cols) == 1:
return table.c[cols[0]].in_([v[0] for v in values])
return or_(*(and_(*(table.c[c] == x for c, x in zip(cols, v))) for v in values))
def _delete_keys(conn: Any, table: Table, key: tuple, root_cols: tuple, root_key: dict, stale: list,
chunk: int = 500) -> None:
"""ONE DELETE for this table's stale rows (per `chunk`), always scoped by the root."""
scope = and_(*(table.c[k] == root_key[k] for k in root_cols))
rest = tuple(key[len(root_cols):])
if not rest:
conn.execute(delete(table).where(scope))
return
tails = [tuple(kv[len(root_cols):]) for kv in stale]
for i in range(0, len(tails), chunk):
conn.execute(delete(table).where(and_(scope, _any_key(table, rest, tails[i:i + chunk]))))
# Listeners told about every pipeline: fn(sa_connection, statements). erp_platform.testing counts them.
PIPELINE_LISTENERS: list = []
PIPELINE_ENV = "ERP_PIPELINE"
def _sa_connection(conn: Any):
"""The SQLAlchemy Connection behind `conn` (a Session's, or itself), or None for anything else."""
if hasattr(conn, "get_bind") and callable(getattr(conn, "connection", None)):
try:
conn = conn.connection()
except Exception: # noqa: BLE001 — a fake: no pipeline
return None
return conn if hasattr(conn, "dialect") and hasattr(conn, "in_transaction") else None
def _driver(conn: Any):
"""The psycopg 3 connection to pipeline on, or None (SQLite, other drivers, fakes, ERP_PIPELINE=0)."""
if os.environ.get(PIPELINE_ENV, "1").strip() in ("0", "false", "off"):
return None, None
sa = _sa_connection(conn)
if sa is None or getattr(sa.dialect, "driver", "") != "psycopg":
return None, None
try:
driver = sa.connection.driver_connection
except Exception: # noqa: BLE001
return None, None
return (sa, driver) if hasattr(driver, "pipeline") else (None, None)
def _read_all(conn: Any, stmts: list) -> list:
"""Run these SELECTs and return each one's rows as dicts — in ONE round trip on psycopg 3 (pipeline mode on
the driver connection under SQLAlchemy's, verified on Postgres and measured on DataK3), in turn elsewhere."""
sa, driver = _driver(conn)
if driver is None or len(stmts) < 2:
sa = _sa_connection(conn) if len(stmts) >= 2 else None
if sa is None or sa.info.get("erp_pipeline") or getattr(sa.dialect, "driver", "") == "psycopg":
return [[dict(r._mapping) for r in conn.execute(s)] for s in stmts]
# 0.9.13: one after another here, but ONE round trip on psycopg 3 — so a statement counter
# (erp_platform.testing) reports what the same load costs on the pg wire, not what it costs on SQLite.
for fn in PIPELINE_LISTENERS:
fn(sa, 0)
sa.info["erp_pipeline"] = True
try:
return [[dict(r._mapping) for r in conn.execute(s)] for s in stmts]
finally:
sa.info.pop("erp_pipeline", None)
if not sa.in_transaction():
sa.begin() # what the first execute would do: autobegin, same txn
dialect = sa.dialect
compiled = [s.compile(dialect=dialect, compile_kwargs={"render_postcompile": True}) for s in stmts]
for fn in PIPELINE_LISTENERS:
fn(sa, len(stmts))
cursors = []
try:
with driver.pipeline():
for c in compiled:
cur = driver.cursor()
cur.execute(c.string, c.params)
cursors.append(cur)
out = []
for s, cur in zip(stmts, cursors):
names = [d.name for d in cur.description]
cols = {c.name: c for c in s.selected_columns}
procs = []
for d in cur.description:
col = cols.get(d.name)
procs.append(col.type.dialect_impl(dialect).result_processor(dialect, d.type_code)
if col is not None else None)
out.append([{n: (p(v) if p else v) for n, p, v in zip(names, procs, row)} for row in cur.fetchall()])
return out
except Exception as exc: # noqa: BLE001 — wrapped like SQLAlchemy would
from sqlalchemy.exc import DBAPIError
import psycopg
if isinstance(exc, psycopg.Error):
if dialect.is_disconnect(exc, driver, None):
sa.invalidate(exc)
raise DBAPIError.instance(compiled[0].string, None, exc, psycopg.Error, dialect=dialect) from exc
raise
finally:
for cur in cursors:
cur.close()
def _abort(conn: Any) -> None:
"""Roll back the transaction `conn` is in (a Session's, or a Connection's), whichever path sent the statements."""
if hasattr(conn, "get_bind") and hasattr(conn, "rollback"):
conn.rollback() # a Session
return
sa = _sa_connection(conn)
tx = sa.get_transaction() if sa is not None else None
if tx is not None and tx.is_active:
tx.rollback()
class _Deferred:
"""A SELECT queued now and read later: inside a psycopg pipeline it goes out with the statements around it and its
rows are fetched once the pipeline has returned; elsewhere it runs at once."""
def __init__(self, cursor=None, rows=None):
self._cursor, self._rows = cursor, rows
def rows(self) -> list:
if self._rows is None:
try:
self._rows = [tuple(r) for r in self._cursor.fetchall()]
finally:
self._cursor.close()
return self._rows
def _deferred_select(conn: Any, stmt) -> _Deferred:
sa = _sa_connection(conn)
if sa is not None and sa.info.get("erp_pipeline") and getattr(sa.dialect, "driver", "") == "psycopg":
driver = sa.connection.driver_connection
c = stmt.compile(dialect=sa.dialect, compile_kwargs={"render_postcompile": True})
for fn in PIPELINE_LISTENERS: # counted as a statement of the pipeline it rides in
fn(sa, 1)
cur = driver.cursor()
cur.execute(c.string, c.params)
return _Deferred(cursor=cur)
return _Deferred(rows=[tuple(r) for r in conn.execute(stmt)])
WRITE_PIPELINE_ENV = "ERP_PIPELINE_WRITES"
@contextmanager
def _write_pipeline(conn: Any, batched: bool = False):
"""The statements inside go out together on psycopg 3 when ERP_PIPELINE_WRITES is on; one by one elsewhere.
A duplicate key or any other error surfaces at the latest when the block ends, still inside the txn.
`batched` (0.9.13): elsewhere (SQLite), a statement counter is told the block is the ONE round trip it is on psycopg."""
sa, driver = _driver(conn)
if driver is None or os.environ.get(WRITE_PIPELINE_ENV, "1").strip() in ("0", "false", "off"):
sa = _sa_connection(conn) if batched else None
if sa is None or sa.info.get("erp_pipeline") or getattr(sa.dialect, "driver", "") == "psycopg":
yield
return
for fn in PIPELINE_LISTENERS:
fn(sa, 0)
sa.info["erp_pipeline"] = True
try:
yield
finally:
sa.info.pop("erp_pipeline", None)
return
if not sa.in_transaction():
sa.begin()
sa.info["erp_pipeline"] = True
try:
for fn in PIPELINE_LISTENERS:
fn(sa, 0)
with driver.pipeline():
yield
finally:
sa.info.pop("erp_pipeline", None)
# ── 0.9.13: the per-request identity map ───────────────────────────────────────────────────────────────────────
# Measured 2026-10-25 on the Lead-to-Cash composition (SQLite, a one-line sale): a sales order is 13 tables, and one
# delivery validation read it twice (inventory asks sales-order for the order's fulfilment, then records the delivery
# on it: two in-process seam requests), a product three times, the same settings rows in every callee. Now an
# aggregate (or a `memo` read) is read once per REQUEST per key per CALLER. Rules, each one a test
# (tests/test_identity_map.py):
# * the map belongs to the outer request: the doors open `request_scope(caller)` around each use-case attempt, and a
# tier-1 seam call carries the map into its neighbour's request in this process (the request tree). It is never
# shared between two outer requests; outside a scope (scripts, tests, jobs, after-response work) nothing is cached.
# * every entry is keyed by the CALLER of the request that read it (its whole identity: a person, or a brick's own
# service grant, or a person through a brick's door), so a seam call on someone's behalf and the brick's own grant
# never answer each other's reads, and two people never do — whatever a use case later decides from the rows.
# * a hit is a deep copy, so editing loaded rows in place never changes what the next load answers.
# * an entry is dropped when its aggregate is saved or removed, when any statement in the tree writes one of its
# tables (INSERT / UPDATE / DELETE naming it, text SQL included), wholly on the ROLLBACK of a transaction that
# wrote, and wholly after a seam
# COMMAND to a neighbour in another process (it may call back into this brick, in a request the map cannot see).
# A read made by a connection that has written in its open transaction is not kept (SQLite and Postgres show a
# transaction its own writes; another request in the tree must only ever be answered committed rows).
# * a retried attempt starts from an empty map.
# What it does not do: re-read between two loads in one request, so a row another request commits in between is not
# seen until the next request — the snapshot a save compares is the first read, and a versioned save still refuses
# a racer through its claim.
_MISS = object()
_TREE: ContextVar = ContextVar("erp_platform_request_tree", default=None)
_WHO: ContextVar = ContextVar("erp_platform_request_caller", default=None)
_AGGREGATES: dict = {} # id(aggregate) -> aggregate, for the aggregates ever cached
_WRITE = re.compile(r"^\s*(?:with\b.*?\)\s*)?(insert|update|delete|merge|replace|upsert|alter|drop|truncate|create|copy)\b",
re.IGNORECASE | re.DOTALL)
_SELECT_INTO = re.compile(r"^\s*select\b.*?\binto\b", re.IGNORECASE | re.DOTALL)
_LEADING_COMMENTS = re.compile(r"^(?:\s*(?:--[^\n]*(?:\n|$)|/\*.*?\*/))*", re.DOTALL)
def is_write(statement: str) -> bool:
"""Does this SQL change rows or tables? Leading comments are skipped; INSERT / UPDATE / DELETE / MERGE / COPY / DDL,
a WITH … that ends in one, and SELECT … INTO count. When unsure it says yes: a needless drop costs one re-read."""
sql = _LEADING_COMMENTS.sub("", statement or "", count=1)
return bool(_WRITE.match(sql) or _SELECT_INTO.match(sql))
class _Tree:
"""One outer request's map: {entry key: value} and {table name: {entry keys reading it}}."""
def __init__(self):
self.entries: dict = {}
self.tables: dict = {}
self.closed = False
self.lock = threading.Lock()
def put(self, key: tuple, value: Any, tables) -> None:
with self.lock:
self.entries[key] = value
for t in tables:
self.tables.setdefault(t.lower(), set()).add(key)
def drop(self, keys) -> None:
with self.lock:
for k in keys:
self.entries.pop(k, None)
def clear(self) -> None:
with self.lock:
self.entries.clear()
self.tables.clear()
def _caller_key(caller: Optional[dict]) -> str:
import hashlib
import json
return hashlib.sha256(json.dumps(caller or {}, sort_keys=True, default=str).encode()).hexdigest()
@contextmanager
def request_scope(caller: Optional[dict] = None, *, fresh: bool = False):
"""One request's view of the identity map, for `caller`. The outermost scope owns the map and closes it on exit;
a scope opened inside the same request tree (a tier-1 seam call's neighbour) shares it. `fresh` (a retried
attempt) empties it first."""
tree = current_tree()
own = tree is None
if own:
tree = _Tree()
elif fresh:
tree.clear()
t_tree, t_who = _TREE.set(tree), _WHO.set(_caller_key(caller))
_ensure_listeners()
try:
yield tree
finally:
_WHO.reset(t_who)
_TREE.reset(t_tree)
if own:
tree.closed = True
tree.clear()
def current_tree() -> Optional[_Tree]:
"""The open map of this request tree, or None (what a tier-1 seam carries into its neighbour's thread)."""
tree = _TREE.get()
return None if tree is None or tree.closed else tree
def adopt_tree(tree: Optional[_Tree]) -> None:
"""Make `tree` this (fresh) thread's request tree — the tier-1 seam transport calls it on the thread it delivers
on. The neighbour's door still keys its entries by ITS caller."""
if tree is not None and not tree.closed:
_TREE.set(tree)
def _who() -> Optional[str]:
return _WHO.get()
def memo(conn: Any, key: Any, tables: tuple, read):
"""`read()` once per request per caller per engine per key; a copy of the answer after that. `conn`: the connection the
read uses (it names the engine, so two bricks never answer each other, and a connection that has written in its open
transaction is never memoised). `tables`: every table the read looks at, so a write to any of them drops it. Outside a
request scope it just calls `read()`.
rules = repository.memo(conn, ("so_order_type", on), ("so_order_type",), lambda: [...])"""
tree, who, bind = current_tree(), _who(), _bind_of(conn)
if tree is None or who is None or bind is None:
return read()
k = ("memo", who, bind, key)
hit = tree.entries.get(k, _MISS)
if hit is not _MISS:
return copy.deepcopy(hit)
value = read()
if not _wrote(conn):
tree.put(k, copy.deepcopy(value), tables)
return value
def detach() -> None:
"""Leave this context's request tree (after-response work, a job started from a request): nothing it reads is cached
or answered from the request's map."""
_TREE.set(None)
_WHO.set(None)
def _bind_of(conn: Any):
"""What a cached aggregate is keyed by besides its key: the engine it was read from (two bricks' engines in one
process never share an entry). None for anything that is not a SQLAlchemy connection or session: not cached."""
sa = _sa_connection(conn)
if sa is None:
return None
return id(sa.engine)
def _wrote(conn: Any) -> bool:
sa = _sa_connection(conn)
return sa is None or bool(sa.info.get("erp_wrote"))
def _copy_loaded(got):
if got is None:
return None
parts = copy.deepcopy(dict(got))
snap = got.snapshot
return Loaded(parts, Snapshot(snap.aggregate, snap.root_key, copy.deepcopy(snap.parts)))
def _forget(agg: "Aggregate", root_key: tuple) -> None:
tree = current_tree()
if tree is not None:
tree.drop([k for k in list(tree.entries) if k[0] == "agg" and k[1] == id(agg) and k[4] == root_key])
def forget_all() -> None:
"""Drop everything this request tree has cached (after a write the map cannot see)."""
tree = current_tree()
if tree is not None:
tree.clear()
def note_write(statement: str, conn: Any = None) -> None:
"""Tell the identity map about a write it could not see: SQL sent on the raw driver (a psycopg pipeline of your own,
`cursor.execute` on the DBAPI connection) bypasses SQLAlchemy's events, so call this once per statement (or once
with any statement naming each table written). `conn` (a Connection or Session) marks it as having written in its
open transaction. Writes through `conn.execute` need none of this."""
sa = _sa_connection(conn) if conn is not None else None
_on_statement(sa, statement, force=True)
def _on_statement(conn, statement: str, force: bool = False) -> None:
if not (force or is_write(statement)):
return
if conn is not None:
conn.info["erp_wrote"] = True
tree = current_tree()
if tree is None or not tree.entries:
return
low = statement.lower()
hit = set()
for t, keys in list(tree.tables.items()):
if t in low and re.search(r"(?<!\w)\"?" + re.escape(t) + r"\"?(?!\w)", low): # bare, quoted or schema.table
hit |= keys
if hit:
tree.drop(hit)
def _on_end(conn, rolled_back: bool) -> None:
# A read-only connection rolls back when it closes: nothing it read can be gone. Only a transaction that WROTE
# and rolled back may have shown this request rows that no longer exist.
if conn.info.pop("erp_wrote", None) and rolled_back:
forget_all()
_LISTENING = threading.Lock()
_LISTENERS_ON = False
def _ensure_listeners() -> None:
global _LISTENERS_ON
if _LISTENERS_ON:
return
with _LISTENING:
if _LISTENERS_ON:
return
event.listen(Engine, "before_cursor_execute", lambda conn, cursor, statement, *a, **k: _on_statement(conn, statement))
event.listen(Engine, "commit", lambda conn: _on_end(conn, False))
event.listen(Engine, "rollback", lambda conn: _on_end(conn, True))
_LISTENERS_ON = True