"""One repository per aggregate: rows ↔ domain objects, through erp_platform.repository (load whole, save whole,
idempotently, with the optimistic version where the aggregate has one). No lazy loading; nothing here loads two
aggregates for one decision. The bulk reads at the end (seats, people) are READ MODELS for reports, the run board
and row visibility — they never feed a save.
"""
import base64
import hashlib
import json
import uuid
import weakref
from contextlib import contextmanager
from contextvars import ContextVar
from collections import defaultdict
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Iterable, Optional
from sqlalchemy import and_, delete, func, insert, or_, select
from sqlalchemy.exc import IntegrityError
from erp_platform import retry
from erp_platform.repository import Aggregate, Part, upsert
from brick_hr.domain.configuration import CATALOGUES, Config
from brick_hr.domain.events import Event
from brick_hr.domain.intake import Intake
from brick_hr.domain.org import Org, PositionSlice, UnitSlice
from brick_hr.domain.payroll import Batch, Line
from brick_hr.domain.person import Document, Engagement, Person
from brick_hr.domain import playbook as pbk
from brick_hr.domain.packs import designed_key
from brick_hr.domain.playbook import Playbook, PlaybookStep
from brick_hr.domain.privacy import PrivacyRequest
from brick_hr.domain.run import Run, RunLog, RunStep
from brick_hr.domain.sensitive import Sensitive
from brick_hr.domain.slices import Slice, Timeline
from brick_hr.infrastructure import batching as B
from brick_hr.infrastructure import tables as T
CHUNK = 192 * 1024
def _utc(v):
"""Stored times as naive UTC (SQLite naive; the pg wire aware — README, 'Known platform behaviour')."""
if isinstance(v, datetime) and v.tzinfo is not None:
return v.astimezone(timezone.utc).replace(tzinfo=None)
return v
def _aware(v):
v = _utc(v)
return v.replace(tzinfo=timezone.utc) if isinstance(v, datetime) else v
def _store_time(v):
"""Business times are aware UTC in the domain; stored naive UTC."""
return _utc(v) if isinstance(v, datetime) else v
def _j(v) -> Optional[str]:
return None if v is None else json.dumps(v, sort_keys=True, 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 _d(v) -> Optional[date]:
if v is None or isinstance(v, date) and not isinstance(v, datetime):
return v
if isinstance(v, datetime):
return v.date()
return date.fromisoformat(str(v)[:10])
def is_duplicate(exc: BaseException) -> bool:
"""A duplicate key — also when the unit of work answered it as the 409 at COMMIT (use_cases._on_engine keeps
the IntegrityError as the cause), so a claim lost at commit is still a claim lost, never a crash."""
seen = 0
while exc is not None and seen < 5:
code = retry.sqlstate(exc)
text = str(exc).lower()
if code == "23505" or "unique constraint failed" in text or "duplicate key" in text:
return True
exc, seen = exc.__cause__, seen + 1
return False
class _Tracked:
"""A repository that remembers what it loaded in this unit of work, so a save writes only what changed
(infrastructure/batching.py says why that is storage). Loads are one round trip on the pg wire."""
def __init__(self, conn):
self.conn = conn
self._seen: dict = {}
self._pre: dict = {} # root key -> results read ahead (HrRepository.prefetch), used once
def _load(self, agg, root_key: dict, likely_new: bool = False) -> Optional[dict]:
pre = self._pre.pop(tuple(root_key[k] for k in agg.root_key), None)
if pre is not None:
rows, snap = B.from_results(agg, root_key, pre)
self._seen[tuple(root_key[k] for k in agg.root_key)] = snap
return rows
"""`likely_new`: ask for the root alone first (one statement instead of one per table) — for a load whose
usual answer is "not there", like a hire's person. The answer is the same either way."""
if likely_new:
rt = agg.root.table
if self.conn.execute(select(*(rt.c[k] for k in agg.root_key)).where(
and_(*(rt.c[k] == root_key[k] for k in agg.root_key)))).first() is None:
self._seen[tuple(root_key[k] for k in agg.root_key)] = B.absent(agg, root_key)
return None
rows, snap = B.load(self.conn, agg, root_key)
self._seen[tuple(root_key[k] for k in agg.root_key)] = snap
return rows
def _absent(self, agg, root_key: dict) -> None:
self._seen.setdefault(tuple(root_key[k] for k in agg.root_key), B.absent(agg, root_key))
def _save(self, agg, rows: dict, expected_version: int) -> int:
root = rows[agg.root.name][0]
k = tuple(root[c] for c in agg.root_key)
written, snap = B.save(self.conn, agg, rows, expected_version=expected_version, snapshot=self._seen.get(k))
if snap is not None:
self._seen[k] = snap
else:
self._seen.pop(k, None)
return written
def _forget(self, agg, root_key: dict) -> None:
self._seen.pop(tuple(root_key[k] for k in agg.root_key), None)
# ── slices ─────────────────────────────────────────────────────────────────────────────────
def _slice(row: dict, fields: tuple) -> Slice:
return Slice(_d(row["valid_from"]), _d(row["valid_to"]), {f: row.get(f) for f in fields}, int(row["rev"]),
row.get("superseded_by"), _aware(row.get("recorded_at")), _aware(row.get("superseded_at")),
row.get("recorded_by") or "", row.get("reason") or "")
def _slice_row(s: Slice, key: dict, fields: tuple) -> dict:
out = {**key, "valid_from": s.valid_from, "rev": int(s.rev), "valid_to": s.valid_to,
"superseded_by": s.superseded_by, "recorded_at": _store_time(s.recorded_at),
"superseded_at": _store_time(s.superseded_at), "recorded_by": s.recorded_by, "reason": s.reason}
for f in fields:
out[f] = s.data.get(f)
return out
DETAIL = ("legal_given_name", "legal_family_name", "preferred_name", "personal_email", "personal_phone",
"work_email", "work_phone")
ASSIGN = ("org_id", "org_unit_id", "position_id", "job_code", "location_code", "schedule_code", "employment_pct",
"employee_subgroup", "cost_centre", "manager_override_person_id")
PAY = ("rate_basis", "rate_amount", "currency", "frequency", "pay_grade")
COMP = ("rate_basis", "rate_amount", "currency")
COMPL = ("pack", "outcome", "checked_on", "checked_by", "follow_up_on", "evidence_ref", "reasons", "detail_json")
class PersonRepository(_Tracked):
def get(self, person_id: int, likely_new: bool = False) -> Optional[Person]:
rows = self._load(T.PERSON, {"person_id": int(person_id)}, likely_new)
return None if rows is None else _person(rows)
def save(self, p: Person) -> int:
pid = p.person_id
rows = {
"hr_person": [{"person_id": pid, "person_key": p.person_key, "staff_no": p.staff_no,
"bp_id": p.bp_id, "pool_user_sub": p.pool_user_sub,
**({"created_at": _store_time(p.created_at)} if p.created_at else {})}],
"hr_person_detail": [_slice_row(s, {"person_id": pid}, DETAIL) for s in p.detail.slices],
"hr_engagement": [{"person_id": pid, "n": e.n, "worker_type": e.worker_type,
"legal_entity_ref": e.legal_entity_ref, "start_date": e.start_date,
"planned_end": e.planned_end, "actual_end": e.actual_end, "leave_reason": e.leave_reason,
"supplier_bp_id": e.supplier_bp_id,
**({"created_at": _store_time(e.created_at)} if e.created_at else {})}
for e in p.engagements.values()],
"hr_assignment": [_slice_row(s, {"person_id": pid, "n": n}, ASSIGN)
for n, t in p.assignments.items() for s in t.slices],
"hr_pay": [_slice_row(s, {"person_id": pid, "n": n}, PAY) for n, t in p.pay.items() for s in t.slices],
"hr_pay_component": [_slice_row(s, {"person_id": pid, "n": n, "component": c}, COMP)
for (n, c), t in p.components.items() for s in t.slices],
"hr_event": [{"person_id": pid, "n": e.n, "seq": e.seq, "event_type": e.event_type, "kind": e.kind,
"reason": e.reason, "effective_date": e.effective_date, "actor": e.actor,
"recorded_at": _store_time(e.recorded_at), "slices_written": _j(e.slices_written),
"run_id": e.run_id, "bp_synced_at": _store_time(e.bp_synced_at),
"payload_json": _j(e.payload)} for e in p.events],
"hr_document": [{"person_id": pid, "doc_id": d.doc_id, "document_type": d.document_type,
"access_class": d.access_class, "filename": d.filename, "content_type": d.content_type,
"object_key": d.object_key, "sha256": d.sha256, "size_bytes": d.size_bytes,
"expires_on": d.expires_on, "contract_id": d.contract_id, "uploaded_by": d.uploaded_by,
"uploaded_at": _store_time(d.uploaded_at)} for d in p.documents.values()],
"hr_compliance_record": [_slice_row(s, {"person_id": pid, "n": n, "kind": k}, COMPL)
for (n, k), t in p.compliance.items() for s in t.slices],
}
p.version = self._save(T.PERSON, rows, p.version)
p.new_events = []
return p.version
def remove(self, person_id: int) -> None:
self._forget(T.PERSON, {"person_id": int(person_id)})
T.PERSON.remove(self.conn, {"person_id": int(person_id)})
def find_me(self, sub: Optional[str], email: Optional[str]) -> Optional[int]:
"""find_by_sub, else find_by_work_email — both asked at once (one round trip); the sub wins."""
if not sub or not email:
return (self.find_by_sub(sub) if sub else None) or (self.find_by_work_email(email) if email else None)
by_sub, by_email = B.select_many(self.conn, [self._by_sub(sub), self._by_email(email)])
if by_sub:
return int(by_sub[0]["person_id"])
return int(by_email[0]["person_id"]) if by_email else None
@staticmethod
def _by_sub(sub: str):
return select(T.person.c.person_id).where(T.person.c.pool_user_sub == sub).limit(1)
@staticmethod
def _by_email(email: str):
return select(T.person_detail.c.person_id).where(and_(
func.lower(T.person_detail.c.work_email) == email.strip().lower(),
T.person_detail.c.superseded_by.is_(None))).limit(1)
def find_by_sub(self, sub: str) -> Optional[int]:
r = self.conn.execute(select(T.person.c.person_id).where(T.person.c.pool_user_sub == sub)).first()
return int(r[0]) if r else None
def find_by_work_email(self, email: str) -> Optional[int]:
if not email:
return None
r = self.conn.execute(select(T.person_detail.c.person_id).where(and_(
func.lower(T.person_detail.c.work_email) == email.strip().lower(),
T.person_detail.c.superseded_by.is_(None)))).first()
return int(r[0]) if r else None
def ids(self) -> list:
return sorted(int(r[0]) for r in self.conn.execute(select(T.person.c.person_id)))
def load_many(self, ids: Optional[Iterable[int]] = None) -> list:
"""Every person (or these), one SELECT per table — a read model for reports and the board."""
wanted = None if ids is None else sorted({int(i) for i in ids})
if wanted == []:
return []
if wanted is not None: # erp_platform 0.9.6: one query per table per 100 roots
loaded = T.PERSON.load_many(self.conn, [{"person_id": i} for i in wanted])
return [_person(rows) for _k, rows in sorted(loaded.items()) if rows]
grouped: dict = defaultdict(lambda: {p.name: [] for p in T.PERSON.parts})
stmts = [select(p.table) for p in T.PERSON.parts]
for part, rows in zip(T.PERSON.parts, B.select_many(self.conn, stmts)):
for row in rows:
grouped[int(row["person_id"])][part.name].append(row)
return [_person(rows) for pid, rows in sorted(grouped.items()) if rows["hr_person"]]
def seats(self) -> list:
"""Every active assignment slice with its engagement's worker type — the Directory read model."""
from brick_hr.domain.services import Seat
pre = self._pre.pop("seats", None)
engagements, assignments = pre if pre is not None else B.select_many(self.conn, self.seat_statements())
eng = {(int(r["person_id"]), int(r["n"])): r for r in engagements}
out = []
for d in assignments:
pct = d.get("employment_pct")
e = eng.get((int(d["person_id"]), int(d["n"]))) or {}
out.append(Seat(int(d["person_id"]), int(d["n"]), _d(d["valid_from"]), _d(d["valid_to"]), d.get("org_id"),
d.get("org_unit_id"), d.get("position_id"), d.get("job_code"), d.get("location_code"),
d.get("schedule_code"), Decimal(str(pct)) if pct is not None else None,
int(d["manager_override_person_id"]) if d.get("manager_override_person_id") else None,
e.get("worker_type") or "", d.get("cost_centre"), e.get("legal_entity_ref"),
_d(e.get("start_date")), _d(e.get("actual_end") or e.get("planned_end"))))
return out
@staticmethod
def seat_statements() -> list:
return [select(T.engagement.c.person_id, T.engagement.c.n, T.engagement.c.worker_type,
T.engagement.c.legal_entity_ref, T.engagement.c.start_date, T.engagement.c.actual_end,
T.engagement.c.planned_end),
select(T.assignment).where(T.assignment.c.superseded_by.is_(None))]
def names(self, ids: Iterable[int], on: date) -> dict:
"""person_id → (preferred or legal given name + family name, work email) on a date — names only."""
ids = sorted({int(i) for i in ids if i})
if not ids:
return {}
rows = [dict(r._mapping) for r in self.conn.execute(select(T.person_detail).where(and_(
T.person_detail.c.person_id.in_(ids), T.person_detail.c.superseded_by.is_(None))))]
out = {}
for pid in ids:
mine = sorted((r for r in rows if int(r["person_id"]) == pid), key=lambda r: _d(r["valid_from"]))
cur = next((r for r in mine if _d(r["valid_from"]) <= on <= _d(r["valid_to"])), mine[0] if mine else None)
if cur:
out[pid] = {"name": f"{cur.get('preferred_name') or cur.get('legal_given_name') or ''} "
f"{cur.get('legal_family_name') or ''}".strip(),
"work_email": cur.get("work_email") or ""}
return out
def events_pending(self) -> list:
"""Events whose run is not expanded yet, or whose business-partner sync has not happened: the engine's
to-do list (eventual consistency after the event committed)."""
rows = self.conn.execute(select(T.event.c.person_id, T.event.c.n, T.event.c.seq, T.event.c.run_id,
T.event.c.kind, T.event.c.effective_date, T.event.c.bp_synced_at))
return [dict(r._mapping) for r in rows]
def engagements(self) -> list:
return [dict(r._mapping) for r in self.conn.execute(select(T.engagement))]
# ── the tick's set-based passes: WHO might be due, by a query, so only they are loaded ──────────────────
def warning_candidates(self, today: date, days_end: int, days_rtw: int) -> set:
"""People who MAY need a warning today — a superset of what engine.warnings decides on: an open-ended
engagement whose planned end falls in the window, or a right-to-work record whose follow-up does."""
from datetime import timedelta
e, c = T.engagement, T.compliance
ends, checks = B.select_many(self.conn, [
select(e.c.person_id).where(and_(e.c.planned_end.is_not(None), e.c.actual_end.is_(None),
e.c.planned_end >= today,
e.c.planned_end <= today + timedelta(days=days_end))),
select(c.c.person_id).where(and_(c.c.kind == "right_to_work_check", c.c.follow_up_on.is_not(None),
c.c.follow_up_on >= today,
c.c.follow_up_on <= today + timedelta(days=days_rtw)))])
return {int(r["person_id"]) for r in ends + checks}
def ended_before(self, today: date) -> set:
"""People with an engagement that ended before today — a superset of who retention can be due for (it
needs EVERY engagement ended before today, which engine.retention_sweep decides on the loaded person)."""
e = T.engagement
return {int(r["person_id"]) for r in B.select_many(self.conn, [
select(e.c.person_id).where(and_(e.c.actual_end.is_not(None), e.c.actual_end < today))])[0]}
def _person(rows: dict) -> Person:
h = rows["hr_person"][0]
p = Person(int(h["person_id"]), h["person_key"], staff_no=h.get("staff_no"),
bp_id=int(h["bp_id"]) if h.get("bp_id") is not None else None, pool_user_sub=h.get("pool_user_sub"),
version=int(h.get("version") or 0), created_at=_aware(h.get("created_at")))
p.detail = Timeline("the person's detail", [_slice(r, DETAIL) for r in rows["hr_person_detail"]])
for r in rows["hr_engagement"]:
n = int(r["n"])
p.engagements[n] = Engagement(n, r["worker_type"], r.get("legal_entity_ref") or "", _d(r["start_date"]),
_d(r.get("planned_end")), _d(r.get("actual_end")), r.get("leave_reason"),
int(r["supplier_bp_id"]) if r.get("supplier_bp_id") is not None else None,
_aware(r.get("created_at")))
p.assignments[n] = Timeline(f"engagement {n}'s assignment")
p.pay[n] = Timeline(f"engagement {n}'s pay")
for r in rows["hr_assignment"]:
s = _slice(r, ASSIGN)
if s.data.get("employment_pct") is not None:
s.data["employment_pct"] = Decimal(str(s.data["employment_pct"]))
if s.data.get("manager_override_person_id") is not None:
s.data["manager_override_person_id"] = int(s.data["manager_override_person_id"])
p.assignments.setdefault(int(r["n"]), Timeline("")).slices.append(s)
for r in rows["hr_pay"]:
s = _slice(r, PAY)
if s.data.get("rate_amount") is not None:
s.data["rate_amount"] = Decimal(str(s.data["rate_amount"])).quantize(Decimal("0.0001"))
p.pay.setdefault(int(r["n"]), Timeline("")).slices.append(s)
for r in rows["hr_pay_component"]:
s = _slice(r, COMP)
s.data["rate_amount"] = Decimal(str(s.data["rate_amount"])).quantize(Decimal("0.0001"))
key = (int(r["n"]), r["component"])
p.components.setdefault(key, Timeline(f"engagement {key[0]}'s {key[1]}")).slices.append(s)
for r in sorted(rows["hr_event"], key=lambda r: (int(r["n"]), int(r["seq"]))):
p.events.append(Event(int(r["n"]), int(r["seq"]), r["event_type"], r["kind"], r.get("reason"),
_d(r["effective_date"]), r.get("actor") or "", _aware(r.get("recorded_at")),
_uj(r.get("slices_written"), []), r.get("run_id"), _aware(r.get("bp_synced_at")),
_uj(r.get("payload_json"), {})))
for r in rows["hr_document"]:
p.documents[r["doc_id"]] = Document(r["doc_id"], r.get("document_type") or "", r.get("access_class") or "",
r.get("filename") or "", r.get("content_type") or "",
r.get("object_key") or "", r.get("sha256") or "",
int(r.get("size_bytes") or 0), r.get("uploaded_by") or "",
_aware(r.get("uploaded_at")), _d(r.get("expires_on")), r.get("contract_id"))
for r in rows["hr_compliance_record"]:
s = _slice(r, COMPL)
for k in ("checked_on", "follow_up_on"):
s.data[k] = _d(s.data.get(k))
key = (int(r["n"]), r["kind"])
p.compliance.setdefault(key, Timeline(f"engagement {key[0]}'s {key[1]}")).slices.append(s)
return p
# ── Sensitive ──────────────────────────────────────────────────────────────────────────────
class SensitiveRepository(_Tracked):
def get(self, person_id: int) -> Optional[Sensitive]:
rows = self._load(T.SENSITIVE, {"person_id": int(person_id)})
if rows is None:
return None
h = rows["hr_sensitive"][0]
s = Sensitive(int(h["person_id"]), int(h.get("version") or 0), _aware(h.get("erased_at")),
_aware(h.get("anonymised_at")))
for r in rows["hr_person_sensitive"]:
sl = _slice(r, ())
sl.data = {"value": _uj(r.get("value_json"), {}), "object_key": r.get("object_key"),
"sha256": r.get("sha256")}
s.timeline(r["kind"]).slices.append(sl)
return s
def save(self, s: Sensitive) -> int:
rows = {"hr_sensitive": [{"person_id": s.person_id, "erased_at": _store_time(s.erased_at),
"anonymised_at": _store_time(s.anonymised_at)}],
"hr_person_sensitive": [
{**_slice_row(sl, {"person_id": s.person_id, "kind": kind}, ()),
"value_json": _j(sl.data.get("value")), "object_key": sl.data.get("object_key"),
"sha256": sl.data.get("sha256"), "created_by": sl.recorded_by}
for kind, t in s.kinds.items() for sl in t.slices]}
s.version = self._save(T.SENSITIVE, rows, s.version)
return s.version
def bank_slices(self) -> dict:
"""person_id → bank slices (for the payroll extract's bank changes; finance-only use case)."""
out: dict = defaultdict(list)
for r in self.conn.execute(select(T.person_sensitive).where(and_(
T.person_sensitive.c.kind == "bank", T.person_sensitive.c.superseded_by.is_(None)))):
d = dict(r._mapping)
sl = _slice(d, ())
sl.data = {"value": _uj(d.get("value_json"), {})}
out[int(d["person_id"])].append(sl)
return out
# ── Org ────────────────────────────────────────────────────────────────────────────────────
class OrgRepository(_Tracked):
def get(self, org_id: str) -> Optional[Org]:
rows = self._load(T.ORG, {"org_id": org_id})
return None if rows is None else _org(rows)
def all(self) -> dict:
"""Every org tree — from the configuration snapshot (an org change bumps its version)."""
snap = configuration_snapshot(self.conn)
grouped: dict = defaultdict(lambda: {p.name: [] for p in T.ORG.parts})
for part in T.ORG.parts:
for row in snap[part.name]:
grouped[row["org_id"]][part.name].append(dict(row))
return {k: _org(v) for k, v in grouped.items() if v["hr_org"]}
def save(self, o: Org) -> int:
rows = {"hr_org": [{"org_id": o.org_id, "name": o.name, "legal_entity_ref": o.legal_entity_ref}],
"hr_org_unit": [{"org_id": o.org_id, **{k: getattr(u, k) for k in (
"org_unit_id", "valid_from", "valid_to", "name", "parent_org_unit_id", "org_unit_type",
"cost_centre", "chief_position_id", "location_code")}} for u in o.units],
"hr_position": [{"org_id": o.org_id, **{k: getattr(p, k) for k in (
"position_id", "valid_from", "valid_to", "org_unit_id", "job_code", "fte_pct", "cost_centre",
"vacant_policy")}} for p in o.positions]}
o.version = self._save(T.ORG, rows, o.version)
changed_configuration(self.conn, T.org.name, T.org_unit.name, T.position.name)
return o.version
def _org(rows: dict) -> Org:
h = rows["hr_org"][0]
o = Org(h["org_id"], h.get("name") or "", h.get("legal_entity_ref") or "", int(h.get("version") or 0))
for r in rows["hr_org_unit"]:
o.units.append(UnitSlice(r["org_unit_id"], _d(r["valid_from"]), _d(r["valid_to"]), r.get("name") or "",
r.get("parent_org_unit_id"), r.get("org_unit_type"), r.get("cost_centre"),
r.get("chief_position_id"), r.get("location_code")))
for r in rows["hr_position"]:
o.positions.append(PositionSlice(r["position_id"], _d(r["valid_from"]), _d(r["valid_to"]),
r.get("org_unit_id") or "", r.get("job_code"), r.get("fte_pct"),
r.get("cost_centre"), r.get("vacant_policy")))
return o
# ── Configuration ──────────────────────────────────────────────────────────────────────────
# THE CONFIGURATION SNAPSHOT. Catalogues, settings, playbooks and the org trees change rarely and are read by
# nearly every unit of work (a View needs the time zone, the groups, the calendars, the org tree): 26 tables,
# which on DataK3 were 26 round trips per unit of work, several units per request. They are read once per
# VERSION of each table and kept in this process as an immutable snapshot (every reader gets copies).
#
# The version is a TOKEN per table in hr_config_version (key = the table's name), rewritten with a fresh random
# value in the SAME transaction as every change to that table (keep_configuration, install_localisation_pack,
# keep_playbook, keep_org). A unit of work reads every token in one SELECT and re-reads only the tables whose
# token is not the one their cached rows were read under — a change to the job catalogue re-reads hr_job, not 26
# tables.
# Why a random token and not a counter: two changes committing at once would both write n+1 under
# last-writer-wins, and a reader that cached between them would never see the second. A fresh value per change
# is never equal to one read before it, so ANY commit after a snapshot is noticed by the next unit of work.
# Why a read per request and not a TTL: replicas do not share memory, so only the bucket can say "changed", and
# the next request after a change must see it (a TTL would serve the old catalogue for its length).
# The tokens are read BEFORE the tables, so cached rows can only be older than their token, never newer.
# A table with no token row has never been changed through hr since this table existed (a bucket from before
# it): its rows are cached under "no token" until the first change writes one.
# A unit of work that changed the configuration reads the tables itself for the rest of that unit (it must not
# fill the cache with what may roll back).
#
# ONE READ PER REQUEST. A request is often several units of work (a hire: the person, then its run, then the
# partner); the tokens are read by the first and reused by the rest of THAT request (request_scope, set around
# every use case by application/use_cases.py). A change made by the request itself drops the memo, so its own
# later units read the new tokens. Outside a request scope (a script, a test calling a function directly) they
# are read once per unit of work.
_SNAPSHOTS: "weakref.WeakKeyDictionary" = weakref.WeakKeyDictionary() # engine -> {table: (token, rows)}
_REQUEST: ContextVar = ContextVar("brick_hr_request", default=None)
_PER_UNIT: "weakref.WeakKeyDictionary" = weakref.WeakKeyDictionary() # connection -> snapshot (read once)
_CHANGED: "weakref.WeakKeyDictionary" = weakref.WeakKeyDictionary() # connection -> tables it changed
NO_TOKEN = ""
def _snapshot_tables() -> list:
out = [(name, T.CONFIG_TABLES[cat.table][0]) for name, cat in CATALOGUES.items()]
return out + [(t.name, t) for t in (T.playbook, T.playbook_step, T.org, T.org_unit, T.position, T.setting_row)]
def _read_tables(conn, tables: list) -> dict:
results = B.select_many(conn, [select(t) for _n, t in tables])
snap = {}
for (name, _t), rows in zip(tables, results):
cat = CATALOGUES.get(name)
if cat is not None:
for d in rows:
for f in cat.lists:
d[f] = _uj(d.get(f), [])
snap[name] = tuple(rows)
return snap
_TOKENS_PER_UNIT: "weakref.WeakKeyDictionary" = weakref.WeakKeyDictionary() # connection -> tokens read ahead
def _known_tokens(conn) -> Optional[dict]:
"""The versions already read for this request (or this unit of work), or None."""
request = _REQUEST.get()
engine = getattr(conn, "engine", None)
if request is not None and engine is not None and request.get("engine") is engine:
return request["tokens"]
return _TOKENS_PER_UNIT.get(conn)
def needs_tokens(conn) -> bool:
"""Would this unit of work read the configuration's versions? (HrRepository.prefetch reads them along.)"""
return conn not in _CHANGED and _PER_UNIT.get(conn) is None and _known_tokens(conn) is None
def remember_tokens(conn, rows: list) -> None:
tokens = {r["key"]: r["token"] for r in rows}
request = _REQUEST.get()
engine = getattr(conn, "engine", None)
if request is not None and engine is not None:
request.update(engine=engine, tokens=tokens)
_TOKENS_PER_UNIT[conn] = tokens
def configuration_snapshot(conn) -> dict:
"""{catalogue or table name: (row, ...)} — READ ONLY; copy a row before changing it."""
tables = _snapshot_tables()
if conn in _CHANGED:
return _read_tables(conn, tables)
mine = _PER_UNIT.get(conn)
if mine is not None:
return mine
engine = getattr(conn, "engine", None)
tokens = _known_tokens(conn)
if tokens is None:
tokens = {r[0]: r[1] for r in conn.execute(select(T.config_version.c.key, T.config_version.c.token))}
remember_tokens(conn, [{"key": k, "token": v} for k, v in tokens.items()])
held = dict(_SNAPSHOTS.get(engine) or {}) if engine is not None else {}
stale = [(name, t) for name, t in tables
if name not in held or held[name][0] != tokens.get(t.name, NO_TOKEN)]
if stale:
fresh = _read_tables(conn, stale)
for name, t in stale:
held[name] = (tokens.get(t.name, NO_TOKEN), fresh[name])
if engine is not None:
_SNAPSHOTS[engine] = held # a new dict: readers on other threads keep theirs
snap = {name: held[name][1] for name, _t in tables}
_PER_UNIT[conn] = snap
return snap
def changed_configuration(conn, *tables: str) -> None:
"""Mark these configuration tables changed, in this transaction: a new token for each (once per unit of
work)."""
done = _CHANGED.get(conn)
if done is None:
done = _CHANGED[conn] = set()
todo = [t for t in tables if t not in done]
_PER_UNIT.pop(conn, None)
_TOKENS_PER_UNIT.pop(conn, None)
request = _REQUEST.get()
if request is not None:
request.clear()
if not todo:
return
done.update(todo)
now = datetime.now(timezone.utc).replace(tzinfo=None)
upsert(conn, T.config_version, [{"key": t, "token": uuid.uuid4().hex, "changed_at": now} for t in todo],
"key", insert_only=())
@contextmanager
def request_scope():
"""One configuration-version read for everything inside (one request)."""
tok = _REQUEST.set({})
try:
yield
finally:
_REQUEST.reset(tok)
def forget_configuration_cache() -> None:
"""Drop every process-held snapshot (tests)."""
_SNAPSHOTS.clear()
class ConfigurationRepository:
def __init__(self, conn):
self.conn = conn
self._cached: Optional[Config] = None
def load(self) -> Config:
if self._cached is None:
snap = configuration_snapshot(self.conn)
self._cached = Config({name: [{k: (list(v) if isinstance(v, list) else v) for k, v in r.items()}
for r in snap[name]] for name in CATALOGUES},
dated=snap.get(T.setting_row.name, ()))
return self._cached
def add_dated(self, row: dict) -> None:
"""Append one dated setting row (HR-55) with a plain INSERT: rows are never edited, so a second row for the same
key, scope and day is a duplicate key (the caller answers 409) and the table is the history."""
changed_configuration(self.conn, T.setting_row.name)
self.conn.execute(insert(T.setting_row).values(**row))
if self._cached is not None:
self._cached.dated.append(dict(row))
def save_row(self, catalogue: str, row: dict) -> None:
self.save_rows(catalogue, [row])
def save_rows(self, catalogue: str, rows: list) -> None:
"""Keep these rows of one catalogue: one multi-row upsert (a config row has no creation columns, and no
children, so this is what one Aggregate(Part(...)).save per row did, in one statement)."""
if not rows:
return
cat = CATALOGUES[catalogue]
t, key = T.CONFIG_TABLES[cat.table]
by_key: dict = {}
for row in rows:
stored = {k: row.get(k) for k in key}
for f in cat.fields:
if f in row:
stored[f] = _j(row[f]) if f in cat.lists else row[f]
by_key[tuple(str(stored[k]) for k in key)] = stored # the last word on a key wins
changed_configuration(self.conn, t.name)
upsert(self.conn, t, list(by_key.values()), key, insert_only=())
if self._cached is not None:
# what this unit of work now holds (there is no read-your-writes inside the transaction)
last = {tuple(str(r.get(k)) for k in key): dict(r) for r in rows}
kept = [r for r in self._cached.rows[catalogue] if tuple(str(r.get(k)) for k in key) not in last]
self._cached.rows[catalogue] = kept + list(last.values())
def remove_row(self, catalogue: str, key_values: dict) -> None:
cat = CATALOGUES[catalogue]
t, key = T.CONFIG_TABLES[cat.table]
changed_configuration(self.conn, t.name)
Aggregate(Part(t, key)).remove(self.conn, {k: key_values[k] for k in key})
if self._cached is not None:
want = tuple(str(key_values[k]) for k in key)
self._cached.rows[catalogue] = [r for r in self._cached.rows[catalogue]
if tuple(str(r.get(k)) for k in key) != want]
def uses(self, catalogue: str, key: dict) -> int:
"""How many records use a catalogue row (a reason in use cannot be removed, only retired)."""
if catalogue == "event_reason":
return int(self.conn.execute(select(func.count()).select_from(T.event).where(and_(
T.event.c.event_type == key["event_type"], T.event.c.reason == key["reason"]))).scalar() or 0)
if catalogue == "worker_type":
return int(self.conn.execute(select(func.count()).select_from(T.engagement).where(
T.engagement.c.worker_type == key["worker_type"])).scalar() or 0)
if catalogue == "job":
return int(self.conn.execute(select(func.count()).select_from(T.assignment).where(
T.assignment.c.job_code == key["job_code"])).scalar() or 0)
if catalogue == "event_type":
return int(self.conn.execute(select(func.count()).select_from(T.event).where(
T.event.c.event_type == key["event_type"])).scalar() or 0)
return 0
# ── Playbook ───────────────────────────────────────────────────────────────────────────────
class PlaybookRepository:
def __init__(self, conn):
self.conn = conn
def get(self, playbook_id: str, version: int) -> Optional[Playbook]:
rows = T.PLAYBOOK.load(self.conn, {"playbook_id": playbook_id, "version": int(version)})
return None if rows is None else _playbook(rows["hr_playbook"][0], rows["hr_playbook_step"])
def latest_version(self, playbook_id: str) -> int:
r = self.conn.execute(select(func.max(T.playbook.c.version)).where(
T.playbook.c.playbook_id == playbook_id)).scalar()
return int(r or 0)
def latest(self) -> list:
"""The latest version of every playbook (what selection chooses among) — from the configuration snapshot
(a new playbook version bumps its version)."""
snap = configuration_snapshot(self.conn)
roots = [dict(r) for r in snap["hr_playbook"]]
best: dict = {}
for r in roots:
if r["playbook_id"] not in best or int(r["version"]) > int(best[r["playbook_id"]]["version"]):
best[r["playbook_id"]] = r
if not best:
return []
steps = [dict(r) for r in snap["hr_playbook_step"]]
return [_playbook(root, [s for s in steps if s["playbook_id"] == pid and int(s["version"]) == int(root["version"])])
for pid, root in sorted(best.items())]
def versions(self, keys: Iterable[tuple]) -> dict:
"""(playbook_id, version) → Playbook, for any versions (open runs keep old ones) — from the configuration
snapshot, which holds every version: no statement of its own."""
want = {(str(k[0]), int(k[1])) for k in keys}
if not want:
return {}
snap = configuration_snapshot(self.conn)
steps: dict = defaultdict(list)
for r in snap["hr_playbook_step"]:
k = (r["playbook_id"], int(r["version"]))
if k in want:
steps[k].append(dict(r))
return {(r["playbook_id"], int(r["version"])): _playbook(dict(r), steps[(r["playbook_id"], int(r["version"]))])
for r in snap["hr_playbook"] if (r["playbook_id"], int(r["version"])) in want}
def backfill_keys(self) -> int:
"""Write, on every step stored before 0.4.0, the key it is already read with (_playbook) — so the key is in
the row, not only derived. Idempotent: a second run finds nothing. Returns the step rows written. It
rewrites the configuration token like any playbook change."""
c = T.playbook_step.c
todo = self.conn.execute(select(c.playbook_id, c.version).where(
or_(c.step_key.is_(None), c.step_key == "")).distinct()).all()
rows = []
for pid, v in sorted(todo):
pb = self.get(pid, int(v))
stored = {int(r["step_no"]): dict(r) for r in self.conn.execute(select(T.playbook_step).where(
and_(c.playbook_id == pid, c.version == int(v)))).mappings()}
rows += [{**stored[s.step_no], "step_key": s.key} for s in pb.steps if not stored[s.step_no].get("step_key")]
if rows:
changed_configuration(self.conn, T.playbook.name, T.playbook_step.name)
upsert(self.conn, T.playbook_step, rows, ("playbook_id", "version", "step_no"))
return len(rows)
def create(self, pb: Playbook) -> None:
"""A NEW version: its root row is a plain INSERT — two admins saving version n+1 at once: one wins, the
other's transaction fails 23505 (409 'someone saved a new version first')."""
changed_configuration(self.conn, T.playbook.name, T.playbook_step.name)
sink = B._Collect()
sink.execute(insert(T.playbook).values(
playbook_id=pb.playbook_id, version=int(pb.version), name=pb.name, event_types=_j(pb.event_types),
worker_types=_j(pb.worker_types), reasons=_j(pb.reasons), org_unit_id=pb.org_unit_id, status=pb.status,
created_by=pb.created_by, created_at=_store_time(pb.created_at)))
upsert(sink, T.playbook_step, [
{"playbook_id": pb.playbook_id, "version": int(pb.version), "step_no": int(s.step_no), "kind": s.kind,
"title": s.title, "responsible_group": s.responsible_group, "capability": s.capability,
"input_template": _j(s.input_template), "timing_rule": _j(s.timing), "depends_on": _j(s.depends_on),
"condition": _j(s.condition), "approval_group": s.approval_group,
"draft_instruction": s.draft_instruction, "template_kind": s.template_kind, "fan_out": s.fan_out,
"step_key": s.key, "outcomes": _j(s.outcomes) if s.outcomes else None}
for s in pb.steps], ("playbook_id", "version", "step_no"), existing=set())
B.execute_many(self.conn, sink.stmts) # the root's plain INSERT first: the claim
def _playbook(root: dict, steps: list) -> Playbook:
pb = Playbook(root["playbook_id"], int(root["version"]), root.get("name") or "", _uj(root.get("event_types"), []),
_uj(root.get("worker_types"), []), _uj(root.get("reasons"), []), root.get("org_unit_id"),
root.get("status") or "active", root.get("created_by") or "", _aware(root.get("created_at")))
for s in sorted(steps, key=lambda s: int(s["step_no"])):
pb.steps.append(PlaybookStep(int(s["step_no"]), s["kind"], s.get("title") or "", s.get("responsible_group") or "",
_uj(s.get("timing_rule"), {}), s.get("capability"), _uj(s.get("input_template")),
_uj(s.get("depends_on"), []), _uj(s.get("condition")), s.get("approval_group"),
s.get("draft_instruction"), s.get("template_kind"), s.get("fan_out"),
s.get("step_key") or "", _uj(s.get("outcomes"), []) or []))
# A row stored before 0.4.0 has no key: it is keyed on read — the pack's designed key while it is still the
# pack's step, else a slug of its title — the same answer every time, so ids stay stable until the backfill.
for s in pb.steps:
if not s.key:
s.key = designed_key(pb.playbook_id, s.step_no, s.title) or ""
pbk.assign_keys(pb.steps)
return pb
# ── Run ────────────────────────────────────────────────────────────────────────────────────
STEP_COLS = ("sort_no", "kind", "title", "responsible_group", "playbook_step_no", "capability", "plan", "manual",
"manual_reason", "state", "attempts", "last_status", "provider_ref", "contract_id", "draft_text",
"done_by", "approval_group", "destructive", "message", "approved", "target", "template_kind",
"draft_instruction")
class RunRepository(_Tracked):
def get(self, run_id: str) -> Optional[Run]:
rows = self._load(T.RUN, {"run_id": run_id})
return None if rows is None else _run(rows)
def exists(self, run_id: str) -> bool:
found = self.conn.execute(select(T.run.c.run_id).where(T.run.c.run_id == run_id)).first() is not None
if not found:
self._absent(T.RUN, {"run_id": run_id})
return found
def save(self, r: Run) -> int:
rows = {"hr_run": [{"run_id": r.run_id, "direction": r.direction, "person_id": r.person_id, "n": r.n,
"event_seq": r.event_seq, "playbook_id": r.playbook_id,
"playbook_version": r.playbook_version, "created_at": _store_time(r.created_at),
"anchors_json": _j({k: (v.isoformat() if hasattr(v, "isoformat") else v)
for k, v in r.anchors.items()}),
"cancelled_at": _store_time(r.cancelled_at), "cancel_reason": r.cancel_reason,
"cancelled_by": r.cancelled_by}],
"hr_run_step": [{"run_id": r.run_id, "step_no": s.step_no,
**{c: getattr(s, c) for c in STEP_COLS},
"template_json": _j(s.template), "input_json": _j(s.input),
"timing_rule": _j(s.timing), "due_at": _store_time(s.due_at),
"result_json": _j(s.result), "done_at": _store_time(s.done_at),
"depends_on": _j(s.depends_on), "condition": _j(s.condition),
"next_check_at": _store_time(s.next_check_at),
"outcomes": _j(s.outcomes) if s.outcomes else None} for s in r.steps.values()],
"hr_run_party": [{"run_id": r.run_id, "group_code": g, "person_id": pid, "set_by": by}
for g, (pid, by) in r.parties.items()],
"hr_run_log": [{"run_id": r.run_id, "seq": l.seq, "what": l.what, "from_value": l.from_value,
"to_value": l.to_value, "actor": l.actor, "event_at": _store_time(l.event_at)}
for l in r.log]}
self._pre.pop("open", None)
r.version = self._save(T.RUN, rows, r.version)
return r.version
def needing_attention(self, now: datetime) -> list:
"""The tick's hot path: runs with a step whose time (due, or next check) has come."""
return sorted({r[0] for r in self.conn.execute(_due(now))})
def tick_start(self, now: datetime) -> tuple:
"""(events_pending, run ids, runs needing attention) in one round trip — what a tick starts from."""
e = T.event
events, runs, due = B.select_many(self.conn, [
select(e.c.person_id, e.c.n, e.c.seq, e.c.run_id, e.c.kind, e.c.effective_date, e.c.bp_synced_at),
select(T.run.c.run_id), _due(now)])
return events, sorted(r["run_id"] for r in runs), sorted({r["run_id"] for r in due})
def ids(self, person_id: Optional[int] = None) -> list:
pre = self._pre.pop(("ids", int(person_id)), None) if person_id is not None else None
if pre is not None:
return sorted(r["run_id"] for r in pre)
stmt = select(T.run.c.run_id)
if person_id is not None:
stmt = stmt.where(T.run.c.person_id == int(person_id))
return sorted(r[0] for r in self.conn.execute(stmt))
# The runs with an open step, read once for a unit of work (prefetch(open_runs=True)): what My tasks and
# row visibility both start from. Dropped as soon as this unit of work saves a run.
@staticmethod
def open_run_statements() -> list:
open_ids = select(T.run_step.c.run_id).where(T.run_step.c.state.in_(("pending", "waiting")))
return [select(p.table).where(p.table.c.run_id.in_(open_ids)) for p in T.RUN.parts]
def _open_rows(self) -> Optional[dict]:
pre = self._pre.get("open")
if pre is None:
return None
grouped: dict = defaultdict(lambda: {p.name: [] for p in T.RUN.parts})
for part, rows in zip(T.RUN.parts, pre):
for row in rows:
grouped[row["run_id"]][part.name].append(row)
return grouped
def load_many(self, ids: Optional[Iterable[str]] = None) -> list:
wanted = None if ids is None else sorted(set(ids))
if wanted == []:
return []
opened = self._open_rows() if wanted is not None else None
if opened is not None and set(wanted) <= set(opened):
return [_run(opened[rid]) for rid in wanted if opened[rid]["hr_run"]]
if wanted is not None: # erp_platform 0.9.6: one query per table per 100 roots
loaded = T.RUN.load_many(self.conn, [{"run_id": i} for i in wanted])
return [_run(rows) for _k, rows in sorted(loaded.items()) if rows]
grouped: dict = defaultdict(lambda: {p.name: [] for p in T.RUN.parts})
stmts = [select(p.table) for p in T.RUN.parts]
for part, rows in zip(T.RUN.parts, B.select_many(self.conn, stmts)):
for row in rows:
grouped[row["run_id"]][part.name].append(row)
return [_run(rows) for _rid, rows in sorted(grouped.items()) if rows["hr_run"]]
def remove(self, run_id: str) -> None:
self._forget(T.RUN, {"run_id": run_id})
T.RUN.remove(self.conn, {"run_id": run_id})
def open_steps(self) -> list:
"""(run_id, person_id, group, parties) for every unfinished step — read model for 'my groups' visibility.
An approval counts for its approval group as well as its responsible group."""
opened = self._open_rows()
if opened is not None:
run_rows = [r for g in opened.values() for r in g["hr_run"]]
party_rows = [r for g in opened.values() for r in g["hr_run_party"]]
step_rows = [r for g in opened.values() for r in g["hr_run_step"] if r["state"] in ("pending", "waiting")]
return self._open_steps(run_rows, party_rows, step_rows)
run_rows, party_rows, step_rows = B.select_many(self.conn, [
select(T.run.c.run_id, T.run.c.person_id, T.run.c.cancelled_at),
select(T.run_party.c.run_id, T.run_party.c.group_code, T.run_party.c.person_id),
select(T.run_step.c.run_id, T.run_step.c.step_no, T.run_step.c.responsible_group,
T.run_step.c.approval_group).where(T.run_step.c.state.in_(("pending", "waiting")))])
return self._open_steps(run_rows, party_rows, step_rows)
@staticmethod
def _open_steps(run_rows, party_rows, step_rows) -> list:
runs = {r["run_id"]: (int(r["person_id"]), r["cancelled_at"]) for r in run_rows}
parties: dict = defaultdict(dict)
for r in party_rows:
parties[r["run_id"]][r["group_code"]] = int(r["person_id"]) if r["person_id"] is not None else None
out = []
for d in step_rows:
r = (d["run_id"], d["step_no"], d["responsible_group"], d["approval_group"])
if r[0] not in runs:
continue
pid, cancelled = runs[r[0]]
if cancelled is not None and not str(r[1]).startswith("R"):
continue
for g in {r[2], r[3]} - {None}:
out.append((r[0], pid, g, parties.get(r[0], {})))
return out
def _due(now: datetime):
"""Steps whose time has come — pending with its due time passed, or waiting with a next check that has —
filtered in the query on due_at / next_check_at, so a done, cancelled or not-yet-due step is never read."""
t = T.run_step
when = func.coalesce(t.c.next_check_at, t.c.due_at)
return select(t.c.run_id).where(and_(
or_(t.c.state == "pending", and_(t.c.state == "waiting", t.c.next_check_at.is_not(None))),
when.is_not(None), when <= _store_time(now))).distinct()
def _run(rows: dict) -> Run:
h = rows["hr_run"][0]
anchors = {}
for k, v in (_uj(h.get("anchors_json"), {}) or {}).items():
if isinstance(v, str) and "T" in v:
anchors[k] = _aware(datetime.fromisoformat(v))
elif isinstance(v, str) and v:
anchors[k] = date.fromisoformat(v)
r = Run(h["run_id"], h.get("direction") or "", int(h["person_id"]), int(h.get("n") or 0), h.get("event_seq") or "",
h.get("playbook_id"), int(h["playbook_version"]) if h.get("playbook_version") is not None else None,
_aware(h.get("created_at")), anchors, int(h.get("version") or 0), _aware(h.get("cancelled_at")),
h.get("cancel_reason"), h.get("cancelled_by"))
for s in rows["hr_run_step"]:
st = RunStep(s["step_no"], int(s.get("sort_no") or 0), s.get("kind") or "task", s.get("title") or "",
s.get("responsible_group") or "")
for c in STEP_COLS:
v = s.get(c)
if c in ("manual", "destructive"):
v = bool(v)
elif c in ("attempts", "sort_no"):
v = int(v or 0)
elif c == "playbook_step_no":
v = int(v) if v is not None else None
elif c == "approved":
v = None if v is None else bool(v)
elif c in ("plan", "manual_reason", "message"):
v = v or ""
setattr(st, c, v)
st.template, st.input = _uj(s.get("template_json")), _uj(s.get("input_json"), {})
st.timing, st.due_at = _uj(s.get("timing_rule"), {}), _aware(s.get("due_at"))
st.result, st.done_at = _uj(s.get("result_json"), {}), _aware(s.get("done_at"))
st.depends_on, st.condition = _uj(s.get("depends_on"), []), _uj(s.get("condition"))
st.next_check_at = _aware(s.get("next_check_at"))
st.outcomes = _uj(s.get("outcomes"), []) or []
r.steps[st.step_no] = st
for p in rows["hr_run_party"]:
r.parties[p["group_code"]] = (int(p["person_id"]) if p.get("person_id") is not None else None,
p.get("set_by") or "")
for l in sorted(rows["hr_run_log"], key=lambda l: int(l["seq"])):
r.log.append(RunLog(int(l["seq"]), l.get("what") or "", l.get("from_value"), l.get("to_value"),
l.get("actor") or "", _aware(l.get("event_at"))))
return r
class StepClaims:
def __init__(self, conn):
self.conn = conn
def claim(self, run_id: str, step_no: str, attempt: int) -> None:
"""A plain INSERT — never an upsert: a second concurrent claim must FAIL (23505), here or at COMMIT."""
self.conn.execute(insert(T.step_claim).values(run_id=run_id, step_no=step_no, attempt=int(attempt),
claimed_at=datetime.now(timezone.utc).replace(tzinfo=None)))
def remove_run(self, run_id: str) -> None:
self.conn.execute(delete(T.step_claim).where(T.step_claim.c.run_id == run_id))
# ── Intake ─────────────────────────────────────────────────────────────────────────────────
class IntakeRepository(_Tracked):
def get(self, intake_id: str) -> Optional[Intake]:
from brick_hr.domain.intake import parse_intake_id
of = self._pre.pop(("of", parse_intake_id(intake_id)[0]), None) if "|" in str(intake_id) else None
if of is not None: # every intake of this person was read ahead
self._pre[(intake_id,)] = [[r for r in of if r["intake_id"] == intake_id]]
rows = self._load(T.INTAKE, {"intake_id": intake_id})
if rows is None:
return None
h = rows["hr_intake"][0]
return Intake(h["intake_id"], int(h["person_id"]), int(h["n"]), h.get("worker_type") or "", h.get("pack") or "",
h.get("link_ref"), _aware(h.get("issued_at")), _aware(h.get("expires_at")),
_aware(h.get("submitted_at")), _uj(h.get("missing_fields"), []), int(h.get("chases") or 0),
int(h.get("version") or 0))
def save(self, i: Intake) -> int:
i.version = self._save(T.INTAKE, {"hr_intake": [{
"intake_id": i.intake_id, "person_id": i.person_id, "n": i.n, "worker_type": i.worker_type,
"pack": i.pack, "link_ref": i.link_ref, "issued_at": _store_time(i.issued_at),
"expires_at": _store_time(i.expires_at), "submitted_at": _store_time(i.submitted_at),
"missing_fields": _j(i.missing_fields), "chases": i.chases}]}, i.version)
return i.version
def remove(self, intake_id: str) -> None:
self._forget(T.INTAKE, {"intake_id": intake_id})
T.INTAKE.remove(self.conn, {"intake_id": intake_id})
# ── PayrollExtract ─────────────────────────────────────────────────────────────────────────
LINE_COLS = ("rate_basis", "rate_amount", "currency", "prorate_from", "prorate_to", "quantity_minutes",
"quantity_days", "payable_amount", "change_kind", "org_unit_id", "cost_centre")
class PayrollExtractRepository(_Tracked):
def get(self, batch_id: str) -> Optional[Batch]:
rows = self._load(T.PAYROLL, {"batch_id": batch_id})
return None if rows is None else _batch(rows["hr_payroll_batch"][0], rows["hr_payroll_line"])
def save(self, b: Batch) -> int:
rows = {"hr_payroll_batch": [{"batch_id": b.batch_id, "legal_entity_ref": b.legal_entity_ref,
"period": b.period, "state": b.state, "produced_at": _store_time(b.produced_at),
"sent_at": _store_time(b.sent_at), "sent_by": b.sent_by}],
"hr_payroll_line": [{"batch_id": b.batch_id, "person_id": ln.person_id, "n": ln.n,
"component": ln.component, "for_period": ln.for_period,
"slice_from": ln.slice_from, **{c: getattr(ln, c) for c in LINE_COLS},
"detail_json": _j(ln.detail)} for ln in b.lines]}
b.version = self._save(T.PAYROLL, rows, b.version)
return b.version
def batches(self, legal_entity_ref: str) -> list:
roots = [dict(r._mapping) for r in self.conn.execute(select(T.payroll_batch).where(
T.payroll_batch.c.legal_entity_ref == legal_entity_ref))]
if not roots:
return []
lines = [dict(r._mapping) for r in self.conn.execute(select(T.payroll_line).where(
T.payroll_line.c.batch_id.in_([r["batch_id"] for r in roots])))]
return sorted((_batch(r, [l for l in lines if l["batch_id"] == r["batch_id"]]) for r in roots),
key=lambda b: b.period)
def _batch(h: dict, lines: list) -> Batch:
b = Batch(h["batch_id"], h.get("legal_entity_ref") or "", h.get("period") or "", h.get("state") or "draft",
_aware(h.get("produced_at")), _aware(h.get("sent_at")), h.get("sent_by"), int(h.get("version") or 0))
for l in lines:
b.lines.append(Line(int(l["person_id"]), int(l["n"]), l["component"], l["for_period"], _d(l["slice_from"]),
l.get("rate_basis"),
Decimal(str(l["rate_amount"])).quantize(Decimal("0.0001")) if l.get("rate_amount") is not None else None,
l.get("currency"), _d(l.get("prorate_from")), _d(l.get("prorate_to")),
l.get("quantity_minutes"), l.get("quantity_days"),
Decimal(str(l["payable_amount"])).quantize(Decimal("0.01")) if l.get("payable_amount") is not None else None,
l.get("change_kind") or "", l.get("org_unit_id"), l.get("cost_centre"),
_uj(l.get("detail_json"), {})))
return b
# ── PrivacyRequest ─────────────────────────────────────────────────────────────────────────
class PrivacyRepository(_Tracked):
def get(self, request_id: str) -> Optional[PrivacyRequest]:
rows = self._load(T.PRIVACY, {"request_id": request_id})
if rows is None:
return None
h = rows["hr_privacy_request"][0]
return PrivacyRequest(h["request_id"], int(h["person_id"]), h["kind"], _d(h["received_on"]), _d(h["due_on"]),
_aware(h.get("done_at")), h.get("done_by"), h.get("export_object_key"),
_uj(h.get("outcome"), {}), int(h.get("version") or 0))
def save(self, q: PrivacyRequest) -> int:
q.version = self._save(T.PRIVACY, {"hr_privacy_request": [{
"request_id": q.request_id, "person_id": q.person_id, "kind": q.kind, "received_on": q.received_on,
"due_on": q.due_on, "done_at": _store_time(q.done_at), "done_by": q.done_by,
"export_object_key": q.export_object_key, "outcome": _j(q.outcome)}]}, q.version)
return q.version
def of_person(self, person_id: int) -> list:
return sorted(r[0] for r in self.conn.execute(select(T.privacy_request.c.request_id).where(
T.privacy_request.c.person_id == int(person_id))))
class RetentionLog:
"""Append-only: run_date|person_id|category. A plain INSERT, so the same sweep twice in a day writes once."""
def __init__(self, conn):
self.conn = conn
def add(self, run_date: date, person_id: int, category: str, action: str, basis: str, actor: str) -> None:
self.conn.execute(insert(T.retention_log).values(run_date=run_date, person_id=int(person_id),
category=category, action=action, basis=basis, actor=actor,
logged_at=datetime.now(timezone.utc).replace(tzinfo=None)))
def done(self, person_id: int) -> dict:
out = {}
for r in self.conn.execute(select(T.retention_log).where(T.retention_log.c.person_id == int(person_id))):
d = dict(r._mapping)
out.setdefault(d["category"], []).append(d)
return out
def of_day(self, run_date: date) -> list:
return [dict(r._mapping) for r in self.conn.execute(select(T.retention_log).where(
T.retention_log.c.run_date == run_date))]
def rows(self, person_id: int) -> list:
return [dict(r._mapping) for r in self.conn.execute(select(T.retention_log).where(
T.retention_log.c.person_id == int(person_id)))]
class ObjectStore:
"""Document bytes in the bucket (hr_object_chunk), keyed hr/<person_id>/…; read only by a use case."""
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)
for i, part in enumerate(parts):
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))
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(self, key: str) -> None:
self.conn.execute(delete(T.object_chunk).where(T.object_chunk.c.object_key == key))
def keys(self, prefix: str) -> list:
return 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 + "%")))})
__all__ = ["PersonRepository", "SensitiveRepository", "OrgRepository", "ConfigurationRepository",
"PlaybookRepository", "RunRepository", "IntakeRepository", "PayrollExtractRepository",
"PrivacyRepository", "RetentionLog", "ObjectStore", "StepClaims", "is_duplicate", "IntegrityError"]