dodil/erp-brick-time-off
DODIL ERP brick: time off
src/brick_time_off/application/year_end.py
"""The year end (TOF-17): carry what may be carried, lapse the rest, expire carried days that were not used by their date - an idempotent job.
The rules, each of them also in domain/calendar.py and tested:
- on the first day after a leave year, each person's AVAILABLE days of that year (entitlement + carried in + adjustments - what was spent, as at the
year's last day) are split: up to the cap carry into the next year (`carry:<person>:<type>:<year>`, written in the NEW year), the rest lapse
(`lapse:<person>:<type>:<year>`, written in the OLD year, so the person and HR see what the year end took). A negative balance carries nothing.
- the cap is in DAYS, in force on the old year's LAST day (so a new cap dated for the new year's first day does not touch the year that just ended,
TOF-14): the smaller of the setting carry_over_max_days and the policy's carry_over_max. For someone who books in hours it is converted at their
contracted day.
- after the carried days' expiry date (the earlier of the setting carry_over_expires and the policy's, counted from the new year's first day) the carried
days not used are written off (`expire:<person>:<type>:<year>`, in the new year). Carried days are used FIRST, by the entries dated up to the expiry
date.
- one claim key per person operation: every entry is keyed, written with a plain INSERT, and each person is its own transaction, so a second pass - or
a second worker at the same moment - writes nothing twice and one person's failure never stops the rest.
GAPS (written in AGENTS.md): hr has no roster seam for time-off, so the pass takes `person_ids` from the composition's worker, or - when given none - the people
this brick holds a request or an entry for; HR's override of a lapse is a manual adjustment (TOF-12) with no deadline enforced; a person whose unit changes
between the years (full-time to part-time) is carried in the old year's unit and flagged for HR."""
from datetime import date, timedelta
from decimal import Decimal
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
from brick_time_off.domain import calendar as cal
TOF = hub.TOF
YEAR_END = "tof:year_end:run"
LeaveYear = Annotated[int, Field(ge=2000, le=2100)]
def _caps(repo, policy: dict, on: date) -> tuple:
"""(cap in days, expiry MM-DD): the smaller cap and the earlier expiry of the company's setting and the policy's row."""
cap = Decimal(repo.settings.on("carry_over_max_days", on))
if policy.get("carry_over_max") is not None:
cap = min(cap, Decimal(str(policy["carry_over_max"])))
mmdd = min(repo.settings.on("carry_over_expires", on), policy.get("carry_over_expires") or "12-31")
return cap, mmdd
def _carry_one(pid: int, year: int, ltype: str, today: date, actor: str, me) -> dict:
"""The carry and the lapse of ONE person, in their own transaction. Returns what was written, or why nothing was."""
with hub.repository() as repo, hub._decide():
# the label names the year a leave year STARTS in, so 31 December of it is always inside it, whatever the policy's start month
probe = date(year, 12, 31)
company = hub._company(pid, probe, actor)
lt = repo.catalog.leave_type(company, ltype, probe)
if lt is None or lt["counts_against"] in (None, "none"):
return {"person_id": str(pid), "skipped": f"{ltype} has no balance for {company}"}
y = hub._year(repo, pid, company, lt, probe, actor)
if today <= y.last:
return {"person_id": str(pid), "skipped": f"the leave year {year} ends on {y.last}"}
if not any(d.employed for d in y.days):
return {"person_id": str(pid), "skipped": f"hr has no employment for this person in {year}"}
if y.ent is None or y.policy is None:
return {"person_id": str(pid), "skipped": "no policy was in force"}
if y.last_day is not None:
return {"person_id": str(pid), "skipped": f"left on {y.last_day}: the balance is paid, not carried"}
carry_ref, lapse_ref = f"carry:{pid}:{ltype}:{year}", f"lapse:{pid}:{ltype}:{year}"
if repo.entries.refs([carry_ref, lapse_ref]):
return {"person_id": str(pid), "unchanged": True}
bal, _, _ = hub._balance(repo, pid, ltype, y, y.last)
cap_days, _ = _caps(repo, y.policy, y.last)
cap = cap_days if bal.unit == "days" else cal.q4(cap_days * Decimal(y.ent.day_minutes) / 60)
carry, lapse = cal.carry_split(bal.available, cap)
out = {"person_id": str(pid), "available": cal.show(bal.available, bal.unit), "carried": cal.show(carry, bal.unit), "lapsed": cal.show(lapse, bal.unit),
"unit": bal.unit}
at = hub._now()
if carry > 0:
repo.entries.add(carry_ref, pid, ltype, year + 1, "carry", carry, bal.unit, y.last + timedelta(days=1), me.email, me.sub,
f"carried from {year}", at)
if lapse > 0:
repo.entries.add(lapse_ref, pid, ltype, year, "lapse", -lapse, bal.unit, y.last, me.email, me.sub, f"over the carry-over limit of {cal.show(cap, bal.unit)}", at)
if carry == 0 and lapse == 0:
out["nothing"] = True
return out
def _expire_one(pid: int, year: int, ltype: str, today: date, actor: str, me) -> dict:
"""The expiry of the carried days of ONE person (carried from `year` into year + 1)."""
with hub.repository() as repo, hub._decide():
probe = date(year + 1, 12, 31)
company = hub._company(pid, probe, actor)
lt = repo.catalog.leave_type(company, ltype, probe)
if lt is None:
return {"person_id": str(pid), "skipped": f"{ltype} has no balance for {company}"}
y = hub._year(repo, pid, company, lt, probe, actor)
if y.policy is None or y.ent is None:
return {"person_id": str(pid), "skipped": "no policy was in force"}
_, mmdd = _caps(repo, y.policy, y.first)
expiry = cal.expiry_date(y.first, mmdd)
if today <= expiry:
return {"person_id": str(pid), "skipped": f"the carried days expire after {expiry}"}
ref = f"expire:{pid}:{ltype}:{year}"
if repo.entries.refs([ref]):
return {"person_id": str(pid), "unchanged": True}
entries = repo.entries.of_person_year(pid, ltype, y.label)
carried = sum((cal.q4(e["quantity"]) for e in entries if e["entry_ref"] == f"carry:{pid}:{ltype}:{year}"), Decimal(0))
spent = -sum((cal.q4(e["quantity"]) for e in entries if e["kind"] in cal.AGAINST and e["on_date"] <= expiry), Decimal(0))
unused = cal.unused_carry(carried, spent)
if unused <= 0:
return {"person_id": str(pid), "expired": cal.show(0, y.ent.unit), "unit": y.ent.unit}
repo.entries.add(ref, pid, ltype, y.label, "expire", -unused, y.ent.unit, expiry, me.email, me.sub, f"carried days unused by {expiry}", hub._now())
return {"person_id": str(pid), "expired": cal.show(unused, y.ent.unit), "unit": y.ent.unit}
@TOF.use_case(kind="command", route="POST /year-end/run", permission=YEAR_END, idempotent=True, tool=False)
def run_tof_year_end(caller: dict, leave_year: LeaveYear, leave_type_code: Code = "ANNUAL",
person_ids: Optional[Annotated[list[BigId], Field(max_length=500)]] = None) -> dict:
"""The year-end pass for one leave year (machine: the composition's daily worker): once the year has ended, each person's unused days carry into the
next year up to the cap and the rest lapse; once the carried days' expiry date has passed, the unused carried days expire. Idempotent: every entry is
keyed carry:/lapse:/expire:<person>:<type>:<year>, so a second run - or two at once - writes nothing twice. `person_ids` is the roster to pass (hr
offers time-off no roster route; at most 500 a call); without it the people this brick holds a request or entry for in that year. Each person is
their own transaction; the answer lists who carried, who lapsed days (HR may override with an adjustment), who expired, who was skipped and why."""
with hub.repository() as repo:
today = hub._today(repo)
pids = sorted({int(p) for p in person_ids}) if person_ids else repo.entries.persons_known(leave_type_code, leave_year)
carriers = {int(r["person_id"]) for r in repo.entries.of_kind("carry", leave_type_code, leave_year + 1)
if r["entry_ref"].endswith(f":{leave_type_code}:{leave_year}")}
actor, me = hub._who(caller), hub._me(caller)
done = {"leave_year": leave_year, "leave_type_code": leave_type_code, "as_of": today.isoformat(), "carried": [], "lapsed": [], "expired": [], "skipped": [],
"unchanged": 0}
for pid in pids:
try:
got = _carry_one(pid, leave_year, leave_type_code, today, actor, me)
except Refused as r:
got = {"person_id": str(pid), "skipped": r.reason}
_tally(done, got, "carried")
for pid in sorted(carriers & {int(p) for p in person_ids} if person_ids else carriers):
try:
got = _expire_one(pid, leave_year, leave_type_code, today, actor, me)
except Refused as r:
got = {"person_id": str(pid), "skipped": r.reason}
_tally(done, got, "expired")
return done
def _tally(done: dict, got: dict, key: str) -> None:
if got.get("unchanged"):
done["unchanged"] += 1
elif "skipped" in got:
done["skipped"].append(got)
elif key == "carried":
if Decimal(got["carried"]) > 0:
done["carried"].append({"person_id": got["person_id"], "quantity": got["carried"], "unit": got["unit"]})
if Decimal(got["lapsed"]) > 0:
done["lapsed"].append({"person_id": got["person_id"], "quantity": got["lapsed"], "unit": got["unit"]})
elif Decimal(got["expired"]) > 0:
done["expired"].append({"person_id": got["person_id"], "quantity": got["expired"], "unit": got["unit"]})
@TOF.use_case(kind="query", route="GET /year-end/lapses")
def list_time_off_lapses(caller: dict, leave_year: LeaveYear, leave_type_code: Code = "ANNUAL") -> dict:
"""The days that lapsed at a leave year's end, per person - what HR reviews before overriding with an adjustment (sickness or family leave that
prevented the leave carries by law). HR and the owner only."""
hub._need(caller, hub.READ_ALL)
with hub.repository() as repo:
rows = repo.entries.of_kind("lapse", leave_type_code, leave_year)
return {"leave_year": leave_year, "leave_type_code": leave_type_code,
"lapses": [{"person_id": str(r["person_id"]), "entry_ref": r["entry_ref"], "quantity": cal.show(-cal.q4(r["quantity"]), r["unit"]), "unit": r["unit"],
"reason": r["reason"]} for r in rows]}