"""The Sensitive aggregate: one person's sensitive data, under its OWN root hr_sensitive(person_id, version).
* a separate root: no manager-facing use case loads it, and the Person repository never joins it;
* read and written only through use cases that require hr:sensitive:read / hr:sensitive:write (the
application answers 404, never 403, to anyone else — a 403 would confirm there is something);
* each kind (bank, NI number, home address …) is sliced; a correction keeps the old value (superseded);
* its version is checked on save: two finance users correcting bank details at once — one gets 409.
"""
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Optional
from brick_hr.domain.configuration import SENSITIVE_KINDS
from brick_hr.domain.errors import Incomplete
from brick_hr.domain.slices import Slice, Timeline, WITHDRAWN
# Which retention category each kind belongs to (hr_retention_rule rows decide how long each is kept).
CATEGORY_OF = {"bank": "sensitive_bank", "ni_number": "sensitive_identity", "tax_id": "sensitive_identity",
"date_of_birth": "sensitive_identity", "vat_number": "sensitive_identity",
"company_number": "sensitive_identity", "home_address": "sensitive_contact",
"emergency_contact": "sensitive_contact", "right_to_work_evidence": "right_to_work"}
def check_value(kind: str, value) -> dict:
"""A sensitive value as stored: an object, or a string wrapped as {"value": ...}."""
if kind not in SENSITIVE_KINDS:
raise Incomplete(f"sensitive data is one of {', '.join(SENSITIVE_KINDS)}, not {kind!r}")
if value is None or value == "" or value == {}:
raise Incomplete(f"{kind}: a value is required")
if isinstance(value, dict):
return {str(k): v for k, v in value.items()}
if isinstance(value, (str, int)):
return {"value": str(value).strip()}
raise Incomplete(f"{kind}: a value is text or an object")
@dataclass
class Sensitive:
person_id: int
version: int = 0
erased_at: Optional[datetime] = None
anonymised_at: Optional[datetime] = None
kinds: dict = field(default_factory=dict) # kind -> Timeline (data: {"value": {...}, "object_key", "sha256"})
def timeline(self, kind: str) -> Timeline:
return self.kinds.setdefault(kind, Timeline(f"{kind}"))
def record(self, kind: str, value, valid_from: date, *, now: datetime, by: str, correction: bool = False,
object_key: Optional[str] = None, sha256: Optional[str] = None) -> Slice:
data = {"value": check_value(kind, value), "object_key": object_key, "sha256": sha256}
t = self.timeline(kind)
if correction:
# a correction of what we hold for that date: the slice covering it is superseded, kept readable
cur = t.at(valid_from) or t.last()
if cur is not None:
return t.correct(cur.valid_from, data, now=now, by=by, reason="corrected")
if any(s.valid_from == valid_from for s in t.active()):
return t.correct(valid_from, data, now=now, by=by, reason="corrected")
return t.change(valid_from, data, now=now, by=by, reason="recorded", merge=False)
def current(self, d: date) -> dict:
"""The current value per kind (the latest slice covering `d`, else the latest recorded)."""
out = {}
for kind, t in sorted(self.kinds.items()):
s = t.at(d) or t.last()
if s is not None:
out[kind] = {"value": s.data.get("value"), "valid_from": s.valid_from.isoformat(),
"has_file": bool(s.data.get("object_key"))}
return out
def present(self, d: date) -> set:
return {k for k, t in self.kinds.items() if (t.at(d) or t.last()) is not None}
def changed_between(self, kind: str, start: date, end: date) -> list:
t = self.kinds.get(kind)
return [s for s in (t.active() if t else []) if start <= s.valid_from <= end]
def purge(self, kinds: set, *, now: datetime) -> list:
"""Erase these kinds (every slice, superseded ones too): the values are gone, not hidden. Returns the
object keys the caller must delete with them."""
keys = []
for k in sorted(kinds):
t = self.kinds.pop(k, None)
if t is None:
continue
keys += [s.data.get("object_key") for s in t.slices if s.data.get("object_key")]
if kinds:
self.erased_at = now
return keys
__all__ = ["Sensitive", "CATEGORY_OF", "check_value", "WITHDRAWN"]