"""The PrivacyRequest aggregate (hr_privacy_request) and the retention schedule's rules.
* a subject-access request is due one month from receipt (UK GDPR Art. 12(3)); its export holds everything:
every slice including superseded ones, events, documents, sensitive data, run steps naming the person, and
their business-partner fields;
* erasure (Art. 17): each data category is erased or anonymised UNLESS a retention rule with a
legal-obligation basis still holds it (Art. 17(3)) — then it is kept and the basis recorded;
* the retention sweep purges or anonymises each category N months after the person's last engagement ended;
every action is a retention-log row (run_date|person_id|category), so a second run changes nothing.
The module records and applies the customer's rules; it gives no legal advice.
"""
import calendar
from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Optional
from brick_hr.domain.configuration import RETENTION_CATEGORIES, Config
from brick_hr.domain.errors import Incomplete
ACCESS, ERASURE = "access", "erasure"
def add_months(d: date, months: int) -> date:
y, m = divmod(d.month - 1 + months, 12)
y, m = d.year + y, m + 1
return date(y, m, min(d.day, calendar.monthrange(y, m)[1]))
def due_on(received_on: date) -> date:
"""One month from receipt (Art. 12(3)): the same day next month, or its last day when shorter."""
return add_months(received_on, 1)
def request_id(kind: str, person_id: int, received_on: date) -> str:
if kind not in (ACCESS, ERASURE):
raise Incomplete("a privacy request is access or erasure")
return f"{kind}:{int(person_id)}:{received_on.isoformat()}"
@dataclass
class PrivacyRequest:
request_id: str
person_id: int
kind: str
received_on: date
due_on: date
done_at: Optional[datetime] = None
done_by: Optional[str] = None
export_object_key: Optional[str] = None
outcome: dict = field(default_factory=dict)
version: int = 0
def retention_due(rule: dict, ended_on: Optional[date], today: date) -> bool:
"""Is a category past its keep period? Only for someone whose last engagement has ended."""
if ended_on is None:
return False
return add_months(ended_on, int(rule["keep_months"])) < today
def erasure_plan(config: Config, ended_on: Optional[date], today: date) -> dict:
"""Per category: (action, basis). A legal-obligation hold still running keeps the category."""
plan = {}
for cat in RETENTION_CATEGORIES:
rule = config.retention_rule(cat)
if rule and rule["basis"] == "legal_obligation" and not retention_due(rule, ended_on, today):
plan[cat] = ("kept", f"legal obligation: kept {rule['keep_months']} months after the engagement ends")
else:
default = "anonymise" if cat in ("person_detail", "payroll", "runs") else "purge"
plan[cat] = ((rule or {}).get("action") or default, (rule or {}).get("basis") or "erasure request")
return plan
__all__ = ["PrivacyRequest", "due_on", "request_id", "retention_due", "erasure_plan", "add_months", "ACCESS",
"ERASURE"]