dodil/erp-brick-hr
src/brick_hr/infrastructure/tables.py
"""The tables, the aggregates over them, and the unit of work (HrRepository) every use case opens.
Every table is prefixed hr_ (BRICKS.md: bricks share a bucket at tier 1). Natural keys only, never SERIAL. Every
child table's key starts with its aggregate root's key — that is how a save finds stale rows at any depth.
schema.sql is generated from this file (scripts/gen_schema.py) and tests/test_schema.py fails when they drift.
Aggregates (erp_platform.repository): Person, Sensitive, Org, Run, Intake and PayrollExtract carry an optimistic
`version` (two people saving one record: one gets 409 "changed since you opened it"). A Playbook version is
immutable and claimed by a plain INSERT of its root row. Configuration rows are one small aggregate each.
Append-only rows (the retention log, step claims, document chunks) are written with plain INSERTs.
"""
from typing import Optional
from sqlalchemy import (BigInteger, Boolean, Column, Date, Integer, MetaData, Numeric, String, Table, TIMESTAMP)
from erp_platform.repository import Aggregate, Part
md = MetaData()
ID = BigInteger
RATE = Numeric(19, 4)
AMOUNT = Numeric(18, 2)
PCT = Numeric(5, 2)
def _pk(name, type_=String):
return Column(name, type_, primary_key=True, autoincrement=False, nullable=False)
def _slice_cols():
return [Column("valid_to", Date, nullable=False), Column("superseded_by", Integer),
Column("recorded_at", TIMESTAMP), Column("superseded_at", TIMESTAMP), Column("recorded_by", String),
Column("reason", String)]
# ── Person ─────────────────────────────────────────────────────────────────────────────────
person = Table("hr_person", md, _pk("person_id", ID), Column("person_key", String, nullable=False),
Column("staff_no", String), Column("bp_id", ID), Column("pool_user_sub", String),
Column("created_at", TIMESTAMP), Column("updated_at", TIMESTAMP), Column("version", Integer))
person_detail = Table("hr_person_detail", md, _pk("person_id", ID), _pk("valid_from", Date), _pk("rev", Integer),
Column("legal_given_name", String), Column("legal_family_name", String),
Column("preferred_name", String), Column("personal_email", String),
Column("personal_phone", String), Column("work_email", String), Column("work_phone", String),
*_slice_cols())
engagement = Table("hr_engagement", md, _pk("person_id", ID), _pk("n", Integer),
Column("worker_type", String, nullable=False), Column("legal_entity_ref", String),
Column("start_date", Date, nullable=False), Column("planned_end", Date), Column("actual_end", Date),
Column("leave_reason", String), Column("supplier_bp_id", ID), Column("created_at", TIMESTAMP))
assignment = Table("hr_assignment", md, _pk("person_id", ID), _pk("n", Integer), _pk("valid_from", Date),
_pk("rev", Integer), Column("org_id", String), Column("org_unit_id", String),
Column("position_id", String), Column("job_code", String), Column("location_code", String),
Column("schedule_code", String), Column("employment_pct", PCT), Column("employee_subgroup", String),
Column("cost_centre", String), Column("manager_override_person_id", ID), *_slice_cols())
pay = Table("hr_pay", md, _pk("person_id", ID), _pk("n", Integer), _pk("valid_from", Date), _pk("rev", Integer),
Column("rate_basis", String), Column("rate_amount", RATE), Column("currency", String),
Column("frequency", String), Column("pay_grade", String), *_slice_cols())
pay_component = Table("hr_pay_component", md, _pk("person_id", ID), _pk("n", Integer), _pk("component"),
_pk("valid_from", Date), _pk("rev", Integer), Column("rate_basis", String),
Column("rate_amount", RATE), Column("currency", String), *_slice_cols())
event = Table("hr_event", md, _pk("person_id", ID), _pk("n", Integer), _pk("seq", Integer),
Column("event_type", String, nullable=False), Column("kind", String, nullable=False),
Column("reason", String), Column("effective_date", Date, nullable=False), Column("actor", String),
Column("recorded_at", TIMESTAMP), Column("slices_written", String), Column("run_id", String),
Column("bp_synced_at", TIMESTAMP), Column("payload_json", String))
document = Table("hr_document", md, _pk("person_id", ID), _pk("doc_id"), Column("document_type", String),
Column("access_class", String), Column("filename", String), Column("content_type", String),
Column("object_key", String), Column("sha256", String), Column("size_bytes", BigInteger),
Column("expires_on", Date), Column("contract_id", String), Column("uploaded_by", String),
Column("uploaded_at", TIMESTAMP))
compliance = Table("hr_compliance_record", md, _pk("person_id", ID), _pk("n", Integer), _pk("kind"),
_pk("valid_from", Date), _pk("rev", Integer), Column("pack", String), Column("outcome", String),
Column("checked_on", Date), Column("checked_by", String), Column("follow_up_on", Date),
Column("evidence_ref", String), Column("reasons", String), Column("detail_json", String),
*_slice_cols())
_SLICE_INSERT_ONLY = ("created_at", "created_by", "recorded_at", "recorded_by")
PERSON = Aggregate(Part(person, ("person_id",)),
Part(person_detail, ("person_id", "valid_from", "rev"), _SLICE_INSERT_ONLY),
Part(engagement, ("person_id", "n")),
Part(assignment, ("person_id", "n", "valid_from", "rev"), _SLICE_INSERT_ONLY),
Part(pay, ("person_id", "n", "valid_from", "rev"), _SLICE_INSERT_ONLY),
Part(pay_component, ("person_id", "n", "component", "valid_from", "rev"), _SLICE_INSERT_ONLY),
Part(event, ("person_id", "n", "seq"), ("recorded_at",)),
Part(document, ("person_id", "doc_id")),
Part(compliance, ("person_id", "n", "kind", "valid_from", "rev"), _SLICE_INSERT_ONLY),
version="version")
# ── Sensitive (its OWN root: the Person repository never joins it) ─────────────────────────
sensitive = Table("hr_sensitive", md, _pk("person_id", ID), Column("erased_at", TIMESTAMP),
Column("anonymised_at", TIMESTAMP), Column("version", Integer))
person_sensitive = Table("hr_person_sensitive", md, _pk("person_id", ID), _pk("kind"), _pk("valid_from", Date),
_pk("rev", Integer), Column("value_json", String), Column("object_key", String),
Column("sha256", String), Column("created_by", String), *_slice_cols())
SENSITIVE = Aggregate(Part(sensitive, ("person_id",)),
Part(person_sensitive, ("person_id", "kind", "valid_from", "rev"), _SLICE_INSERT_ONLY),
version="version")
# ── Org ────────────────────────────────────────────────────────────────────────────────────
org = Table("hr_org", md, _pk("org_id"), Column("name", String), Column("legal_entity_ref", String),
Column("version", Integer))
org_unit = Table("hr_org_unit", md, _pk("org_id"), _pk("org_unit_id"), _pk("valid_from", Date),
Column("valid_to", Date, nullable=False), Column("name", String),
Column("parent_org_unit_id", String), Column("org_unit_type", String), Column("cost_centre", String),
Column("chief_position_id", String), Column("location_code", String))
position = Table("hr_position", md, _pk("org_id"), _pk("position_id"), _pk("valid_from", Date),
Column("valid_to", Date, nullable=False), Column("org_unit_id", String), Column("job_code", String),
Column("fte_pct", String), Column("cost_centre", String), Column("vacant_policy", String))
ORG = Aggregate(Part(org, ("org_id",)), Part(org_unit, ("org_id", "org_unit_id", "valid_from")),
Part(position, ("org_id", "position_id", "valid_from")), version="version")
# ── Configuration (one small aggregate per row, rooted at its natural key) ────────────────
_cfg = {}
def _config_table(name, key, cols):
t = Table(name, md, *[_pk(k) for k in key], *cols)
_cfg[name] = (t, key)
return t
A = lambda: Column("active", Boolean) # noqa: E731
L = lambda: Column("label", String) # noqa: E731
worker_type = _config_table("hr_worker_type", ("worker_type",), [
L(), Column("category", String), Column("payroll", Boolean), Column("via_company", Boolean),
Column("bp_roles", String), Column("employee_subgroup", String), A()])
employee_subgroup = _config_table("hr_employee_subgroup", ("subgroup",), [L(), A()])
org_unit_type = _config_table("hr_org_unit_type", ("org_unit_type",), [L(), A()])
event_type = _config_table("hr_event_type", ("event_type",), [L(), Column("kind", String), A()])
event_reason = _config_table("hr_event_reason", ("event_type", "reason"), [L(), Column("immediate", Boolean), A()])
infogroup = _config_table("hr_infogroup", ("event_type", "record"), [Column("required", Boolean)])
rate_basis = _config_table("hr_rate_basis", ("rate_basis",), [L(), A()])
pay_component_type = _config_table("hr_pay_component_type", ("component",), [
L(), Column("worker_types", String), Column("rate_basis", String), A()])
pay_grade = _config_table("hr_pay_grade", ("pay_grade",), [
L(), Column("job_code", String), Column("rate_basis", String), Column("currency", String),
Column("min_amount", String), Column("max_amount", String)])
responsible_group = _config_table("hr_responsible_group", ("group_code",), [L(), Column("resolution", String), A()])
document_type = _config_table("hr_document_type", ("document_type",), [
L(), Column("access_class", String), Column("expires", Boolean), Column("retention_category", String),
Column("sensitive_kind", String), A()])
intake_field = _config_table("hr_intake_field", ("pack", "field"), [
L(), Column("worker_types", String), Column("required", Boolean), Column("sensitive_kind", String),
Column("input", String), Column("sort_no", Integer), A()])
draft_instruction = _config_table("hr_draft_instruction", ("draft_code",), [
L(), Column("instructions", String), Column("fields", String)])
retention_rule = _config_table("hr_retention_rule", ("category",), [
L(), Column("keep_months", Integer), Column("action", String), Column("basis", String)])
settings = _config_table("hr_settings", ("key",), [Column("value", String)])
job = _config_table("hr_job", ("job_code",), [
Column("title", String), Column("family", String), Column("default_subgroup", String),
Column("default_pay_grade", String), A()])
location = _config_table("hr_location", ("location_code",), [
Column("name", String), Column("address", String), Column("country", String), Column("timezone", String),
Column("holiday_calendar", String)])
work_schedule = _config_table("hr_work_schedule", ("schedule_code",), [
L(), Column("weekly_minutes", Integer), Column("daily_pattern", String),
Column("full_time_weekly_minutes", Integer), Column("working_weekdays", String)])
holiday = _config_table("hr_holiday", ("holiday_calendar", "holiday_date"), [Column("name", String)])
unit_access = _config_table("hr_unit_access", ("org_unit_id", "kind", "target"), [Column("role", String), L()])
# 0.6.0 (hr v4): the companies an hr grant covers — one row per principal (a person's sign-in sub or email, or a
# service client's id) and company. A principal with no row covers every company (the spec's "none named").
# dodil-appid's role rows cannot carry company_codes (a role maps to permission strings only), so it lives here.
company_grant = _config_table("hr_company_grant", ("principal", "company_code"), [Column("note", String),
Column("granted_by", String)])
CONFIG_TABLES = _cfg
# 0.6.0 (HR-55): dated settings — rows only ever added (a plain INSERT; a second row for the same key, scope and day
# is refused), so the table is the audit trail: who set what, when, from which day. scope '' is the default; a leave
# reason overrides it for that reason. The row in force on the business date is the value.
setting_row = Table("hr_setting_row", md, _pk("setting_key"), _pk("scope"), _pk("valid_from", Date),
Column("value", String, nullable=False), Column("set_by", String), Column("set_sub", String),
Column("set_at", TIMESTAMP), Column("note", String))
# The configuration's version: ONE row whose token is rewritten (a fresh random value) in the same transaction as
# every change to a catalogue, a playbook or an org tree — what lets a process keep the configuration in memory
# and still see the next change (repositories.configuration_snapshot says why a token and not a counter).
config_version = Table("hr_config_version", md, _pk("key"), Column("token", String, nullable=False),
Column("changed_at", TIMESTAMP))
# ── Playbook (immutable versions) ──────────────────────────────────────────────────────────
playbook = Table("hr_playbook", md, _pk("playbook_id"), _pk("version", Integer), Column("name", String),
Column("event_types", String), Column("worker_types", String), Column("reasons", String),
Column("org_unit_id", String), Column("status", String), Column("created_by", String),
Column("created_at", TIMESTAMP))
playbook_step = Table("hr_playbook_step", md, _pk("playbook_id"), _pk("version", Integer), _pk("step_no", Integer),
Column("kind", String), Column("title", String), Column("responsible_group", String),
Column("capability", String), Column("input_template", String), Column("timing_rule", String),
Column("depends_on", String), Column("condition", String), Column("approval_group", String),
Column("draft_instruction", String), Column("template_kind", String), Column("fan_out", String),
# 0.4.0: the step's stable key (domain/playbook.py) — at the END, and an ALTER (ADDED below)
Column("step_key", String),
# 0.6.0: the outcomes completing the step must record (accepted / declined) — ADDED below
Column("outcomes", String))
PLAYBOOK = Aggregate(Part(playbook, ("playbook_id", "version")),
Part(playbook_step, ("playbook_id", "version", "step_no")))
# ── Run ────────────────────────────────────────────────────────────────────────────────────
run = Table("hr_run", md, _pk("run_id"), Column("direction", String), Column("person_id", ID), Column("n", Integer),
Column("event_seq", String), Column("playbook_id", String), Column("playbook_version", Integer),
Column("created_at", TIMESTAMP), Column("anchors_json", String), Column("cancelled_at", TIMESTAMP),
Column("cancel_reason", String), Column("cancelled_by", String), Column("version", Integer))
run_step = Table("hr_run_step", md, _pk("run_id"), _pk("step_no"), Column("sort_no", Integer), Column("kind", String),
Column("title", String), Column("responsible_group", String), Column("playbook_step_no", Integer),
Column("capability", String), Column("template_json", String), Column("input_json", String),
Column("plan", String), Column("manual", Boolean), Column("manual_reason", String),
Column("timing_rule", String), Column("due_at", TIMESTAMP), Column("state", String),
Column("attempts", Integer), Column("last_status", String), Column("provider_ref", String),
Column("result_json", String), Column("contract_id", String), Column("draft_text", String),
Column("done_by", String), Column("done_at", TIMESTAMP), Column("depends_on", String),
Column("condition", String), Column("approval_group", String), Column("destructive", Boolean),
Column("message", String), Column("next_check_at", TIMESTAMP), Column("approved", Boolean),
Column("target", String), Column("template_kind", String), Column("draft_instruction", String),
Column("outcomes", String)) # 0.6.0, ADDED below
run_party = Table("hr_run_party", md, _pk("run_id"), _pk("group_code"), Column("person_id", ID),
Column("set_by", String))
run_log = Table("hr_run_log", md, _pk("run_id"), _pk("seq", Integer), Column("what", String),
Column("from_value", String), Column("to_value", String), Column("actor", String),
Column("event_at", TIMESTAMP))
RUN = Aggregate(Part(run, ("run_id",)), Part(run_step, ("run_id", "step_no")), Part(run_party, ("run_id", "group_code")),
Part(run_log, ("run_id", "seq"), ("event_at",)), version="version")
# A tick claims a step's attempt with a plain INSERT before it acts: a concurrent second claim fails 23505 and that
# tick skips the step (SC-HR-21). Append-only.
step_claim = Table("hr_step_claim", md, _pk("run_id"), _pk("step_no"), _pk("attempt", Integer),
Column("claimed_at", TIMESTAMP, nullable=False))
# ── Intake ─────────────────────────────────────────────────────────────────────────────────
intake = Table("hr_intake", md, _pk("intake_id"), Column("person_id", ID), Column("n", Integer),
Column("worker_type", String), Column("pack", String), Column("link_ref", String),
Column("issued_at", TIMESTAMP), Column("expires_at", TIMESTAMP), Column("submitted_at", TIMESTAMP),
Column("missing_fields", String), Column("chases", Integer), Column("version", Integer))
INTAKE = Aggregate(Part(intake, ("intake_id",)), version="version")
# ── PayrollExtract ─────────────────────────────────────────────────────────────────────────
payroll_batch = Table("hr_payroll_batch", md, _pk("batch_id"), Column("legal_entity_ref", String),
Column("period", String), Column("state", String), Column("produced_at", TIMESTAMP),
Column("sent_at", TIMESTAMP), Column("sent_by", String), Column("version", Integer))
payroll_line = Table("hr_payroll_line", md, _pk("batch_id"), _pk("person_id", ID), _pk("n", Integer),
_pk("component"), _pk("for_period"), _pk("slice_from", Date), Column("rate_basis", String),
Column("rate_amount", RATE), Column("currency", String), Column("prorate_from", Date),
Column("prorate_to", Date), Column("quantity_minutes", Integer),
Column("quantity_days", PCT), Column("payable_amount", AMOUNT), Column("change_kind", String),
Column("org_unit_id", String), Column("cost_centre", String), Column("detail_json", String))
PAYROLL = Aggregate(Part(payroll_batch, ("batch_id",)),
Part(payroll_line, ("batch_id", "person_id", "n", "component", "for_period", "slice_from")),
version="version")
# ── PrivacyRequest, the retention log, document storage ────────────────────────────────────
privacy_request = Table("hr_privacy_request", md, _pk("request_id"), Column("person_id", ID), Column("kind", String),
Column("received_on", Date), Column("due_on", Date), Column("done_at", TIMESTAMP),
Column("done_by", String), Column("export_object_key", String), Column("outcome", String),
Column("version", Integer))
PRIVACY = Aggregate(Part(privacy_request, ("request_id",)), version="version")
retention_log = Table("hr_retention_log", md, _pk("run_date", Date), _pk("person_id", ID), _pk("category"),
Column("action", String), Column("basis", String), Column("actor", String),
Column("logged_at", TIMESTAMP))
# The bytes of a document, in the bucket, in parts — read only through a use case that checked the document's
# class permission, and streamed; never a link (see infrastructure/documents.py for why not a DataK3 object yet).
object_chunk = Table("hr_object_chunk", md, _pk("object_key"), _pk("part_no", Integer), Column("content_b64", String),
Column("sha256", String), Column("size_bytes", BigInteger), Column("content_type", String),
Column("stored_at", TIMESTAMP))
# Columns added after a table first shipped: schema.sql repeats each as an explicit ALTER below the CREATEs (a bucket
# that predates it gets it; erp_platform.migrate applies and verifies it). (table, column, why) — append only.
ADDED = [
("hr_playbook_step", "step_key", "0.4.0: the stable step key BPMN ids, scenario paths and deviations join on; a "
"row from before it is keyed on read (repositories._playbook) until "
"scripts/backfill_step_keys.py writes the key"),
("hr_playbook_step", "outcomes", "0.6.0: the outcomes a step's completion records (the outplacement offer: "
"accepted / declined); empty = none"),
("hr_run_step", "outcomes", "0.6.0: the playbook step's outcomes, frozen on the run step"),
]
AGGREGATES = {"person": PERSON, "sensitive": SENSITIVE, "org": ORG, "playbook": PLAYBOOK, "run": RUN,
"intake": INTAKE, "payroll": PAYROLL, "privacy": PRIVACY}
class HrRepository:
"""The unit of work a use case opens: ONE connection (one transaction), with a repository per aggregate.
The use case loads and saves ONE aggregate per unit of work (BRICKS.md); cross-aggregate work goes through
the run engine in a later transaction. (This is the class the scenario harness wires to its storage.)"""
def __init__(self, conn):
from brick_hr.infrastructure import repositories as r
self.conn = conn
self.persons = r.PersonRepository(conn)
self.sensitive = r.SensitiveRepository(conn)
self.orgs = r.OrgRepository(conn)
self.config = r.ConfigurationRepository(conn)
self.playbooks = r.PlaybookRepository(conn)
self.runs = r.RunRepository(conn)
self.intakes = r.IntakeRepository(conn)
self.payroll = r.PayrollExtractRepository(conn)
self.privacy = r.PrivacyRepository(conn)
self.objects = r.ObjectStore(conn)
self.retention = r.RetentionLog(conn)
self.claims = r.StepClaims(conn)
def prefetch(self, *, person: Optional[int] = None, sensitive: Optional[int] = None,
intakes_of: Optional[int] = None, run_ids_of: Optional[int] = None, seats: bool = False,
open_runs: bool = False, run_id: Optional[str] = None) -> None:
"""Read ahead, in ONE round trip, what this unit of work is about to ask for — each repository then answers
its next ask for it from what was read (once; a second ask reads again). The same SELECTs the asks would
send, in the same transaction; on DataK3 every statement sees what is committed when it runs, read ahead
or not. The configuration's versions come along when this request has not read them yet."""
from sqlalchemy import select
from brick_hr.infrastructure import batching as B
from brick_hr.infrastructure import repositories as r
plan: list = [] # (where the results go, statements)
if r.needs_tokens(self.conn):
plan.append((lambda res: r.remember_tokens(self.conn, res[0]),
[select(config_version.c.key, config_version.c.token)]))
if person is not None:
plan.append((lambda res, k=(int(person),): self.persons._pre.__setitem__(k, res),
B.load_statements(PERSON, {"person_id": int(person)})))
if sensitive is not None:
plan.append((lambda res, k=(int(sensitive),): self.sensitive._pre.__setitem__(k, res),
B.load_statements(SENSITIVE, {"person_id": int(sensitive)})))
if intakes_of is not None:
plan.append((lambda res, k=("of", int(intakes_of)): self.intakes._pre.__setitem__(k, res[0]),
[select(intake).where(intake.c.person_id == int(intakes_of))]))
if run_ids_of is not None:
plan.append((lambda res, k=("ids", int(run_ids_of)): self.runs._pre.__setitem__(k, res[0]),
[select(run.c.run_id).where(run.c.person_id == int(run_ids_of))]))
if seats:
plan.append((lambda res: self.persons._pre.__setitem__("seats", res), r.PersonRepository.seat_statements()))
if run_id is not None:
plan.append((lambda res, k=(run_id,): self.runs._pre.__setitem__(k, res),
B.load_statements(RUN, {"run_id": run_id})))
if open_runs:
plan.append((lambda res: self.runs._pre.__setitem__("open", res), r.RunRepository.open_run_statements()))
if not plan:
return
results = B.select_many(self.conn, [s for _f, stmts in plan for s in stmts])
i = 0
for put, stmts in plan:
put(results[i:i + len(stmts)])
i += len(stmts)
__all__ = ["md", "HrRepository", "AGGREGATES", "CONFIG_TABLES"]