"""Fewer round trips, the same statements: pipelined reads and writes, and an aggregate save that writes only
what changed since the load — inside hr's infrastructure, over erp_platform.repository's rules.
WHY. On DataK3 a statement costs a network round trip (~70-100 ms from a laptop; measured 2026-10-17 on
erp-platform-test with scripts/bench_live.py), and the server time of a small keyed statement is a fraction of
that. A hire was ~130 sequential round trips. psycopg's PIPELINE mode sends several statements before reading any
answer, and the DataK3 pg wire honours it (measured the same day: six keyed SELECTs, or six upserts, plus BEGIN and
COMMIT, in ~230 ms against ~580 ms one by one; a duplicate claim inside a pipeline still fails 23505 and rolls the
whole transaction back). Nothing about WHAT is read or written changes — only how many times we wait.
* `select_many(conn, stmts)` → the rows of each SELECT, in order. One round trip on psycopg; one statement after
another on any other driver (SQLite in the tests), through the same SQLAlchemy connection.
* `execute_many(conn, stmts)` → the same for writes. An error is raised as SQLAlchemy raises it (an
IntegrityError for a duplicate key), so erp_platform.repository.as_conflict and is_duplicate read it the same.
* `load` / `save` (0.6.0) are erp_platform 0.9.6's Aggregate.load and save(snapshot=): one round trip, and a
save that writes only what changed since the load — claiming version n+1 with the same plain INSERT, so a
racer who saved after the load makes the whole transaction fail 23505 → Conflict. Until 0.6.0 hr carried
its own copy of that; the platform's is the one now.
"""
import time
from datetime import date, datetime, timezone
from typing import Any, Callable, Iterable, Optional
from sqlalchemy import and_, delete, insert, or_, select
from sqlalchemy.exc import DBAPIError
from erp_platform import repository as platform
# The benchmark (scripts/bench_live.py) counts batches here: observer(statements, pipelined, seconds, first SQL).
# Pipelined statements bypass SQLAlchemy's cursor events; sequential ones (SQLite) fire them, and a batch is
# still ONE wire round trip.
observer: Optional[Callable[[int, bool, float, str], None]] = None
def _observe(n, pipelined, t0, stmt):
if observer:
observer(n, pipelined, time.monotonic() - t0, str(stmt)[:80])
def _pipeline_conn(conn) -> Any:
"""The psycopg connection under a SQLAlchemy connection, when pipelining is possible; else None."""
try:
if conn.dialect.name != "postgresql":
return None
raw = conn.connection.dbapi_connection
except AttributeError:
return None
raw = getattr(raw, "driver_connection", raw)
return raw if hasattr(raw, "pipeline") else None
def _compile(conn, stmt):
c = stmt.compile(dialect=conn.dialect, compile_kwargs={"render_postcompile": True}) # IN (...) expanded
params = c.construct_params()
procs = getattr(c, "_bind_processors", {}) or {}
return str(c), {k: (procs[k](v) if k in procs else v) for k, v in params.items()}
def _raise(exc: Exception, sql: str, params: dict):
import psycopg
raise DBAPIError.instance(sql, params, exc, psycopg.Error) from exc
def select_many(conn, stmts: list) -> list:
"""Each SELECT's rows as dicts, in order — one round trip on psycopg."""
if not stmts:
return []
pc = _pipeline_conn(conn)
t0 = time.monotonic()
if pc is None:
out = [[dict(r._mapping) for r in conn.execute(s)] for s in stmts]
_observe(len(stmts), False, t0, stmts[0])
return out
compiled = [_compile(conn, s) for s in stmts]
curs = [pc.cursor() for _ in stmts]
out = []
current = ("", {})
try:
with pc.pipeline() as p:
for cur, (sql, params) in zip(curs, compiled):
current = (sql, params)
cur.execute(sql, params)
p.sync()
for cur, (sql, params) in zip(curs, compiled):
current = (sql, params)
names = [d.name for d in cur.description]
out.append([dict(zip(names, row)) for row in cur.fetchall()])
except Exception as exc: # noqa: BLE001 — re-raised as SQLAlchemy would raise it
import psycopg
if isinstance(exc, psycopg.Error):
_raise(exc, *current)
raise
finally:
for cur in curs:
cur.close()
_observe(len(stmts), True, t0, stmts[0])
return out
def execute_many(conn, stmts: list) -> None:
"""Run DML statements in order — one round trip on psycopg. The first failure stops the rest (pipeline
semantics; the transaction is then failed and rolled back by its owner, as with one-by-one execution)."""
if not stmts:
return
pc = _pipeline_conn(conn)
t0 = time.monotonic()
if pc is None:
for s in stmts:
conn.execute(s)
_observe(len(stmts), False, t0, stmts[0])
return
compiled = [_compile(conn, s) for s in stmts]
# 0.9.13: the pipeline bypasses SQLAlchemy's cursor events, so tell the request's identity map first — every
# cached aggregate on a table written here is dropped, and the connection is marked as having written (a
# rollback then forgets the map). Without it the next load in the same request answers rows from before.
for sql, _params in compiled:
platform.note_write(sql, conn)
current = ("", {})
try:
with pc.pipeline():
with pc.cursor() as cur:
for sql, params in compiled:
current = (sql, params)
cur.execute(sql, params)
_observe(len(stmts), True, t0, stmts[0])
except Exception as exc: # noqa: BLE001
import psycopg
if isinstance(exc, psycopg.Error):
_raise(exc, *current)
raise
class _Collect:
"""Stands in for a connection so erp_platform.repository.upsert builds its statements without running them."""
def __init__(self):
self.stmts: list = []
def execute(self, stmt):
self.stmts.append(stmt)
# ── aggregates: erp_platform 0.9.6's snapshots ─────────────────────────────────────────────
# 0.9.6 moved what this module used to do itself into erp_platform.repository: a load is one round trip and
# returns `.snapshot`, `save(snapshot=)` writes only the diff (one DELETE per table), `nothing_stored(key)` is the
# snapshot of an aggregate that is not there. hr keeps two things: the prefetch (an aggregate's SELECTs sent in
# the same round trip as other reads, from_results) and carrying the snapshot past a save, so a unit of work that
# saves the same aggregate twice still writes only the second change.
Snapshot = platform.Snapshot
def absent(aggregate, root_key: dict) -> Snapshot:
return aggregate.nothing_stored(root_key)
def _t(part):
return getattr(part.table, "__table__", part.table)
def _scope(aggregate, part, root_key: dict):
t = _t(part)
return and_(*(t.c[k] == root_key[k] for k in aggregate.root_key))
def load_statements(aggregate, root_key: dict) -> list:
"""Aggregate.load's SELECTs: one per table, scoped by the root's key, the ROOT FIRST — for a prefetch that sends
them along with other reads (HrRepository.prefetch)."""
return [select(_t(p)).where(_scope(aggregate, p, root_key)) for p in aggregate.parts]
def load(conn, aggregate, root_key: dict) -> tuple:
"""(rows or None, Snapshot) — erp_platform's Aggregate.load: one round trip."""
loaded = aggregate.load(conn, root_key)
if loaded is None:
return None, absent(aggregate, root_key)
return loaded, loaded.snapshot
def from_results(aggregate, root_key: dict, results: list) -> tuple:
"""(rows or None, Snapshot) from the results of load_statements, in order — what Aggregate.load would answer."""
out = {p.name: rows for p, rows in zip(aggregate.parts, results)}
if not out[aggregate.root.name]:
return None, absent(aggregate, root_key)
if aggregate.version:
root = out[aggregate.root.name][0]
root[aggregate.version] = int(root.get(aggregate.version) or 0)
return out, Snapshot.of(aggregate, root_key, out)
def save(conn, aggregate, rows: dict, *, expected_version: Optional[int], snapshot: Optional[Snapshot],
now: Optional[datetime] = None) -> tuple:
"""(version written, the snapshot of what is now stored) — erp_platform's save(snapshot=) when the snapshot is
the version being saved from; its full save otherwise (then the returned snapshot is None)."""
if (snapshot is None or not aggregate.version or expected_version is None
or int(expected_version) != snapshot.version):
return aggregate.save(conn, rows, now=now, expected_version=expected_version), None
written = aggregate.save(conn, rows, now=now, expected_version=expected_version,
snapshot=snapshot)
root = rows[aggregate.root.name][0]
root_key = {k: root[k] for k in aggregate.root_key}
after = {**rows, aggregate.root.name: [{**root, aggregate.version: written}]}
return written, Snapshot.of(aggregate, root_key, after)
__all__ = ["select_many", "execute_many", "load", "save", "from_results", "load_statements", "absent", "Snapshot",
"observer", "Iterable"]