dodil/erp-brick-time-off
DODIL ERP brick: time off
src/brick_time_off/application/family.py
"""Family leave (TOF-09): maternity, paternity, adoption and shared parental leave RECORDED by HR with their statutory facts.
The leave is a tof_request (approved by its recording: nobody requests it), so it claims its half days like any absence and payroll reads it through GET
/absences; the facts payroll needs to assess statutory pay - the expected week of childbirth (ewc) and when the evidence arrived - are kept in tof_family_leave,
and up to ten keeping-in-touch days are recorded on it and NEVER end it. time-off computes no pay. A maternity leave that starts before the 11th week before
the EWC is refused with the date named. HR never records a leave for themselves (their own person in hr, found by sub: second person).
GAPS (AGENTS.md): the MATB1 document is not uploaded (its date is the fact); the person's own login is unknown to HR's recording, so the person reads their leave
through their balance and payroll's pull, not as 'their request'; enhanced maternity pay is payroll's (a scheme row); annual leave keeps accruing over the leave
because the entitlement is derived from hr's employment, which this brick does not change."""
from datetime import timedelta
from typing import Annotated, Optional
from erp_platform.bricks import Refused
from erp_platform.types import BigId
from pydantic import Field
from brick_time_off.application import use_cases as hub
from brick_time_off.application.inputs import Code, DateText, Ref
from brick_time_off.domain import calendar as cal
from brick_time_off.domain import request as rq
from brick_time_off.infrastructure import seams
TOF = hub.TOF
ABSENCE_RECORD, MAX_KIT_DAYS, EWC_WEEKS_BEFORE = "tof:absence:record", 10, 11
def _kind(lt: dict) -> str:
"""The statutory kind payroll reads: the pay treatment without its prefix (maternity, paternity, adoption, shared_parental)."""
return str(lt["pay_treatment"]).removeprefix("statutory_")
def _view(r: rq.Request, fam: dict, kit: list, kind: str) -> dict:
return {"request_id": r.request_id, "person_id": str(r.person_id), "company_code": r.company_code, "leave_type_code": r.leave_type_code, "status": r.status,
"from_date": r.from_date.isoformat(), "to_date": r.to_date.isoformat(), "weeks": fam["weeks"], "cost": cal.show(r.cost, r.unit), "unit": r.unit,
"statutory": {"kind": kind, "ewc": fam["ewc"].isoformat() if fam["ewc"] else None,
"evidence_received_on": fam["evidence_received_on"].isoformat() if fam["evidence_received_on"] else None, "kit_days": len(kit),
"kit_dates": sorted(kit)}}
def _my_pid(caller: dict):
sub = hub._me(caller).sub
person = seams.person_by_sub(sub, hub._who(caller)) if sub else None
return int(person["person_id"]) if person else None
@TOF.use_case(kind="command", route="POST /family-leave", permission=ABSENCE_RECORD, status=201)
def record_family_leave(caller: dict, person_id: BigId, leave_type_code: Code, from_date: DateText, weeks: Annotated[int, Field(ge=1, le=52)] = 52,
ewc: Optional[DateText] = None, evidence_received_on: Optional[DateText] = None) -> dict:
"""Record a family leave (HR): MATERNITY, PATERNITY, ADOPTION or SHARED_PARENTAL - a leave type recorded rather than requested, with a statutory pay
treatment. From the first day for `weeks` weeks (52 is the whole entitlement): the leave claims its half days (a day already in other leave or sickness is
refused 409 day_free) and payroll reads it with its statutory facts: ewc (the expected week of childbirth, a Sunday or any day of the week) and
evidence_received_on. Maternity needs the ewc and may not start before the 11th week before it (refused with the date). Never your own leave (409)."""
first = hub._date(from_date, "from_date")
due = hub._date(ewc, "ewc") if ewc else None
got = hub._date(evidence_received_on, "evidence_received_on") if evidence_received_on else None
last = first + timedelta(weeks=weeks) - timedelta(days=1)
pid, actor = int(person_id), hub._who(caller)
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
if got is not None and got > today:
raise hub._bounds(f"the evidence cannot have arrived on {got}, which is after today ({today})")
if _my_pid(caller) == pid:
raise Refused(409, "This is your own leave: another HR user records it", gate="second_person", fix="ask another HR user")
emp = seams.employment(pid, first, actor)
company = emp.get("company_code")
if not emp.get("employed") or not company:
raise Refused(409, f"not employed on {first} (hr)", gate="employed", fix="HR checks the start or last day")
lt = repo.catalog.leave_type(company, leave_type_code, first)
if lt is None or lt["recorded_or_requested"] != "recorded" or not str(lt["pay_treatment"]).startswith("statutory_"):
raise Refused(409, f"{leave_type_code} is not a family leave type of {company}: one that is recorded and has a statutory pay treatment", gate="leave_type")
if leave_type_code == "MATERNITY":
if due is None:
raise hub._bounds("maternity leave is recorded with the expected week of childbirth (ewc)")
earliest = due - timedelta(weeks=EWC_WEEKS_BEFORE)
if first < earliest:
raise Refused(422, f"maternity leave cannot start before {earliest}, the 11th week before the expected week of childbirth ({due})", gate="earliest_start",
fix=f"start on or after {earliest}")
days = seams.work_pattern(pid, first, last, actor)
gone = cal.unemployed_day(days, first, last)
if gone:
raise Refused(409, f"not employed on {gone} (hr)", gate="employed", fix="HR checks the leave's dates against the contract")
live = [d for d in days if d.employed]
unit = "hours" if live and live[0].fte < 1 else "days"
slots = cal.slots_of(days, first, last, "full", "full", unit)
if not slots:
raise Refused(422, "there is no working day in these dates", gate="empty")
hub._check_free(repo, pid, slots, first, last)
me = hub._me(caller)
number = repo.claims.take("series", f"{company}-{today.year}", hub._now(), actor)
r = rq.Request(f"{company}-TOF-{today.year}-{number:05d}", pid, company, rq.Identity(), leave_type_code, "", first.year, first, "full", last, "full",
cal.cost_of(slots), unit, status="approved", revision=1, approved_by=me.email or me.sub, approved_sub=me.sub, decided_at=hub._now(),
reason=f"recorded by HR ({leave_type_code})")
repo.requests.create(r, hub._now())
repo.days.add(pid, r.request_id, slots, hub._now())
repo.entries.add(f"req:{r.request_id}:r1", pid, leave_type_code, first.year, "take", -r.cost, unit, first, me.email, me.sub, "", hub._now())
fam = {"request_id": r.request_id, "person_id": pid, "company_code": company, "leave_type_code": leave_type_code, "ewc": due, "evidence_received_on": got,
"weeks": weeks, "recorded_by": me.email or me.sub, "recorded_sub": me.sub}
repo.family.add(fam, hub._now())
return _view(r, fam, [], _kind(lt))
@TOF.use_case(kind="command", route="POST /family-leave/{request_id}/keeping-in-touch-day", permission=ABSENCE_RECORD, status=201)
def record_keeping_in_touch_day(caller: dict, request_id: Ref, work_date: DateText) -> dict:
"""Record a keeping-in-touch day on a family leave (HR): a day inside the leave on which the person worked. At most ten in all, each date once, and they
never end the leave - the days stay claimed. Payroll reads the count as statutory.kit_days."""
day = hub._date(work_date, "work_date")
with hub.repository() as repo, hub._decide():
r = repo.requests.get(request_id)
fam = repo.family.get(request_id) if r is not None else None
if r is None or fam is None:
raise Refused(404, f"no family leave {request_id}")
if r.status != "approved":
raise Refused(409, f"{request_id} is {r.status}: keeping-in-touch days go on a leave that stands", gate="status")
if not r.from_date <= day <= r.to_date:
raise Refused(422, f"{day} is outside the leave ({r.from_date} .. {r.to_date})", gate="bounds")
if day > hub._today(repo):
raise Refused(422, f"{day} has not happened yet", gate="bounds")
kit = repo.family.kit_days([request_id]).get(request_id, [])
if day.isoformat() in kit:
raise Refused(409, f"{day} is already a keeping-in-touch day of {request_id}", gate="kit_once")
if len(kit) >= MAX_KIT_DAYS:
raise Refused(409, f"{request_id} already has the {MAX_KIT_DAYS} keeping-in-touch days the law allows", gate="kit_limit")
repo.claims.take("kit_day", request_id, hub._now(), hub._who(caller), ref=day.isoformat())
return _view(r, fam, kit + [day.isoformat()], _kind(repo.catalog.leave_type(r.company_code, r.leave_type_code, r.from_date)))
@TOF.use_case(kind="query", route="GET /family-leave/{request_id}", permission=ABSENCE_RECORD)
def get_time_off_family_leave(caller: dict, request_id: Ref) -> dict:
"""One family leave with its statutory facts and keeping-in-touch days (HR)."""
with hub.repository() as repo:
r = repo.requests.get(request_id)
fam = repo.family.get(request_id) if r is not None else None
if r is None or fam is None:
raise Refused(404, f"no family leave {request_id}")
return _view(r, fam, repo.family.kit_days([request_id]).get(request_id, []), _kind(repo.catalog.leave_type(r.company_code, r.leave_type_code, r.from_date)))