"""The Configuration aggregates: the customer's catalogues, one small aggregate per row, rooted at its natural key.
These are the POLICIES the customer edits (event types → kinds, reasons, infogroups, worker types, documents,
retention, settings …). Domain code evaluates them; the rules about what a row may say are here:
* an event reason belongs to exactly one event type (its key is event_type|reason);
* an infogroup names only records the event's kind can write;
* a pay component is valid for the worker types it names, and they must exist;
* a worker type must say whether payroll applies and which business-partner roles it gives;
* a draft instruction may name only non-sensitive fields (DRAFTABLE) — a draft is never given sensitive data;
* a reason (or worker type, job) in use cannot be removed, only retired.
"""
import json
from dataclasses import dataclass
from datetime import date
from typing import Any, Iterable, Optional
from erp_platform import work
from erp_platform.money import BASES, ROUNDING
from brick_hr.domain.errors import Incomplete, RuleBroken
KINDS = ("Hired", "Transferred", "PayChanged", "ContractExtended", "Left", "Rehired", "Converted")
# What each kind of personnel event can write (infogroups may name only these).
KIND_RECORDS = {
"Hired": {"detail", "engagement", "assignment", "pay", "pay_component"},
"Rehired": {"detail", "engagement", "assignment", "pay", "pay_component"},
"Converted": {"engagement", "assignment", "pay", "pay_component"},
"Transferred": {"assignment"},
"PayChanged": {"pay", "pay_component"},
"ContractExtended": {"engagement"},
"Left": {"engagement"},
}
CATEGORIES = ("employee", "freelancer", "contractor", "intern")
BP_ROLES = ("employee", "worker", "supplier")
ACCESS_CLASSES = ("personnel", "pay", "sensitive")
SENSITIVE_KINDS = ("ni_number", "tax_id", "bank", "home_address", "date_of_birth", "emergency_contact",
"right_to_work_evidence", "vat_number", "company_number")
# The fields a draft may be given (HR-33). Nothing from the Sensitive aggregate is, or can be added here by a row.
DRAFTABLE = ("preferred_name", "legal_given_name", "job_title", "org_unit_name", "location_name", "start_date",
"last_day", "buddy_name", "manager_name", "first_project", "open_steps", "line", "units_led",
"unit_access", "worker_type_label", "reason_label")
RETENTION_ACTIONS = ("purge", "anonymise")
RETENTION_BASES = ("legal_obligation", "contract", "legitimate_interest", "consent")
ACCESS_KINDS = ("group", "chat_space", "drive_folder")
RESOLUTIONS = ("manager", "run_party", "person") # plus "pool_role:<role>"
SETTINGS_DEFAULTS = {
"timezone": "Europe/London",
"intake_validity_days": "7",
"intake_chase_every_days": "2", "intake_chase_max": "3",
"task_chase_every_days": "1", "task_chase_max": "3",
"approval_chase_every_days": "1", "approval_chase_max": "3",
"chase_unit": "working_days",
"max_action_attempts": "3", "action_backoff_minutes": "5",
"warning_days_contract_end": "14", "warning_days_right_to_work": "28",
"ai_model": "kimi-k2.6", "rounding_mode": "HALF_UP",
"max_backdate_days": "30",
"default_org_id": "", "default_legal_entity": "",
"email_domain": "", "pack": "gb", "user_pool": "",
"hr_mailbox": "", # the mailbox reminders and intake links are sent from
"contract_signer_person_id": "", # who signs contracts for the company when a run names no director
"retention_sweep_hour": "2",
}
SETTING_PREFIXES = ("client_size:",) # client_size:<legal_entity_ref> = small | medium | large
CLIENT_SIZES = ("small", "medium", "large")
# Dated settings (0.6.0, HR-55): policy the owner (hr:offboarding:policy) sets as rows only ever added, each from a day
# (from tomorrow; the first row of a key may start today), the default (scope '') overridable per leave reason.
# key -> (default, allowed values).
REVOKE_ACCESS_AT = "offboarding.revoke_access_at"
LAST_DAY_END, IMMEDIATE = "last_day_end", "immediate"
DATED_SETTINGS = {REVOKE_ACCESS_AT: (LAST_DAY_END, (LAST_DAY_END, IMMEDIATE))}
REVOKES_ACCESS = "identity.revoke_app_user" # the step whose timing offboarding.revoke_access_at decides
def revoke_access_timing(value: str) -> Optional[dict]:
"""The timing of the ERP sign-in revocation for a policy value: None keeps the playbook step's own (last_day_end:
the last day at the business close, 17:30 in the business timezone, in the gb pack); immediate is due as soon as
the run exists — the leaver recorded — and still waits for what the step depends on (the access-removal
approval)."""
if value == IMMEDIATE:
return {"anchor": "run.created", "offset": 0, "unit": "days", "time": None}
return None
def check_dated_setting(key: str, value: Any, scope: str, config: "Config") -> str:
"""The value as stored, or Incomplete in words. `scope` is '' (the default) or a leave reason that exists."""
if key not in DATED_SETTINGS:
raise Incomplete(f"no dated setting {key!r} (there is {', '.join(sorted(DATED_SETTINGS))})")
_default, allowed = DATED_SETTINGS[key]
v = "" if value is None else str(value).strip()
if v not in allowed:
raise Incomplete(f"{key} is {' or '.join(allowed)}, not {value!r}")
if scope and scope not in config.leave_reasons():
raise Incomplete(f"{key} is overridden per leave reason; {scope!r} is not one "
f"(they are {', '.join(config.leave_reasons()) or 'none yet'})")
return v
def _bool(v: Any, what: str) -> bool:
if isinstance(v, bool):
return v
if isinstance(v, str) and v.strip().lower() in ("true", "false"):
return v.strip().lower() == "true"
raise Incomplete(f"{what} is true or false, not {v!r}")
def _list(v: Any, what: str) -> list:
if v is None or v == "":
return []
if isinstance(v, str):
try:
v = json.loads(v)
except ValueError:
v = [x.strip() for x in v.split(",") if x.strip()]
if not isinstance(v, list):
raise Incomplete(f"{what} is a list")
return [x for x in v]
def _int(v: Any, what: str, lo: int = 0) -> int:
if isinstance(v, bool):
raise Incomplete(f"{what} is a whole number")
try:
n = int(str(v))
except ValueError:
raise Incomplete(f"{what} is a whole number, not {v!r}") from None
if n < lo:
raise Incomplete(f"{what} is at least {lo}")
return n
def _req(row: dict, name: str, what: str) -> Any:
v = row.get(name)
if v is None or (isinstance(v, str) and not v.strip()):
raise Incomplete(f"{what}: {name} is required")
return v.strip() if isinstance(v, str) else v
@dataclass(frozen=True)
class Catalogue:
name: str
table: str
key: tuple
fields: tuple # non-key columns, in table order
lists: tuple = () # columns holding a JSON list
CATALOGUES = {c.name: c for c in (
Catalogue("worker_type", "hr_worker_type", ("worker_type",),
("label", "category", "payroll", "via_company", "bp_roles", "employee_subgroup", "active"), ("bp_roles",)),
Catalogue("employee_subgroup", "hr_employee_subgroup", ("subgroup",), ("label", "active")),
Catalogue("org_unit_type", "hr_org_unit_type", ("org_unit_type",), ("label", "active")),
Catalogue("event_type", "hr_event_type", ("event_type",), ("label", "kind", "active")),
Catalogue("event_reason", "hr_event_reason", ("event_type", "reason"), ("label", "immediate", "active")),
Catalogue("infogroup", "hr_infogroup", ("event_type", "record"), ("required",)),
Catalogue("rate_basis", "hr_rate_basis", ("rate_basis",), ("label", "active")),
Catalogue("pay_component_type", "hr_pay_component_type", ("component",),
("label", "worker_types", "rate_basis", "active"), ("worker_types",)),
Catalogue("pay_grade", "hr_pay_grade", ("pay_grade",),
("label", "job_code", "rate_basis", "currency", "min_amount", "max_amount")),
Catalogue("responsible_group", "hr_responsible_group", ("group_code",), ("label", "resolution", "active")),
Catalogue("document_type", "hr_document_type", ("document_type",),
("label", "access_class", "expires", "retention_category", "sensitive_kind", "active")),
Catalogue("intake_field", "hr_intake_field", ("pack", "field"),
("label", "worker_types", "required", "sensitive_kind", "input", "sort_no", "active"), ("worker_types",)),
Catalogue("draft_instruction", "hr_draft_instruction", ("draft_code",), ("label", "instructions", "fields"),
("fields",)),
Catalogue("retention_rule", "hr_retention_rule", ("category",), ("label", "keep_months", "action", "basis")),
Catalogue("settings", "hr_settings", ("key",), ("value",)),
Catalogue("job", "hr_job", ("job_code",), ("title", "family", "default_subgroup", "default_pay_grade", "active")),
Catalogue("location", "hr_location", ("location_code",), ("name", "address", "country", "timezone",
"holiday_calendar")),
Catalogue("work_schedule", "hr_work_schedule", ("schedule_code",),
("label", "weekly_minutes", "daily_pattern", "full_time_weekly_minutes", "working_weekdays"),
("working_weekdays",)),
Catalogue("holiday", "hr_holiday", ("holiday_calendar", "holiday_date"), ("name",)),
Catalogue("unit_access", "hr_unit_access", ("org_unit_id", "kind", "target"), ("role", "label")),
# hr v4: the companies an hr grant covers (principal = a sub, an email or a service client id; lower case)
Catalogue("company_grant", "hr_company_grant", ("principal", "company_code"), ("note", "granted_by")),
)}
RETENTION_CATEGORIES = ("person_detail", "sensitive_bank", "sensitive_identity", "sensitive_contact",
"right_to_work", "documents_personnel", "documents_pay", "payroll", "runs")
def validate_row(cat: Catalogue, row: dict, config: "Config") -> dict:
"""The row as it will be stored, or Incomplete/RuleBroken in words. `config` is the current catalogues, for
the cross-row rules (a reason's event type exists, a component's worker types exist)."""
if not isinstance(row, dict):
raise Incomplete(f"{cat.name}: a row is an object")
extra = sorted(set(row) - set(cat.key) - set(cat.fields))
if extra:
raise Incomplete(f"{cat.name}: unknown field(s) {', '.join(extra)} (it has {', '.join(cat.key + cat.fields)})")
out = {k: str(_req(row, k, cat.name)).strip() for k in cat.key}
for f in cat.fields:
if f in row:
out[f] = row[f]
n = cat.name
if "active" in cat.fields:
out["active"] = _bool(out.get("active", True), f"{n}.active")
for f in cat.lists:
out[f] = _list(out.get(f), f"{n}.{f}")
if n == "worker_type":
out["label"] = _req(out, "label", n)
if out.get("category") not in CATEGORIES:
raise Incomplete(f"worker_type.category is one of {', '.join(CATEGORIES)}")
if "payroll" not in row:
raise Incomplete("a worker type must say whether payroll applies (payroll: true or false)")
if "bp_roles" not in row:
raise Incomplete("a worker type must say which business-partner roles it gives (bp_roles)")
out["payroll"] = _bool(out["payroll"], "worker_type.payroll")
out["via_company"] = _bool(out.get("via_company", False), "worker_type.via_company")
bad = sorted(set(out["bp_roles"]) - set(BP_ROLES))
if bad:
raise Incomplete(f"bp_roles are {', '.join(BP_ROLES)}, not {', '.join(bad)}")
if out["via_company"] and "supplier" in out["bp_roles"]:
raise RuleBroken("a freelancer via a company is a worker; the COMPANY is the supplier finance pays")
if out["category"] == "employee" and (out["via_company"] or "supplier" in out["bp_roles"]):
raise RuleBroken("an employee is paid through payroll, never as a supplier or via a company")
elif n == "company_grant":
out["principal"] = out["principal"].lower()
elif n == "event_type":
if out.get("kind") not in KINDS:
raise Incomplete(f"event_type.kind is one of the seven kinds: {', '.join(KINDS)}")
elif n == "event_reason":
if config.event_type(out["event_type"]) is None:
raise RuleBroken(f"there is no event type {out['event_type']!r} for this reason to belong to")
out["immediate"] = _bool(out.get("immediate", False), "event_reason.immediate")
elif n == "infogroup":
et = config.event_type(out["event_type"])
if et is None:
raise RuleBroken(f"there is no event type {out['event_type']!r}")
if out["record"] not in KIND_RECORDS[et["kind"]]:
raise RuleBroken(f"a {et['kind']} event cannot write {out['record']!r} "
f"(it writes {', '.join(sorted(KIND_RECORDS[et['kind']]))})")
out["required"] = _bool(out.get("required", False), "infogroup.required")
elif n == "rate_basis":
if out["rate_basis"] not in BASES:
raise Incomplete(f"a rate basis is one of {', '.join(BASES)} (the label is yours, the basis is not)")
elif n == "pay_component_type":
missing = [w for w in out["worker_types"] if config.worker_type(w) is None]
if missing:
raise RuleBroken(f"pay component {out['component']}: no worker type {', '.join(missing)}")
if out.get("rate_basis") not in BASES:
raise Incomplete(f"pay_component_type.rate_basis is one of {', '.join(BASES)}")
elif n == "pay_grade":
for f in ("min_amount", "max_amount"):
if out.get(f) is not None:
out[f] = str(out[f])
elif n == "responsible_group":
res = str(out.get("resolution") or "")
if not (res in RESOLUTIONS or (res.startswith("pool_role:") and len(res) > 10)):
raise Incomplete("a responsible group resolves as manager, run_party, person or pool_role:<role>")
elif n == "document_type":
if out.get("access_class") not in ACCESS_CLASSES:
raise Incomplete(f"document_type.access_class is one of {', '.join(ACCESS_CLASSES)}")
out["expires"] = _bool(out.get("expires", False), "document_type.expires")
if out["access_class"] == "sensitive" and out.get("sensitive_kind") not in SENSITIVE_KINDS:
raise Incomplete("a sensitive document type names its sensitive_kind")
elif n == "intake_field":
if out.get("sensitive_kind") not in SENSITIVE_KINDS:
raise Incomplete(f"an intake field lands in the Sensitive aggregate: sensitive_kind is one of "
f"{', '.join(SENSITIVE_KINDS)}")
out["required"] = _bool(out.get("required", True), "intake_field.required")
out["sort_no"] = _int(out.get("sort_no", 0), "intake_field.sort_no")
if out.get("input", "text") not in ("text", "file"):
raise Incomplete("intake_field.input is text or file")
out["input"] = out.get("input", "text")
elif n == "draft_instruction":
_req(out, "instructions", n)
bad = sorted(set(out["fields"]) - set(DRAFTABLE))
if bad:
raise RuleBroken(f"a draft is never given {', '.join(bad)}: it may use only {', '.join(DRAFTABLE)}")
elif n == "retention_rule":
if out["category"] not in RETENTION_CATEGORIES:
raise Incomplete(f"retention categories are {', '.join(RETENTION_CATEGORIES)}")
out["keep_months"] = _int(out.get("keep_months"), "retention_rule.keep_months")
if out.get("action") not in RETENTION_ACTIONS:
raise Incomplete(f"retention_rule.action is {' or '.join(RETENTION_ACTIONS)}")
if out.get("basis") not in RETENTION_BASES:
raise Incomplete(f"retention_rule.basis is one of {', '.join(RETENTION_BASES)}")
elif n == "settings":
key = out["key"]
if key not in SETTINGS_DEFAULTS and not key.startswith(SETTING_PREFIXES):
raise Incomplete(f"no setting {key!r} (settings are {', '.join(sorted(SETTINGS_DEFAULTS))}, "
"client_size:<legal entity>)")
out["value"] = "" if out.get("value") is None else str(out["value"])
_check_setting(key, out["value"])
elif n == "location":
work.zone(str(_req(out, "timezone", n)))
if len(str(_req(out, "country", n))) != 2:
raise Incomplete("location.country is ISO 3166 alpha-2 (GB)")
out["country"] = out["country"].upper()
elif n == "work_schedule":
out["weekly_minutes"] = _int(_req(out, "weekly_minutes", n), "weekly_minutes", 0)
out["full_time_weekly_minutes"] = _int(_req(out, "full_time_weekly_minutes", n), "full_time_weekly_minutes", 1)
days = [_int(d, "working_weekdays") for d in out["working_weekdays"]] or [0, 1, 2, 3, 4]
if not set(days) <= set(range(7)):
raise Incomplete("working_weekdays are 0 (Monday) … 6 (Sunday)")
out["working_weekdays"] = sorted(set(days))
elif n == "holiday":
try:
date.fromisoformat(out["holiday_date"])
except ValueError:
raise Incomplete(f"holiday_date is a date, not {out['holiday_date']!r}") from None
elif n == "unit_access":
if out["kind"] not in ACCESS_KINDS:
raise Incomplete(f"unit access kind is one of {', '.join(ACCESS_KINDS)}")
elif n == "job":
out["title"] = _req(out, "title", n)
return out
def _check_setting(key: str, value: str) -> None:
if key == "timezone":
work.zone(value)
elif key in ("intake_validity_days", "intake_chase_every_days", "intake_chase_max", "task_chase_every_days",
"task_chase_max", "approval_chase_every_days", "approval_chase_max", "max_action_attempts",
"action_backoff_minutes", "warning_days_contract_end", "warning_days_right_to_work",
"max_backdate_days", "retention_sweep_hour"):
_int(value, key)
elif key == "rounding_mode" and value not in ROUNDING:
raise Incomplete(f"rounding_mode is one of {', '.join(ROUNDING)}")
elif key == "contract_signer_person_id" and value and not value.isdigit():
raise Incomplete("contract_signer_person_id is a person_id (the director who signs), or empty")
elif key == "chase_unit" and value not in work.UNITS:
raise Incomplete(f"chase_unit is one of {', '.join(work.UNITS)}")
elif key.startswith("client_size:") and value not in CLIENT_SIZES:
raise Incomplete(f"a client size is {', '.join(CLIENT_SIZES)} (Companies Act 2006 thresholds — yours to judge)")
class Config:
"""The catalogues as one read model: rows by catalogue, looked up by key. Built per unit of work."""
def __init__(self, rows: Optional[dict] = None, dated: Iterable = ()):
self.rows = {name: list((rows or {}).get(name, [])) for name in CATALOGUES}
self.dated = [dict(r) for r in dated] # hr_setting_row: every dated setting row, the history
def leave_reasons(self) -> list:
"""The reasons of every event type of kind Left (a dated setting's scope)."""
left = set(self.event_types_of("Left"))
return sorted({r["reason"] for r in self.rows["event_reason"] if r["event_type"] in left})
def dated_setting(self, key: str, on: date, scope: str = "") -> str:
"""The value in force on `on` (a business date): the latest row from on-or-before that day for `scope` (a leave
reason), else for the default scope '', else the setting's default."""
for s in ((scope, "") if scope else ("",)):
mine = [r for r in self.dated if r["setting_key"] == key and r["scope"] == s and r["valid_from"] <= on]
if mine:
return str(max(mine, key=lambda r: r["valid_from"])["value"])
return DATED_SETTINGS[key][0]
def dated_history(self, key: str) -> list:
return sorted((r for r in self.dated if r["setting_key"] == key), key=lambda r: (r["valid_from"], r["scope"]))
def _find(self, cat: str, **key) -> Optional[dict]:
for r in self.rows[cat]:
if all(str(r.get(k)) == str(v) for k, v in key.items()):
return r
return None
def all(self, cat: str, active_only: bool = False) -> list:
return [r for r in self.rows[cat] if not active_only or r.get("active", True)]
def companies_for(self, *principals: Optional[str]) -> Optional[frozenset]:
"""hr v4: the companies the grant of whoever is named by any of `principals` (their sub, their email)
covers — the union of their company_grant rows; None (every company) when no row names them."""
names = {str(p).strip().lower() for p in principals if p and str(p).strip()}
codes = frozenset(str(r["company_code"]) for r in self.rows["company_grant"]
if str(r.get("principal") or "").lower() in names)
return codes or None
def worker_type(self, code: str) -> Optional[dict]:
return self._find("worker_type", worker_type=code)
def event_type(self, code: str) -> Optional[dict]:
return self._find("event_type", event_type=code)
def event_types_of(self, kind: str) -> list:
return [r["event_type"] for r in self.rows["event_type"] if r["kind"] == kind and r.get("active", True)]
def reason(self, event_type: str, reason: str) -> Optional[dict]:
return self._find("event_reason", event_type=event_type, reason=reason)
def job(self, code: str) -> Optional[dict]:
return self._find("job", job_code=code)
def location(self, code: str) -> Optional[dict]:
return self._find("location", location_code=code)
def schedule(self, code: str) -> Optional[dict]:
return self._find("work_schedule", schedule_code=code)
def group(self, code: str) -> Optional[dict]:
return self._find("responsible_group", group_code=code)
def document_type(self, code: str) -> Optional[dict]:
return self._find("document_type", document_type=code)
def draft_instruction(self, code: str) -> Optional[dict]:
return self._find("draft_instruction", draft_code=code)
def component(self, code: str) -> Optional[dict]:
return self._find("pay_component_type", component=code)
def pay_grade(self, code: str) -> Optional[dict]:
return self._find("pay_grade", pay_grade=code)
def retention_rule(self, category: str) -> Optional[dict]:
return self._find("retention_rule", category=category)
def access(self, org_unit_id: str) -> list:
return [r for r in self.rows["unit_access"] if r["org_unit_id"] in (org_unit_id, "*")]
def intake_fields(self, pack: str, worker_type: str) -> list:
out = [r for r in self.rows["intake_field"] if r["pack"] == pack and r.get("active", True)
and (not r.get("worker_types") or worker_type in r["worker_types"])]
return sorted(out, key=lambda r: (int(r.get("sort_no") or 0), r["field"]))
def holidays(self, calendar: str) -> frozenset:
return frozenset(date.fromisoformat(r["holiday_date"]) for r in self.rows["holiday"]
if r["holiday_calendar"] == calendar)
# settings
def setting(self, key: str, default: Optional[str] = None) -> str:
r = self._find("settings", key=key)
if r is not None and r.get("value") not in (None, ""):
return str(r["value"])
return default if default is not None else SETTINGS_DEFAULTS.get(key, "")
def int_setting(self, key: str) -> int:
return int(self.setting(key) or SETTINGS_DEFAULTS.get(key) or 0)
@property
def timezone(self) -> str:
return self.setting("timezone")
def client_size(self, legal_entity: str) -> str:
return self.setting(f"client_size:{legal_entity}", "")
# the working-day calendar (domain service WorkingDayCalendar)
def calendar(self, location_code: Optional[str], schedule_code: Optional[str]) -> work.WorkCalendar:
sched = self.schedule(schedule_code) if schedule_code else None
weekdays = frozenset(sched.get("working_weekdays") or [0, 1, 2, 3, 4]) if sched else frozenset({0, 1, 2, 3, 4})
loc = self.location(location_code) if location_code else None
holidays = self.holidays(loc["holiday_calendar"]) if loc and loc.get("holiday_calendar") else frozenset()
return work.WorkCalendar(working_weekdays=weekdays, holidays=holidays)
def in_use_refusal(cat: str, key: tuple, uses: int) -> None:
if uses:
raise RuleBroken(f"{cat} {'|'.join(key)} is in use ({uses} record(s)); retire it (active: false) instead")
__all__ = ["CATALOGUES", "Catalogue", "Config", "validate_row", "KINDS", "KIND_RECORDS", "DRAFTABLE",
"SENSITIVE_KINDS", "SETTINGS_DEFAULTS", "BP_ROLES", "RETENTION_CATEGORIES", "in_use_refusal", "Iterable"]