"""The org and the customer's configuration: keep_org (units, chief positions, positions — dated), keep_configuration
(every catalogue row), keep_playbook (a new immutable version per change), install_localisation_pack, and the reads
(org_chart, list_playbooks, show_configuration). Every config_row of the spec is kept through these.
"""
from datetime import date, timezone
from typing import Literal, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId, Email, Limit, Name, Note, Quantity
from erp_platform.repository import CONFLICT_REASON, as_conflict
from brick_hr.application import support as S
from brick_hr.application import use_cases as uc
from brick_hr.domain import bpmn
from brick_hr.domain import deviations as dv
from brick_hr.domain import playbook as pbk
from brick_hr.domain.events import direction_of
from brick_hr.domain.configuration import (CATALOGUES, DATED_SETTINGS, Config, check_dated_setting, in_use_refusal,
validate_row)
from brick_hr.domain.values import DAY
from brick_hr.domain.org import Org
from brick_hr.domain.packs import PACKS
from brick_hr.domain.playbook import Playbook, PlaybookStep
from brick_hr.infrastructure import capability_acl as acl
HR = uc.HR
Catalogue = Literal["worker_type", "employee_subgroup", "org_unit_type", "event_type", "event_reason", "infogroup",
"rate_basis", "pay_component_type", "pay_grade", "responsible_group", "document_type",
"intake_field", "draft_instruction", "retention_rule", "settings", "job", "location",
"work_schedule", "holiday", "unit_access", "company_grant"]
@HR.use_case(kind="command", route="POST /org", permission=S.ORG_WRITE, if_match="expected_version")
@S.retry_unless_versioned
def keep_org(caller: dict, org_id: Name, kind: Literal["org", "unit", "position"], values: dict,
id: Optional[Name] = None, valid_from: Optional[S.Day] = None, valid_to: Optional[S.Day] = None,
expected_version: Optional[int] = None) -> dict:
"""Keep the org (HR-12): the org tree itself (kind org: name, legal_entity_ref), an org unit (name, parent,
type, cost centre, chief position, location) or a position (unit, job, fte) — each a dated slice from
valid_from that delimits the one before; until then the old tree is current. The derived manager follows on
that date for everyone in the unit without touching their records. A move that would make a cycle is refused;
so is a second chief for one unit on one date."""
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
o = uow.orgs.get(org_id)
if expected_version is not None and int(expected_version) != (o.version if o else 0):
raise Refused(409, CONFLICT_REASON)
if kind == "org":
o = o or Org(org_id, "", "")
o.name = values.get("name") or o.name
o.legal_entity_ref = values.get("legal_entity_ref") or o.legal_entity_ref
if not o.name:
raise Refused(422, "an org tree needs a name")
extra = sorted(set(values) - {"name", "legal_entity_ref"})
if extra:
raise Refused(422, f"an org has name and legal_entity_ref; not {', '.join(extra)}")
else:
if o is None:
raise Refused(404, f"no org {org_id}")
if not id:
raise Refused(422, f"a {kind} needs its id")
d = valid_from or view.today
if kind == "unit":
o.keep_unit(id, d, values, valid_to)
else:
o.keep_position(id, d, values, valid_to)
uow.orgs.save(o)
return {"org_id": org_id, "kind": kind, "id": id, "version": o.version,
"valid_from": (valid_from or view.today).isoformat()}
@HR.use_case(kind="query", route="GET /org/{org_id}", permission=S.ORG_READ)
def org_chart(caller: dict, org_id: Name, on: Optional[S.Day] = None) -> dict:
"""The org tree on a date (today by default): each unit, its parent, its chief position and who holds it,
and how many people are assigned. As the org was THEN for a past date."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
o = uow.orgs.get(org_id)
if o is None:
raise Refused(404, f"no org {org_id}")
d = on or view.today
seats = view.directory.seats_on(d)
holders = {}
for u in o.units_on(d):
if u.chief_position_id:
holders[u.org_unit_id] = view.directory.holders(u.chief_position_id, d)
names = uow.persons.names([h for hs in holders.values() for h in hs], d)
units = [{"org_unit_id": u.org_unit_id, "name": u.name, "parent_org_unit_id": u.parent_org_unit_id,
"valid_from": u.valid_from.isoformat(), "valid_to": S.due_label(u.valid_to) or "open",
"chief_position_id": u.chief_position_id,
"chief": [{"person_id": h, "name": names.get(h, {}).get("name")} for h in holders.get(u.org_unit_id, [])],
"people": len([s for s in seats if s.org_id == org_id and s.org_unit_id == u.org_unit_id])}
for u in o.units_on(d)]
positions = [{"position_id": p.position_id, "org_unit_id": p.org_unit_id, "job_code": p.job_code,
"holders": view.directory.holders(p.position_id, d)}
for p in sorted((p for p in o.positions if p.covers(d)), key=lambda p: p.position_id)]
return {"org_id": org_id, "name": o.name, "legal_entity_ref": o.legal_entity_ref, "on": d.isoformat(),
"version": o.version, "units": units, "positions": positions}
def _save_rows(uow, catalogue: str, rows: list, remove: list) -> tuple:
cat = CATALOGUES[catalogue]
cfg = uow.config.load()
current = Config({k: list(v) for k, v in cfg.rows.items()})
saved = 0
pending: list = []
for row in rows:
clean = validate_row(cat, row, current)
key = tuple(clean[k] for k in cat.key)
existing = next((r for r in current.rows[catalogue] if tuple(str(r[k]) for k in cat.key) == key), None)
merged = {**(existing or {}), **clean}
if existing is not None and all(_same(existing.get(f), merged.get(f)) for f in cat.key + cat.fields):
continue
pending.append(merged)
current.rows[catalogue] = [r for r in current.rows[catalogue] if tuple(str(r[k]) for k in cat.key) != key]
current.rows[catalogue].append(merged)
saved += 1
uow.config.save_rows(catalogue, pending) # one multi-row write, before any removal below
removed = 0
for key in remove:
kv = {k: str(key.get(k, "")) for k in cat.key}
in_use_refusal(catalogue, tuple(kv.values()), uow.config.uses(catalogue, kv))
if any(tuple(str(r[k]) for k in cat.key) == tuple(kv.values()) for r in current.rows[catalogue]):
uow.config.remove_row(catalogue, kv)
removed += 1
return saved, removed
def _same(a, b) -> bool:
return (a if a not in ("", None) else None) == (b if b not in ("", None) else None) or str(a) == str(b)
@HR.use_case(kind="command", route="POST /configuration/{catalogue}", permission=S.CONFIG_WRITE, idempotent=True)
def keep_configuration(caller: dict, catalogue: Catalogue, rows: Optional[list[dict]] = None,
remove: Optional[list[dict]] = None) -> dict:
"""Keep a catalogue (HR-28): worker types, event types (each mapped to one of the seven kinds) and reasons,
infogroups, jobs, locations, schedules, holidays, unit access (groups, chat spaces, folders), rate bases, pay
components, pay grades, responsible groups, document types, intake fields, draft instructions, retention rules
and settings. New events and hires can use a new row at once. A row in use cannot be removed — retire it
(active: false). Saving the same rows twice changes nothing."""
with S.rules():
with uc.repository() as uow:
if catalogue == "company_grant":
S.check_company_grant_change(S.View(uow, caller), list(rows or []) + list(remove or []))
rows = [{**r, "granted_by": S.actor(caller)} if isinstance(r, dict) else r for r in rows or []]
_save_rows(uow, catalogue, rows or [], remove or [])
cat = CATALOGUES[catalogue]
# The answer is the state asked for — the keys now kept and removed — so a replay answers the same.
return {"catalogue": catalogue,
"kept": [{k: str(r.get(k)) for k in cat.key} for r in rows or []],
"removed": [{k: str(r.get(k)) for k in cat.key} for r in remove or []]}
@HR.use_case(kind="query", route="GET /configuration/{catalogue}", permission=S.CONFIG_WRITE)
def show_configuration(caller: dict, catalogue: Catalogue) -> dict:
"""The rows of one catalogue, as stored (what Settings shows)."""
with S.rules():
with uc.repository() as uow:
return {"catalogue": catalogue, "rows": uow.config.load().all(catalogue)}
# ── dated settings (HR-55): the owner's policy rows, only ever added ─────────────────────────────
def _dated_view(cfg: Config, key: str, today: date) -> dict:
scopes = sorted({r["scope"] for r in cfg.dated if r["setting_key"] == key} - {""})
return {"setting_key": key, "today": today.isoformat(), "default": DATED_SETTINGS[key][0],
"in_force": {"": cfg.dated_setting(key, today), **{s: cfg.dated_setting(key, today, s) for s in scopes}},
"history": [{**r, "valid_from": r["valid_from"].isoformat() if hasattr(r["valid_from"], "isoformat")
else r["valid_from"], "set_at": S.iso(r.get("set_at"))} for r in cfg.dated_history(key)]}
@HR.use_case(kind="query", route="GET /dated-settings/{setting_key}", permission=S.CONFIG_WRITE)
def show_dated_setting(caller: dict, setting_key: str) -> dict:
"""A dated setting (HR-55, offboarding.revoke_access_at: when a leaver's ERP sign-in is revoked): the value in force
today (the business date) for the default and for each leave reason that overrides it, and every row ever set —
who, when, from which day. Rows are only ever added, so this is the whole history."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
if setting_key not in DATED_SETTINGS:
raise Refused(404, f"no dated setting {setting_key}")
return _dated_view(view.config, setting_key, view.today)
@HR.use_case(kind="command", route="POST /dated-settings/{setting_key}", permission=S.OFFBOARDING_POLICY,
idempotent=True)
def set_dated_setting(caller: dict, setting_key: str, value: Name, scope: Optional[Name] = None,
valid_from: Optional[date] = None, note: Optional[Note] = None) -> dict:
"""The owner sets a dated setting (HR-55) as a NEW row: offboarding.revoke_access_at = last_day_end (the default:
the last day at the business close, after the access-removal approval) or immediate (as soon as the leaver is
recorded and the access removal approved). `scope` is a leave reason (dismissal) overriding the default for that
reason; left out, the row is the default. It takes effect from tomorrow (the business date) unless a later day is
given; only the first row of a key and scope may start today. A run already made keeps its timing. Needs
hr:offboarding:policy. Setting the same row again answers the same; a different value for the same day is 409."""
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
cfg, today, scope_ = view.config, view.today, (scope or "").strip()
v = check_dated_setting(setting_key, value, scope_, cfg)
mine = [r for r in cfg.dated if r["setting_key"] == setting_key and r["scope"] == scope_]
start = valid_from or (today if not mine else today + DAY)
same = next((r for r in mine if r["valid_from"] == start), None)
if same is not None:
if str(same["value"]) == v:
return _dated_view(cfg, setting_key, today) # a replay: nothing new
raise Refused(409, f"{setting_key} already has a row for {scope_ or 'the default'} from {start}: rows "
"are never edited, date the new one later", gate="dated_row_exists",
fix=f"set valid_from to {start + DAY} or later")
if start < today or (start == today and mine):
raise Refused(409, f"{setting_key} takes effect after today ({today}, the business date in {view.tz})"
+ ("" if not mine else " (only the first row of a key may start today)"),
gate="dated_from_tomorrow", fix=f"set valid_from to {today + DAY} or leave it out")
row = {"setting_key": setting_key, "scope": scope_, "valid_from": start, "value": v,
"set_by": S.actor(caller), "set_sub": str(caller.get("sub") or ""),
"set_at": view.now.astimezone(timezone.utc).replace(tzinfo=None), "note": note or None}
uow.config.add_dated(row)
return _dated_view(cfg, setting_key, today)
def _playbook_from(body: dict, version: int, by: str, at, previous: Optional[Playbook] = None) -> Playbook:
"""A playbook from its body; a step without a key gets one (playbook.assign_keys: the previous version's key for
the same step, else a slug of its title)."""
steps = []
for s in body.get("steps") or []:
if not isinstance(s, dict):
raise Refused(422, "a step is an object")
extra = sorted(set(s) - {"step_no", "kind", "title", "responsible_group", "timing", "capability",
"input_template", "depends_on", "condition", "approval_group", "draft_instruction",
"template_kind", "fan_out", "key", "outcomes"})
if extra:
raise Refused(422, f"unknown step field(s) {', '.join(extra)}")
try:
steps.append(PlaybookStep(int(s["step_no"]), s.get("kind") or "", s.get("title") or "",
s.get("responsible_group") or "",
dict(s.get("timing") or {"anchor": "run.created", "offset": 0, "unit": "days",
"time": None}),
s.get("capability"), s.get("input_template"),
[int(d) for d in s.get("depends_on") or []], s.get("condition"),
s.get("approval_group"), s.get("draft_instruction"), s.get("template_kind"),
s.get("fan_out"), str(s.get("key") or ""), list(s.get("outcomes") or [])))
except (KeyError, TypeError, ValueError) as e:
raise Refused(422, f"a step needs step_no, kind, title, responsible_group ({e})") from None
pbk.assign_keys(steps, previous)
return Playbook(body["playbook_id"], version, body.get("name") or "", list(body.get("event_types") or []),
list(body.get("worker_types") or []), list(body.get("reasons") or []), body.get("org_unit_id"),
body.get("status") or "active", by, at, steps)
def body_of(pb: Playbook) -> dict:
"""A stored playbook as keep_playbook's body (every step with its key)."""
return {"playbook_id": pb.playbook_id, "name": pb.name, "event_types": list(pb.event_types),
"worker_types": list(pb.worker_types), "reasons": list(pb.reasons), "org_unit_id": pb.org_unit_id,
"status": pb.status, "steps": [dict(s.__dict__) for s in pb.steps]}
def save_playbook(uow, body: dict, by: str, at) -> tuple:
cfg = uow.config.load()
latest_v = uow.playbooks.latest_version(body["playbook_id"])
prev = uow.playbooks.get(body["playbook_id"], latest_v) if latest_v else None
pb = _playbook_from(body, latest_v + 1, by, at, prev)
kinds = {r["event_type"]: r["kind"] for r in cfg.all("event_type")}
for et in pb.event_types:
if et not in kinds:
raise Refused(422, f"there is no event type {et!r}")
for wt in pb.worker_types:
if cfg.worker_type(wt) is None:
raise Refused(422, f"there is no worker type {wt!r}")
pb.validate(known=acl.known, destructive=acl.is_destructive, kind_of_event=lambda et: kinds.get(et),
groups=lambda g: cfg.group(g) is not None)
for s in pb.steps:
if s.kind == "draft" and cfg.draft_instruction(s.draft_instruction) is None:
raise Refused(409, f"step {s.step_no}: there is no draft instruction {s.draft_instruction!r}")
from brick_hr.application import process
broken = dv.foundation(process.deviations_of(cfg, pb))
if broken:
raise Refused(409, "; ".join(d.reason for d in broken) + ". That is part of the kept foundation, so it "
"cannot be changed by configuration", gate="kept_foundation",
fix="playbook_deviations lists what can be changed")
if prev is not None and prev.content() == pb.content() and prev.status == pb.status:
return prev, False
for other in uow.playbooks.latest():
if pbk.ties(pb, other):
raise Refused(409, f"{pb.playbook_id} and {other.playbook_id} would both match the same event at the same "
"specificity — narrow one (worker types, reasons or org unit) so selection has one answer")
uow.playbooks.create(pb)
return pb, True
@HR.use_case(kind="command", route="POST /playbooks", permission=S.CONFIG_WRITE, idempotent=True)
def keep_playbook(caller: dict, playbook_id: Name, name: Name, event_types: list[str], steps: list[dict],
worker_types: Optional[list[str]] = None, reasons: Optional[list[str]] = None,
org_unit_id: Optional[Name] = None, status: Literal["active", "retired"] = "active") -> dict:
"""Keep a checklist as a playbook (HR-11). Any change makes a NEW version that new runs use; open runs keep the
version they were expanded from. Each step: kind (task, action, document, draft, intake, approval), title,
responsible group, timing {anchor: start_date | last_day | event_date | run.created | step:<n>.done, offset,
unit: days | working_days, time: HH:MM}, depends_on, and for an action its capability and input_template.
Refused in words: a destructive action with no approval before it, a delete with no transfer before it, a
dependency cycle, a tie with another playbook. Saving the same content twice is one version."""
at = S.now()
body = {"playbook_id": playbook_id, "name": name, "event_types": event_types, "steps": steps,
"worker_types": worker_types or [], "reasons": reasons or [], "org_unit_id": org_unit_id,
"status": status}
with S.rules():
try:
with uc.repository() as uow:
pb, created = save_playbook(uow, body, S.actor(caller), at)
except Exception as exc: # noqa: BLE001
from brick_hr.infrastructure.repositories import is_duplicate
if is_duplicate(exc) and as_conflict(exc) is None:
raise Refused(409, "someone saved a new version of this playbook first — reload it") from None
raise
return {"playbook_id": pb.playbook_id, "version": pb.version}
@HR.use_case(kind="query", route="GET /playbooks", permission=S.CONFIG_WRITE)
def list_playbooks(caller: dict, playbook_id: Optional[Name] = None) -> dict:
"""The latest version of every playbook (or of one), with its steps."""
with S.rules():
with uc.repository() as uow:
out = []
for pb in uow.playbooks.latest():
if playbook_id and pb.playbook_id != playbook_id:
continue
out.append({"playbook_id": pb.playbook_id, "version": pb.version, "name": pb.name,
"event_types": pb.event_types, "worker_types": pb.worker_types, "reasons": pb.reasons,
"org_unit_id": pb.org_unit_id, "status": pb.status,
"steps": [s.__dict__ for s in pb.steps]})
return {"playbooks": out}
def bpmn_setting(cfg: Config, pb: Playbook, signer_name: str = "") -> bpmn.Setting:
"""What the BPMN picture needs from the configuration around one playbook (labels, intake fields, the zone,
the direction its events start, how waiting steps are chased, who signs a contract) and from the capability
contract."""
from brick_hr.application import process
signer = cfg.setting("contract_signer_person_id")
kinds = [(cfg.event_type(et) or {}).get("kind") for et in pb.event_types]
directions = [d for d in (direction_of(k, (pb.reasons or [None])[0]) for k in kinds if k) if d]
return bpmn.Setting(
direction=directions[0] if directions else "change",
timezone=cfg.setting("timezone") or "UTC",
groups={r["group_code"]: r.get("label") or r["group_code"] for r in cfg.all("responsible_group")},
drafts={r["draft_code"]: (r.get("label") or r["draft_code"], tuple(r.get("fields") or ()))
for r in cfg.all("draft_instruction")},
documents={r["document_type"]: r.get("label") or r["document_type"] for r in cfg.all("document_type")},
intake_fields=tuple((r["field"], r.get("label") or r["field"], tuple(r.get("worker_types") or ()))
for r in sorted(cfg.all("intake_field", active_only=True),
key=lambda r: (r.get("sort_no") or 0, r["field"]))),
capabilities=acl.described(s.capability for s in pb.steps),
exporter_version=uc.VERSION,
chase=process.chase_words(cfg),
signer=(f"{signer_name or 'person ' + signer} (the contract signer setting), unless the run names its own "
"director" if signer else "the director named on the run (assign_run_party director), if any"))
@HR.use_case(kind="query", route="GET /playbooks/{playbook_id}/bpmn", permission=S.CONFIG_WRITE, etag="version")
def show_playbook_bpmn(caller: dict, playbook_id: Name, version: Optional[Quantity] = None) -> dict:
"""A playbook (its latest version, or the version given) as a BPMN 2.0 process model — valid XML that Camunda
Modeler, Signavio or bpmn.io open as is: a lane per responsible group, a user task per manual step, a service
task per capability call (a manual task with the same input when no provider is wired), the dependencies as
sequence flows, conditions as exclusive gateways, timing rules as timers, the outside systems as message
flows, and the intake fields, drafts and contracts as data objects with what reads and writes them. Read-only.
Over HTTP the answer is the XML itself (application/xml; Accept: application/json gives this envelope)."""
with S.rules():
with uc.repository() as uow:
v = version or uow.playbooks.latest_version(playbook_id)
pb = uow.playbooks.get(playbook_id, v) if v else None
if pb is None:
raise Refused(404, f"no playbook {playbook_id}" + (f" version {version}" if version else ""))
from brick_hr.application import process
cfg = uow.config.load()
signer = cfg.setting("contract_signer_person_id")
name = (uow.persons.names([int(signer)], S.now().date()).get(int(signer)) or {}).get("name", "") \
if signer else ""
xml = bpmn.to_bpmn(pb, bpmn_setting(cfg, pb, name), tuple(process.deviations_of(cfg, pb)))
return {"playbook_id": pb.playbook_id, "version": pb.version, "media_type": "application/xml",
"filename": f"{pb.playbook_id}.v{pb.version}.bpmn", "bpmn": xml}
@HR.use_case(kind="command", route="POST /localisation-packs/{pack}", permission=S.CONFIG_WRITE, idempotent=True)
def install_localisation_pack(caller: dict, pack: Name, with_playbooks: bool = False) -> dict:
"""Install a country's localisation pack (gb: UK intake fields, document types, compliance records, retention
defaults, bank holidays from the published list) — rows the customer then edits. Rows already there are left
as they are (the customer's values win). with_playbooks adds the pack's starter playbooks where none exists.
Installing twice changes nothing."""
mod = PACKS.get(pack)
if mod is None:
raise Refused(404, f"no localisation pack {pack!r} (packs: {', '.join(PACKS)})")
at = S.now()
added: dict = {}
order = ["worker_type", "event_type", "event_reason", "infogroup", "rate_basis", "pay_component_type",
"responsible_group", "document_type", "intake_field", "draft_instruction", "retention_rule", "settings",
"holiday"]
with S.rules():
cats = mod.catalogues()
with uc.repository() as uow:
cfg = uow.config.load()
for name in order:
cat = CATALOGUES[name]
have = {tuple(str(r[k]) for k in cat.key) for r in cfg.all(name)}
new = [r for r in cats.get(name, []) if tuple(str(r[k]) for k in cat.key) not in have]
if new:
saved, _ = _save_rows(uow, name, new, [])
added[name] = saved
cfg = uow.config.load()
playbooks = []
if with_playbooks:
for body in mod.starter_playbooks():
with uc.repository() as uow:
# any version stored? (from the configuration snapshot, current to this request's versions)
if body["playbook_id"] in {pb.playbook_id for pb in uow.playbooks.latest()}:
continue
pb, created = save_playbook(uow, body, S.actor(caller), at)
if created:
playbooks.append(pb.playbook_id)
with uc.repository() as uow:
cfg = uow.config.load()
present = {}
for name in order:
cat = CATALOGUES[name]
have = {tuple(str(r[k]) for k in cat.key) for r in cfg.all(name)}
present[name] = sum(1 for r in cats.get(name, []) if tuple(str(r[k]) for k in cat.key) in have)
stored = {pb.playbook_id for pb in uow.playbooks.latest()} # a playbook with any version
pbs = sorted(b["playbook_id"] for b in mod.starter_playbooks() if b["playbook_id"] in stored)
# The state, not what this call added: a second install answers the same.
return {"pack": pack, "rows": present, "playbooks": pbs if with_playbooks else []}
__all__ = ["keep_org", "org_chart", "keep_configuration", "show_configuration", "keep_playbook", "list_playbooks",
"show_playbook_bpmn", "bpmn_setting", "body_of",
"install_localisation_pack", "save_playbook"]