"""Sensitive data (HR-13): NI/tax id, bank, home address, date of birth, emergency contact, right-to-work evidence.
Its own aggregate and its own routes. These use cases declare NO door permission on purpose: a door-level 403
would tell a manager the record exists. Instead the use case checks hr:sensitive:read / hr:sensitive:write itself
and answers 404 to anyone without it — the same through the API and through MCP, and for a manager's own
reports too. The Person repository has no join path to these tables; no manager-facing use case loads them.
"""
from datetime import date
from typing import Optional, Union
from erp_platform.bricks import Refused
from erp_platform.types import BigId, Email, Limit, Name, Note
from erp_platform.repository import CONFLICT_REASON
from brick_hr.application import support as S
from brick_hr.application import use_cases as uc
from brick_hr.domain.sensitive import Sensitive
HR = uc.HR
def _gate(caller: dict, perm: str, person_id: int) -> None:
if not S.has(caller, perm):
raise Refused(404, f"no sensitive data for person {person_id}")
@HR.use_case(kind="query", route="GET /people/{person_id}/sensitive", etag="version")
def show_sensitive(caller: dict, person_id: BigId) -> dict:
"""A person's sensitive data — the current value per kind, and its history (corrections kept). HR and finance
only (hr:sensitive:read); for anyone else the route exists and the record does not: 404."""
_gate(caller, S.SENSITIVE_READ, person_id)
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller)
if uow.persons.get(person_id) is None or not view.person_in_companies(person_id):
raise Refused(404, f"no sensitive data for person {person_id}")
s = uow.sensitive.get(person_id)
if s is None:
return {"person_id": person_id, "version": 0, "current": {}, "history": {}}
history = {k: [{"valid_from": sl.valid_from.isoformat(), "valid_to": S.due_label(sl.valid_to) or "open",
"rev": sl.rev, "active": sl.active, "value": sl.data.get("value"),
"recorded_by": sl.recorded_by, "recorded_at": S.iso(sl.recorded_at)}
for sl in sorted(t.slices, key=lambda x: (x.valid_from, x.rev))]
for k, t in sorted(s.kinds.items())}
return {"person_id": person_id, "version": s.version, "current": s.current(view.today),
"history": history}
@HR.use_case(kind="command", route="POST /people/{person_id}/sensitive", if_match="expected_version")
@S.retry_unless_versioned
def record_sensitive(caller: dict, person_id: BigId, kind: Name, value: Union[str, dict],
valid_from: Optional[S.Day] = None, correction: bool = False,
expected_version: Optional[int] = None) -> dict:
"""Record or correct one kind of sensitive data (bank, ni_number, home_address …). A change is a new slice from
valid_from (today by default); correction=true supersedes what we hold for that date and keeps the old value.
Pass expected_version from show_sensitive: two people correcting bank details at once — one gets 409. HR and
finance only (hr:sensitive:write); 404 for anyone else."""
_gate(caller, S.SENSITIVE_WRITE, person_id)
at = S.now()
with S.rules():
with uc.repository() as uow:
view = S.View(uow, caller, at)
if uow.persons.get(person_id) is None or not view.person_in_companies(person_id):
raise Refused(404, f"no sensitive data for person {person_id}")
s = uow.sensitive.get(person_id) or Sensitive(person_id)
if expected_version is not None and int(expected_version) != s.version:
raise Refused(409, CONFLICT_REASON)
sl = s.record(kind, value, valid_from or view.today, now=at, by=S.actor(caller), correction=correction)
uow.sensitive.save(s)
return {"person_id": person_id, "kind": kind, "valid_from": sl.valid_from.isoformat(), "rev": sl.rev,
"version": s.version}
__all__ = ["show_sensitive", "record_sensitive"]