"""One repository per aggregate: rows ↔ domain objects, through erp_platform.repository (load whole, save whole).
PERFORMANCE, designed in from the start (brick-hr was slow on DataK3: its config reloaded per request, saves wrote
row by row, its tick scanned everything):
* the Configuration snapshot (types, policies, cadences, consents, retention, consumers, settings, the current
template per type) is cached per engine and checked against ONE row (ct_config_version) per request;
* template versions and clause words are cached under the config token too (an approved version is immutable);
* a contract save writes only the PARTS whose rows changed since load (an OPENED event writes the root, one signer
row and one event — not eleven tables), and events go in as one multi-row INSERT;
* list queries are set-based (one SELECT, visibility in the WHERE), the tick selects only what is due
(ct_contract.next_due_at, ct_key_date.next_remind_on) and loads the due contracts in one SELECT per table.
"""
import base64
import hashlib
import json
import threading
import uuid
from collections import defaultdict
from datetime import date, datetime, timezone
from typing import Iterable, Optional
from sqlalchemy import and_, delete, func, insert, or_, select
from sqlalchemy.exc import IntegrityError
from erp_platform.repository import Aggregate, Conflict, Part, upsert
from brick_contracts.domain import contract as C
from brick_contracts.domain.contract_type import type_from_row
from brick_contracts.domain.library import Clause, ClauseVersion, Slot, Template, TemplateVersion, VariableDef
from brick_contracts.domain.policies import (ApprovalPolicy, Configuration, cadence_from_row, consumer_from_row,
retention_from_row, rule_from_row)
from brick_contracts.domain.values import ConsentText, T_APPROVED, VariableValue
from brick_contracts.infrastructure import tables as T
CHUNK = 192_000
_LOCK = threading.Lock()
_SNAPSHOTS: dict = {} # engine key → (token, Configuration)
_LIBRARY: dict = {} # (engine key, token, kind, id, version) → immutable library object
def _engine_key(conn) -> tuple:
e = conn.engine
return (id(e), str(e.url))
def naive(t: Optional[datetime]) -> Optional[datetime]:
"""What a TIMESTAMP column holds: naive UTC."""
if t is None:
return None
if t.tzinfo is not None:
t = t.astimezone(timezone.utc).replace(tzinfo=None)
return t
def aware(t) -> Optional[datetime]:
"""Back to an aware UTC datetime (DataK3 hands back aware, SQLite naive — and Python refuses to compare them)."""
if t is None:
return None
if isinstance(t, str):
t = datetime.fromisoformat(t.replace("Z", "+00:00"))
if isinstance(t, datetime):
return t.replace(tzinfo=timezone.utc) if t.tzinfo is None else t.astimezone(timezone.utc)
return t
def _d(v) -> Optional[date]:
if v is None:
return None
if isinstance(v, datetime):
return v.date()
if isinstance(v, date):
return v
return date.fromisoformat(str(v)[:10])
def _j(v) -> Optional[str]:
return None if v is None else json.dumps(v, sort_keys=True, ensure_ascii=False, default=str)
def _uj(v, default=None):
if v is None or v == "":
return default
try:
return json.loads(v)
except (TypeError, ValueError):
return default
def is_duplicate(exc: BaseException) -> bool:
text = str(exc).lower()
return isinstance(exc, IntegrityError) or "23505" in text or "duplicate key" in text or "unique constraint" in text
def _norm(rows: list) -> list:
def one(r):
return {k: (naive(v).isoformat() if isinstance(v, datetime) else v.isoformat() if isinstance(v, date)
else bool(v) if isinstance(v, bool) else v) for k, v in r.items() if k not in ("updated_at",)}
return sorted((json.dumps(one(r), sort_keys=True, default=str) for r in rows))
# ── Contract ───────────────────────────────────────────────────────────────────────────────
def contract_rows(c: C.Contract) -> dict:
root = {"contract_id": c.contract_id, "type_code": c.type_code, "template_id": c.template_id,
"template_version": int(c.template_version), "bp_id": int(c.bp_id), "context_ref": c.context_ref,
"source_ref": c.source_ref, "origin": c.origin, "owner": c.owner, "owner_group": c.owner_group,
"title": c.title, "state": c.state, "sensitivity": c.sensitivity, "execution": c.execution,
"starts_on": c.starts_on, "ends_on": c.ends_on, "notice_days": c.notice_days,
"amends_contract_id": c.amends_contract_id, "amendment_kind": c.amendment_kind,
"draft_fingerprint": c.draft_fingerprint, "request_hash": c.request_hash,
"missing_json": _j(c.missing), "deviations_json": _j(c.deviations), "reference": c.reference,
"current_version": int(c.current_version), "sent_at": naive(c.sent_at),
"link_expires_at": naive(c.link_expires_at), "signed_at": naive(c.signed_at),
"principal_statement_issued_at": naive(c.principal_statement_issued_at), "last_note": c.last_note,
"retention_until": c.retention_until, "next_due_at": naive(c.next_due_at),
"created_at": naive(c.created_at), "created_by": c.created_by, "version": int(c.version)}
cid = c.contract_id
out = {T.contract.name: [root],
T.contract_variable.name: [{"contract_id": cid, "name": n, "type": v.type, "value": v.value,
"currency": v.currency, "source": v.source, "resolved_at": v.resolved_at,
"frozen_in_version": c.frozen_in.get(n)} for n, v in sorted(c.variables.items())],
T.contract_section.name: [{"contract_id": cid, "position": int(s.position), "slot_no": int(s.slot_no),
"family": s.family, "clause_id": s.clause_id, "clause_version": int(s.clause_version),
"position_kind": s.position_kind, "deviation_level": int(s.deviation_level),
"overridden": bool(s.overridden), "required": bool(s.required), "title": s.title,
"body": s.body} for s in c.sections],
T.contract_version.name: [{"contract_id": cid, "version": int(v.version), "object_key": v.object_key,
"sha256": v.sha256, "size_bytes": int(v.size_bytes), "frozen_at": naive(v.frozen_at),
"frozen_by": v.frozen_by, "draft_fingerprint": v.draft_fingerprint,
"status": v.status, "text_body": v.text_body} for v in c.versions.values()],
T.signer.name: [{"contract_id": cid, "signer_no": int(s.signer_no), "kind": s.kind, "side": s.side,
"role": s.role, "signing_order": int(s.signing_order), "email": s.email,
"full_name": s.full_name, "title": s.title, "authority_basis": s.authority_basis,
"bp_id": int(s.bp_id) if s.bp_id is not None else None, "pool_sub": s.pool_sub,
"status": s.status, "link_id": s.link_id, "copy_link_id": s.copy_link_id,
"signed_at": naive(s.signed_at), "signed_version": s.signed_version,
"first_opened_at": naive(s.first_opened_at), "last_opened_at": naive(s.last_opened_at),
"delivered_at": naive(s.delivered_at), "chase_count": int(s.chase_count),
"last_chased_at": naive(s.last_chased_at)} for s in c.signers],
T.executed_copy.name: ([{"contract_id": cid, "object_key": c.executed.object_key, "sha256": c.executed.sha256,
"size_bytes": int(c.executed.size_bytes), "provenance": c.executed.provenance,
"signed_version": c.executed.signed_version, "signed_where": c.executed.signed_where,
"filed_by": c.executed.filed_by, "filed_at": naive(c.executed.filed_at),
"filename": c.executed.filename, "content_type": c.executed.content_type}]
if c.executed else []),
T.approval.name: [{"contract_id": cid, "seq": int(a.seq), "draft_fingerprint": a.draft_fingerprint,
"deviations_json": _j(a.deviations), "approver_group": a.approver_group,
"fallback_group": a.fallback_group, "original_group": a.original_group,
"due_at": naive(a.due_at), "state": a.state, "decided_by": a.decided_by,
"decided_at": naive(a.decided_at), "reason": a.reason, "requested_by": a.requested_by,
"requested_at": naive(a.requested_at), "escalated_at": naive(a.escalated_at)}
for a in c.approvals],
T.delivery.name: [{"contract_id": cid, "signer_no": int(d.signer_no), "kind": d.kind, "seq": int(d.seq),
"channel": d.channel, "capability": d.capability, "idempotency_key": d.idempotency_key,
"recipient": d.recipient, "subject": d.subject, "message": d.message, "plan": d.plan,
"reason": d.reason, "status": d.status, "provider_ref": d.provider_ref,
"done_by": d.done_by, "done_at": naive(d.done_at), "created_at": naive(d.created_at),
"retry_at": naive(d.retry_at)} for d in c.deliveries],
T.key_date.name: [{"contract_id": cid, "kind": k.kind, "seq": int(k.seq), "due_on": k.due_on,
"owner_group": k.owner_group, "days_before_json": _j(list(k.days_before)),
"reminded_json": _j(sorted(k.reminded)), "next_remind_on": k.next_remind_on,
"done_at": naive(k.done_at), "done_by": k.done_by, "note": k.note} for k in c.key_dates],
T.amendment.name: [{"contract_id": cid, "seq": int(a.seq), "kind": a.kind,
"child_contract_id": a.child_contract_id, "note": a.note, "recorded_by": a.recorded_by,
"recorded_at": naive(a.recorded_at)} for a in c.amendments]}
return out
def event_row(cid: str, e: C.SignatureEvent) -> dict:
return {"contract_id": cid, "seq": int(e.seq), "kind": e.kind, "signer_no": e.signer_no, "signer_kind": e.signer_kind,
"version": e.version, "version_sha256_shown": e.version_sha256_shown, "email": e.email,
"token_ref": e.token_ref, "pool_sub": e.pool_sub, "session_id": e.session_id, "ip": e.ip,
"user_agent": e.user_agent, "typed_name": e.typed_name, "consent_text_version": e.consent_text_version,
"consent_text": e.consent_text, "auth_method": e.auth_method, "occurred_at": naive(e.occurred_at),
"detail": e.detail, "prev_hash": e.prev_hash, "row_hash": e.row_hash}
def _event(r: dict) -> C.SignatureEvent:
return C.SignatureEvent(int(r["seq"]), r["kind"], aware(r["occurred_at"]),
int(r["signer_no"]) if r.get("signer_no") is not None else None, r.get("signer_kind"),
int(r["version"]) if r.get("version") is not None else None, r.get("version_sha256_shown"),
r.get("email"), r.get("token_ref"), r.get("pool_sub"), r.get("session_id"), r.get("ip"),
r.get("user_agent"), r.get("typed_name"), r.get("consent_text_version"),
r.get("consent_text"), r.get("auth_method"), r.get("detail"), r.get("prev_hash") or "",
r.get("row_hash") or "")
def build_contract(rows: dict) -> C.Contract:
r = rows[T.contract.name][0]
c = C.Contract(r["contract_id"], r["type_code"], r["template_id"], int(r["template_version"] or 0), int(r["bp_id"]),
r["source_ref"], r.get("owner") or "", r.get("created_by") or "", aware(r.get("created_at")),
title=r.get("title") or "", owner_group=r.get("owner_group") or "", context_ref=r.get("context_ref"),
state=r["state"], sensitivity=r.get("sensitivity") or "commercial",
execution=r.get("execution") or "SIMPLE", starts_on=_d(r.get("starts_on")),
ends_on=_d(r.get("ends_on")),
notice_days=int(r["notice_days"]) if r.get("notice_days") is not None else None,
amends_contract_id=r.get("amends_contract_id"), amendment_kind=r.get("amendment_kind"),
draft_fingerprint=r.get("draft_fingerprint") or "", request_hash=r.get("request_hash") or "",
missing=_uj(r.get("missing_json"), []), deviations=_uj(r.get("deviations_json"), []),
reference=r.get("reference") or "", current_version=int(r.get("current_version") or 0),
sent_at=aware(r.get("sent_at")), link_expires_at=aware(r.get("link_expires_at")),
signed_at=aware(r.get("signed_at")),
principal_statement_issued_at=aware(r.get("principal_statement_issued_at")),
last_note=r.get("last_note") or "", retention_until=_d(r.get("retention_until")),
next_due_at=aware(r.get("next_due_at")), version=int(r.get("version") or 0))
for v in rows.get(T.contract_variable.name, []):
c.variables[v["name"]] = VariableValue(v["name"], v["type"], v["value"], v.get("source") or "",
v.get("currency"), v.get("resolved_at"))
if v.get("frozen_in_version") is not None:
c.frozen_in[v["name"]] = int(v["frozen_in_version"])
c.sections = sorted((C.Section(int(s["position"]), s["family"], s["clause_id"], int(s["clause_version"]),
s["position_kind"], int(s["deviation_level"] or 0), s.get("title") or "",
s.get("body") or "", int(s.get("slot_no") or 0), bool(s.get("required")),
bool(s.get("overridden")))
for s in rows.get(T.contract_section.name, [])), key=lambda s: s.position)
for v in rows.get(T.contract_version.name, []):
c.versions[int(v["version"])] = C.Version(int(v["version"]), v.get("object_key") or "", v.get("sha256") or "",
int(v.get("size_bytes") or 0), aware(v.get("frozen_at")),
v.get("draft_fingerprint") or "", v.get("text_body") or "",
v.get("frozen_by") or "", v.get("status") or "ISSUED")
c.signers = sorted((C.Signer(int(s["signer_no"]), s["kind"], s["side"], int(s["signing_order"]), s.get("email") or "",
s.get("full_name") or "", s.get("title") or "", s.get("authority_basis") or "",
s.get("role") or "SIGNER", int(s["bp_id"]) if s.get("bp_id") is not None else None,
s.get("pool_sub"), s.get("status") or "PENDING", s.get("link_id"), s.get("copy_link_id"),
aware(s.get("signed_at")),
int(s["signed_version"]) if s.get("signed_version") is not None else None,
aware(s.get("first_opened_at")), aware(s.get("last_opened_at")),
aware(s.get("delivered_at")), int(s.get("chase_count") or 0),
aware(s.get("last_chased_at")))
for s in rows.get(T.signer.name, [])), key=lambda s: s.signer_no)
ex = rows.get(T.executed_copy.name) or []
if ex:
e = ex[0]
c.executed = C.ExecutedCopy(e["object_key"], e["sha256"], int(e.get("size_bytes") or 0), e["provenance"],
int(e["signed_version"]) if e.get("signed_version") is not None else None,
e.get("filed_by") or "", aware(e.get("filed_at")), e.get("filename") or "",
e.get("content_type") or "application/pdf", e.get("signed_where") or "")
c.approvals = sorted((C.Approval(int(a["seq"]), a["draft_fingerprint"], _uj(a.get("deviations_json"), []),
a.get("approver_group") or "", a.get("fallback_group") or "", aware(a.get("due_at")),
aware(a.get("requested_at")), a.get("requested_by") or "", a.get("state") or "pending",
a.get("decided_by"), aware(a.get("decided_at")), a.get("reason"),
aware(a.get("escalated_at")), a.get("original_group"))
for a in rows.get(T.approval.name, [])), key=lambda a: a.seq)
c.events = sorted((_event(e) for e in rows.get(T.signature_event.name, [])), key=lambda e: e.seq)
c.deliveries = sorted((C.Delivery(int(d["signer_no"]), d["kind"], int(d["seq"]), d.get("channel") or "manual",
d.get("recipient") or "", d.get("subject") or "", d.get("message") or "",
aware(d.get("created_at")), d.get("status") or "pending",
d.get("capability") or "mail.send", d.get("idempotency_key") or "",
d.get("plan") or "", d.get("reason") or "", d.get("provider_ref"), d.get("done_by"),
aware(d.get("done_at")), aware(d.get("retry_at")))
for d in rows.get(T.delivery.name, [])), key=lambda d: (d.signer_no, d.kind, d.seq))
c.key_dates = sorted((C.KeyDateRow(k["kind"], int(k["seq"]), _d(k["due_on"]), k.get("owner_group") or "",
tuple(_uj(k.get("days_before_json"), [])), list(_uj(k.get("reminded_json"), [])),
aware(k.get("done_at")), k.get("done_by"), k.get("note") or "")
for k in rows.get(T.key_date.name, [])), key=lambda k: (k.due_on, k.kind, k.seq))
c.amendments = sorted((C.AmendmentLink(int(a["seq"]), a["kind"], a.get("child_contract_id"), a.get("note") or "",
a.get("recorded_by") or "", aware(a.get("recorded_at")))
for a in rows.get(T.amendment.name, [])), key=lambda a: a.seq)
return c
class Head:
"""A contract's root row as attributes (owner, owner_group, sensitivity, state, source_ref, …)."""
def __init__(self, row: dict):
self.__dict__.update(row)
self.version = int(row.get("version") or 0)
def originated_by(self, namespaces: list) -> bool:
from brick_contracts.domain.values import namespace_of
ns = namespace_of(self.source_ref or "")
return bool(ns) and ns != "manual" and ns in namespaces
class ContractRepository:
PARTS = (T.CONTRACT_ROOT,) + T.CONTRACT_PARTS + (T.EVENTS,)
def __init__(self, conn):
self.conn = conn
def _load(self, ids: list) -> dict:
grouped: dict = defaultdict(lambda: {p.name: [] for p in self.PARTS})
if not ids:
return {}
for part in self.PARTS:
t = part.table
stmt = select(t).where(t.c.contract_id == ids[0]) if len(ids) == 1 else select(t).where(t.c.contract_id.in_(ids))
for r in self.conn.execute(stmt):
row = dict(r._mapping)
grouped[row["contract_id"]][t.name].append(row)
return grouped
def _attach(self, c: C.Contract, rows: dict) -> C.Contract:
snap = contract_rows(c)
c._snapshot = {name: _norm(rs) for name, rs in snap.items()}
return c
def get(self, contract_id: str) -> Optional[C.Contract]:
rows = self._load([contract_id]).get(contract_id)
if not rows or not rows[T.contract.name]:
return None
return self._attach(build_contract(rows), rows)
def load_many(self, ids: Iterable[str]) -> list:
wanted = sorted(set(ids))
out = []
for cid, rows in self._load(wanted).items():
if rows[T.contract.name]:
out.append(self._attach(build_contract(rows), rows))
return sorted(out, key=lambda c: c.contract_id)
def head(self, contract_id: str):
"""The root row alone (ONE SELECT) — enough to decide visibility, a replay or a state before loading the
whole aggregate (eleven SELECTs)."""
r = self.conn.execute(select(T.contract).where(T.contract.c.contract_id == contract_id)).first()
return Head(dict(r._mapping)) if r is not None else None
def exists(self, contract_id: str) -> bool:
return self.conn.execute(select(T.contract.c.contract_id).where(
T.contract.c.contract_id == contract_id)).first() is not None
def save(self, c: C.Contract, *, now: Optional[datetime] = None) -> int:
"""Save what changed: the root (always: it claims the next optimistic version) and only the parts whose rows
differ from what was loaded; then the new events as one multi-row plain INSERT (append-only)."""
rows = contract_rows(c)
snap = getattr(c, "_snapshot", None)
if snap is None: # a new contract: its key is derived and new, so no stale rows to find
parts = [p for p in T.CONTRACT_PARTS if rows[p.name]]
else:
parts = [p for p in T.CONTRACT_PARTS if _norm(rows[p.name]) != snap.get(p.name)]
agg = Aggregate(T.CONTRACT_ROOT, *parts, version="version")
written = agg.save(self.conn, {T.contract.name: rows[T.contract.name],
**{p.name: rows[p.name] for p in parts}},
now=now, expected_version=int(c.version))
if c.new_events:
try:
self.conn.execute(insert(T.signature_event).values([event_row(c.contract_id, e) for e in c.new_events]))
except IntegrityError as exc:
if is_duplicate(exc): # a racer took this seq: the chain is decided by the claim
raise Conflict(T.signature_event.name, {"contract_id": c.contract_id}) from None
raise
c.version = int(written)
c.new_events = []
rows[T.contract.name][0]["version"] = c.version
c._snapshot = {name: _norm(rs) for name, rs in rows.items()}
return c.version
def remove(self, contract_id: str) -> None:
T.CONTRACT.remove(self.conn, {"contract_id": contract_id})
self.conn.execute(delete(T.signature_event).where(T.signature_event.c.contract_id == contract_id))
# ── set-based reads ─────────────────────────────────────────────────────────────────────
def headers(self, *, where=None, limit: int = 200) -> list:
c = T.contract.c
stmt = select(c.contract_id, c.type_code, c.title, c.state, c.bp_id, c.owner, c.owner_group, c.sensitivity,
c.starts_on, c.ends_on, c.current_version, c.source_ref, c.origin, c.amends_contract_id,
c.amendment_kind, c.signed_at, c.sent_at, c.version, c.reference)
if where is not None:
stmt = stmt.where(where)
stmt = stmt.order_by(c.contract_id).limit(limit)
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def next_key_dates(self, ids: list) -> dict:
if not ids:
return {}
k = T.key_date.c
out: dict = {}
for r in self.conn.execute(select(k.contract_id, k.kind, k.due_on).where(
and_(k.contract_id.in_(ids), k.done_at.is_(None)))):
cur = out.get(r[0])
if cur is None or _d(r[2]) < cur["due_on"]:
out[r[0]] = {"kind": r[1], "due_on": _d(r[2])}
return out
def due(self, now: datetime, limit: int = 500) -> list:
c = T.contract.c
return [r[0] for r in self.conn.execute(select(c.contract_id).where(
and_(c.next_due_at.is_not(None), c.next_due_at <= naive(now))).order_by(c.next_due_at).limit(limit))]
def key_dates_due(self, today: date, limit: int = 1000) -> list:
k = T.key_date.c
return sorted({r[0] for r in self.conn.execute(select(k.contract_id).where(
and_(k.done_at.is_(None), k.next_remind_on.is_not(None), k.next_remind_on <= today)).limit(limit))})
def retention_due(self, today: date, limit: int = 1000) -> list:
c = T.contract.c
return [r[0] for r in self.conn.execute(select(c.contract_id).where(
and_(c.retention_until.is_not(None), c.retention_until < today)).limit(limit))]
def signed_by(self, *, email: str = "", sub: str = "") -> list:
s = T.signer.c
conds = [s.email == email.lower()] if email else []
if sub:
conds.append(s.pool_sub == sub)
if not conds:
return []
return sorted({r[0] for r in self.conn.execute(select(s.contract_id).where(
and_(s.kind == "INTERNAL", or_(*conds))))})
def with_signer_email(self, email: str) -> list:
s = T.signer.c
return sorted({r[0] for r in self.conn.execute(select(s.contract_id).where(s.email == email.lower()))})
def approvals_for(self, groups: Iterable[str]) -> list:
groups = sorted({g for g in groups if g})
if not groups:
return []
a = T.approval.c
stmt = select(a.contract_id, a.seq, a.state, a.approver_group, a.fallback_group, a.due_at, a.draft_fingerprint,
a.deviations_json, a.requested_by, a.requested_at).where(
or_(and_(a.state == "pending", a.approver_group.in_(groups)),
and_(a.state == "escalated", a.fallback_group.in_(groups))))
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def clause_usage(self, clause_id: str, version: Optional[int] = None) -> int:
s = T.contract_section.c
cond = s.clause_id == clause_id
if version is not None:
cond = and_(cond, s.clause_version == int(version))
return int(self.conn.execute(select(func.count(func.distinct(s.contract_id))).where(cond)).scalar() or 0)
def clause_usage_by_version(self, clause_id: str) -> dict:
s = T.contract_section.c
return {int(r[0]): int(r[1]) for r in self.conn.execute(
select(s.clause_version, func.count(func.distinct(s.contract_id))).where(s.clause_id == clause_id)
.group_by(s.clause_version))}
def register(self, ids: Optional[list] = None, open_only: bool = True) -> list:
k = T.key_date.c
stmt = select(T.key_date)
conds = []
if ids is not None:
if not ids:
return []
conds.append(k.contract_id.in_(ids))
if open_only:
conds.append(k.done_at.is_(None))
if conds:
stmt = stmt.where(and_(*conds))
return [dict(r._mapping) for r in self.conn.execute(stmt.order_by(k.due_on))]
# ── the library ────────────────────────────────────────────────────────────────────────────
def _cached(conn, token: str, kind: str, ident, version, build):
key = (_engine_key(conn), token, kind, ident, version)
with _LOCK:
if key in _LIBRARY:
return _LIBRARY[key]
value = build()
if value is not None:
with _LOCK:
if len(_LIBRARY) > 4096:
_LIBRARY.clear()
_LIBRARY[key] = value
return value
class ClauseRepository:
def __init__(self, conn):
self.conn = conn
def get(self, clause_id: str) -> Optional[Clause]:
rows = T.CLAUSE.load(self.conn, {"clause_id": clause_id})
if not rows:
return None
r = rows[T.clause.name][0]
c = Clause(r["clause_id"], r["family"], r["position"], int(r["deviation_level"] or 0), r.get("category") or "",
r.get("status") or "active")
for v in rows[T.clause_version.name]:
c.versions[int(v["version"])] = ClauseVersion(int(v["version"]), v["title"], v["body"],
v.get("created_by") or "", aware(v.get("created_at")))
return c
def save(self, c: Clause, *, now: datetime) -> None:
for n in c.new_versions: # a new version is CLAIMED: a racer cutting the same n gets 409
v = c.versions[n]
try:
self.conn.execute(insert(T.clause_version).values(clause_id=c.clause_id, version=n, title=v.title,
body=v.body, created_by=v.created_by,
created_at=naive(v.created_at or now)))
except IntegrityError as exc:
if is_duplicate(exc):
raise Conflict(T.clause_version.name, {"clause_id": c.clause_id}) from None
raise
c.new_versions = []
T.CLAUSE.save(self.conn, {
T.clause.name: [{"clause_id": c.clause_id, "family": c.family, "position": c.position,
"deviation_level": int(c.deviation_level), "category": c.category, "status": c.status,
"created_at": naive(now)}],
T.clause_version.name: [{"clause_id": c.clause_id, "version": int(v.version), "title": v.title,
"body": v.body, "created_by": v.created_by,
"created_at": naive(v.created_at or now)} for v in c.versions.values()]}, now=now)
def all(self) -> list:
out = {}
for r in self.conn.execute(select(T.clause)):
d = dict(r._mapping)
out[d["clause_id"]] = Clause(d["clause_id"], d["family"], d["position"], int(d["deviation_level"] or 0),
d.get("category") or "", d.get("status") or "active")
for r in self.conn.execute(select(T.clause_version)):
d = dict(r._mapping)
if d["clause_id"] in out:
out[d["clause_id"]].versions[int(d["version"])] = ClauseVersion(
int(d["version"]), d["title"], d["body"], d.get("created_by") or "", aware(d.get("created_at")))
return sorted((c for c in out.values() if c.versions), key=lambda c: (c.family, c.position, c.clause_id))
def words(self, pairs: list, token: str) -> dict:
"""(clause_id, version) → (title, body, position, level) for every pair, in two SELECTs (cached by token)."""
out, missing = {}, []
for cid, v in pairs:
hit = _LIBRARY.get((_engine_key(self.conn), token, "clause_words", cid, v))
if hit is not None:
out[(cid, v)] = hit
else:
missing.append((cid, v))
if missing:
ids = sorted({cid for cid, _ in missing})
heads = {r[0]: (r[1], int(r[2] or 0)) for r in self.conn.execute(
select(T.clause.c.clause_id, T.clause.c.position, T.clause.c.deviation_level).where(
T.clause.c.clause_id.in_(ids)))}
for r in self.conn.execute(select(T.clause_version).where(T.clause_version.c.clause_id.in_(ids))):
d = dict(r._mapping)
k = (d["clause_id"], int(d["version"]))
if k in missing and d["clause_id"] in heads:
pos, lvl = heads[d["clause_id"]]
out[k] = (d["title"], d["body"], pos, lvl)
with _LOCK:
_LIBRARY[(_engine_key(self.conn), token, "clause_words", k[0], k[1])] = out[k]
return out
class TemplateRepository:
def __init__(self, conn):
self.conn = conn
def get(self, template_id: str) -> Optional[Template]:
rows = T.TEMPLATE.load(self.conn, {"template_id": template_id})
return _template(rows) if rows else None
def version(self, template_id: str, version: int, token: str) -> Optional[tuple]:
"""(Template head, TemplateVersion), cached under the config token (any library write bumps it)."""
def build():
t = self.get(template_id)
if t is None or int(version) not in t.versions:
return None
return t, t.versions[int(version)]
return _cached(self.conn, token, "template", template_id, int(version), build)
def save(self, t: Template, *, now: datetime) -> None:
for n in t.new_versions:
v = t.versions[n]
try:
self.conn.execute(insert(T.template_version).values(
template_id=t.template_id, version=n, status=v.status, authored_by=v.authored_by,
authored_at=naive(v.authored_at or now), note=v.note))
except IntegrityError as exc:
if is_duplicate(exc):
raise Conflict(T.template_version.name, {"template_id": t.template_id}) from None
raise
t.new_versions = []
tid = t.template_id
T.TEMPLATE.save(self.conn, {
T.template.name: [{"template_id": tid, "type_code": t.type_code, "name": t.name, "created_at": naive(now)}],
T.template_version.name: [{"template_id": tid, "version": int(v.version), "status": v.status,
"authored_by": v.authored_by, "authored_at": naive(v.authored_at),
"approved_by": v.approved_by, "approved_at": naive(v.approved_at),
"retired_at": naive(v.retired_at), "note": v.note} for v in t.versions.values()],
T.template_slot.name: [{"template_id": tid, "version": int(v.version), "slot_no": int(s.slot_no),
"family": s.family, "kind": s.kind, "alternative_of_slot": s.alternative_of_slot,
"clause_id": s.clause_id, "clause_version": int(s.clause_version),
"condition_json": _j(s.condition)} for v in t.versions.values() for s in v.slots],
T.variable_def.name: [{"template_id": tid, "version": int(v.version), "name": d.name, "type": d.type,
"source": d.source, "required": bool(d.required), "label": d.label,
"validation_json": _j(d.validation), "enum_values_json": _j(list(d.enum_values)),
"aliases_json": _j(d.aliases), "condition_json": _j(d.condition),
"default_value": d.default_value, "currency_var": d.currency_var}
for v in t.versions.values() for d in v.variables.values()]}, now=now)
def all(self) -> list:
heads = [dict(r._mapping) for r in self.conn.execute(select(T.template))]
out = []
for h in heads:
t = self.get(h["template_id"])
if t is not None:
out.append(t)
return sorted(out, key=lambda t: t.template_id)
def pointers(self) -> dict:
"""type_code → [(template_id, current APPROVED version or None)] — two SELECTs, for the snapshot."""
types = {r[0]: r[1] for r in self.conn.execute(select(T.template.c.template_id, T.template.c.type_code))}
best: dict = {}
for r in self.conn.execute(select(T.template_version.c.template_id, T.template_version.c.version,
T.template_version.c.status)):
if r[2] == T_APPROVED:
best[r[0]] = max(best.get(r[0], 0), int(r[1]))
out: dict = defaultdict(list)
for tid, tc in sorted(types.items()):
out[tc].append((tid, best.get(tid)))
return dict(out)
def using_clause(self, clause_id: str) -> list:
ids = sorted({r[0] for r in self.conn.execute(select(T.template_slot.c.template_id).where(
T.template_slot.c.clause_id == clause_id))})
return [t for t in (self.get(i) for i in ids) if t is not None]
def _template(rows: dict) -> Template:
h = rows[T.template.name][0]
t = Template(h["template_id"], h["type_code"], h.get("name") or h["template_id"])
for v in rows[T.template_version.name]:
t.versions[int(v["version"])] = TemplateVersion(int(v["version"]), v.get("status") or "DRAFT",
v.get("authored_by") or "", aware(v.get("authored_at")),
v.get("approved_by"), aware(v.get("approved_at")),
aware(v.get("retired_at")), v.get("note") or "")
for s in rows[T.template_slot.name]:
tv = t.versions.get(int(s["version"]))
if tv is not None:
tv.slots.append(Slot(int(s["slot_no"]), s["family"], s["kind"], s["clause_id"], int(s["clause_version"]),
int(s["alternative_of_slot"]) if s.get("alternative_of_slot") is not None else None,
_uj(s.get("condition_json"))))
for d in rows[T.variable_def.name]:
tv = t.versions.get(int(d["version"]))
if tv is not None:
tv.variables[d["name"]] = VariableDef(d["name"], d["type"], d["source"], bool(d.get("required")),
d.get("label") or d["name"], _uj(d.get("validation_json"), {}) or {},
tuple(_uj(d.get("enum_values_json"), []) or []),
_uj(d.get("aliases_json"), {}) or {}, _uj(d.get("condition_json")),
d.get("default_value"), d.get("currency_var") or "currency")
for tv in t.versions.values():
tv.slots.sort(key=lambda s: s.slot_no)
return t
class ContractTypeRepository:
def __init__(self, conn):
self.conn = conn
def save(self, ct, *, by: str, now: datetime) -> None:
upsert(self.conn, T.ctype, [{**ct.as_row(), "updated_at": naive(now), "updated_by": by}], ("type_code",))
# ── configuration: one snapshot per config token ───────────────────────────────────────────
class ConfigRepository:
def __init__(self, conn):
self.conn = conn
def token(self) -> str:
r = self.conn.execute(select(T.config_version.c.token).where(T.config_version.c.scope == "config")).first()
return (r[0] or "") if r else ""
def load(self) -> Configuration:
token = self.token()
key = _engine_key(self.conn)
with _LOCK:
hit = _SNAPSHOTS.get(key)
if hit is not None and hit[0] == token:
return hit[1]
cfg = self._build(token)
with _LOCK:
_SNAPSHOTS[key] = (token, cfg)
return cfg
def _rows(self, table) -> list:
return [dict(r._mapping) for r in self.conn.execute(select(table))]
def _build(self, token: str) -> Configuration:
rules: dict = defaultdict(list)
for r in self._rows(T.approval_policy):
rules[r["policy_code"]].append(rule_from_row(r))
policies = {}
for code, rs in rules.items():
try:
policies[code] = ApprovalPolicy(code, tuple(sorted(rs, key=lambda r: r.rule_no)))
except ValueError:
continue # an incomplete policy is not usable; keep_settings refuses one
return Configuration(
token=token,
types={r["type_code"]: type_from_row(r) for r in self._rows(T.ctype)},
policies=policies,
cadences={r["cadence_code"]: cadence_from_row(r) for r in self._rows(T.reminder_cadence)},
retention={r["rule_code"]: retention_from_row(r) for r in self._rows(T.retention_rule)},
consents={r["version"]: ConsentText(r["version"], r["text"]) for r in self._rows(T.consent_text)},
consumers={r["namespace"]: consumer_from_row(r) for r in self._rows(T.consumer)},
settings={r["setting_key"]: r.get("value") for r in self._rows(T.settings)},
templates=TemplateRepository(self.conn).pointers())
def bump(self, now: datetime) -> str:
token = uuid.uuid4().hex
upsert(self.conn, T.config_version, [{"scope": "config", "token": token, "changed_at": naive(now)}], ("scope",))
return token
def save_rows(self, kind: str, rows: list, *, now: datetime) -> None:
table, key = T.CONFIG_TABLES[kind]
if not rows:
return
stamped = [{**r, "updated_at": naive(now)} if "updated_at" in table.c else dict(r) for r in rows]
if kind == "approval_policy":
# A policy is replaced whole: rules no longer listed for a policy_code are removed (by full key).
for code in sorted({r["policy_code"] for r in rows}):
keep = {int(r["rule_no"]) for r in rows if r["policy_code"] == code}
stored = {int(r[0]) for r in self.conn.execute(select(T.approval_policy.c.rule_no).where(
T.approval_policy.c.policy_code == code))}
for n in sorted(stored - keep):
self.conn.execute(delete(T.approval_policy).where(and_(
T.approval_policy.c.policy_code == code, T.approval_policy.c.rule_no == n)))
upsert(self.conn, table, stamped, key)
def rows(self, kind: str) -> list:
table, _ = T.CONFIG_TABLES[kind]
return self._rows(table)
def forget_snapshots() -> None:
"""Tests: drop every cached snapshot and library object (a new engine never collides anyway)."""
with _LOCK:
_SNAPSHOTS.clear()
_LIBRARY.clear()
# ── bytes, logs, claims ────────────────────────────────────────────────────────────────────
class ObjectStore:
"""Version and executed-copy bytes in the bucket (ct_object_chunk), keyed contracts/<contract_id>/…; read only
by a use case that checked the caller may see them, and streamed — never a link."""
def __init__(self, conn):
self.conn = conn
def put(self, key: str, content: bytes, content_type: str) -> tuple:
digest = hashlib.sha256(content).hexdigest()
self.conn.execute(delete(T.object_chunk).where(T.object_chunk.c.object_key == key))
enc = base64.b64encode(content).decode()
parts = [enc[i:i + CHUNK] for i in range(0, len(enc), CHUNK)] or [""]
now = datetime.now(timezone.utc).replace(tzinfo=None)
self.conn.execute(insert(T.object_chunk).values([
{"object_key": key, "part_no": i, "content_b64": part, "sha256": digest, "size_bytes": len(content),
"content_type": content_type, "stored_at": now} for i, part in enumerate(parts)]))
return digest, len(content)
def get(self, key: str) -> Optional[tuple]:
rows = [dict(r._mapping) for r in self.conn.execute(select(T.object_chunk).where(
T.object_chunk.c.object_key == key))]
if not rows:
return None
rows.sort(key=lambda r: int(r["part_no"]))
content = base64.b64decode("".join(r["content_b64"] or "" for r in rows))
return content, rows[0].get("content_type") or "application/octet-stream", rows[0].get("sha256") or ""
def delete_prefix(self, prefix: str) -> int:
keys = sorted({r[0] for r in self.conn.execute(select(T.object_chunk.c.object_key).where(
T.object_chunk.c.object_key.like(prefix + "%")))})
for k in keys:
self.conn.execute(delete(T.object_chunk).where(T.object_chunk.c.object_key == k))
return len(keys)
class RetentionLog:
def __init__(self, conn):
self.conn = conn
def record(self, run_date: date, contract_id: str, category: str, detail: str) -> bool:
"""Once per (day, contract, category): a plain INSERT; a second sweep the same day writes nothing."""
try:
self.conn.execute(insert(T.retention_log).values(run_date=run_date, contract_id=contract_id,
category=category, detail=detail[:1000],
purged_at=datetime.now(timezone.utc).replace(tzinfo=None)))
return True
except IntegrityError as exc:
if is_duplicate(exc):
return False
raise
def rows(self, contract_id: Optional[str] = None) -> list:
stmt = select(T.retention_log)
if contract_id:
stmt = stmt.where(T.retention_log.c.contract_id == contract_id)
return [dict(r._mapping) for r in self.conn.execute(stmt)]
def done(self, run_date: date, contract_id: str, category: str) -> bool:
r = T.retention_log.c
return self.conn.execute(select(r.contract_id).where(and_(r.run_date == run_date, r.contract_id == contract_id,
r.category == category))).first() is not None
class TickClaims:
def __init__(self, conn):
self.conn = conn
def claim(self, job: str, slot: str) -> bool:
try:
self.conn.execute(insert(T.tick_claim).values(job=job, slot=slot,
claimed_at=datetime.now(timezone.utc).replace(tzinfo=None)))
return True
except IntegrityError as exc:
if is_duplicate(exc):
return False
raise
__all__ = ["ContractRepository", "ClauseRepository", "TemplateRepository", "ContractTypeRepository",
"ConfigRepository", "ObjectStore", "RetentionLog", "TickClaims", "forget_snapshots", "build_contract",
"contract_rows", "naive", "aware", "is_duplicate"]