dodil/erp-brick-time-off
DODIL ERP brick: time off
src/brick_time_off/application/use_cases.py
"""The use cases - what a person (or their agent) can do with their time off.
Each is declared ONCE and served as an API route and as an MCP tool. Two things are decided here and nowhere else: WHO MAY (a
PERMISSION, never a role's name) and WHICH ROWS (an employee sees their own requests and balance; a manager the requests routed to
them; HR and the owner every request). A row the caller may not see is a 404, never a 403.
The rules that shape every use case (module.yaml in_code_not_rows and tests.invariants; tests/ has one test per story):
- 'today' is the date in business_timezone (SettingsRepository.timezone), never the UTC date; absence dates are calendar dates
- nobody decides their own time off, under either login (email OR sub); the approver is hr's derived manager, fixed at the request
- one claim key per person/day operation: every half day is a tof_day_claim row (plain INSERT), a ledger entry is keyed by its
entry_ref, a request number by the company's yearly series, a decision by (request, revision), a leaver notice by hr's source_ref;
the loser of a race is a 409, never a second effect
- state only moves forward; "taken" is derived from dates, a cancellation ask is a fact on the approved request
- public holidays, the work pattern and the manager come from hr on every request: this brick keeps no copy of them
- no read-your-writes inside an open transaction (DataK3): a use case reads committed rows, writes, and reports from what it holds;
the leaver's balance is read in a SECOND transaction after the first commits
- this brick has no outbound write: hr is read-only for it, and payroll and timesheets PULL approved absences, so there is no outbox
in slice A (it arrives with the reminders and the notice of a changed closed period)
"""
import json
import re
from contextlib import contextmanager
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from decimal import Decimal
from typing import Literal, Optional
from zoneinfo import ZoneInfo
from erp_platform import auth, work
from erp_platform.bricks import Brick, Refused
from erp_platform.types import BigId
from sqlalchemy.exc import IntegrityError
from brick_time_off.application.inputs import Code, DateText, Note, Part, Reason, Ref
from brick_time_off.domain import calendar as cal
from brick_time_off.domain import request as rq
from brick_time_off.domain.errors import RuleBroken
from brick_time_off.infrastructure import seams
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared.
READ_OWN, WRITE_OWN, READ_TEAM, READ_ALL, APPROVE = ("tof:request:read_own", "tof:request:write_own", "tof:request:read_team",
"tof:request:read_all", "tof:request:approve")
# machine: held only by a service client, and by an EXACT grant - no wildcard reaches them
LEAVER_INTAKE, PAYROLL_BALANCE = "tof:leaver:intake", "tof:balance:read_for_payroll"
MACHINE = ("tof:balance:accrue", "tof:year_end:run", "tof:reminder:tick", LEAVER_INTAKE, "tof:absence:read_for_payroll", PAYROLL_BALANCE,
"tof:migration:load", "tof:migration:reconcile")
TOF = Brick("time-off", namespace="tof", version="0.2.0", ids_as_strings=True, machine_permissions=MACHINE, instructions=(
"Time off: an employee asks for days off (request_leave) and sees at once what they cost - bank holidays and days they do not work "
"are never counted, and a part-timer books in hours. Each half day can be booked once. Their manager, read from hr on the first "
"day and fixed there, approves (approve_leave) or rejects with a reason (reject_leave); nobody decides their own time off under "
"either login, so a manager's own request goes one level up. An approved request is a ledger entry; the balance "
"(get_time_off_balance) is the entitlement derived from the policy and hr plus the sum of the entries. Approved leave is cancelled "
"by asking (ask_leave_cancellation) and the manager confirming (cancel_leave); a day already past is HR's to correct. Sickness is "
"recorded, not requested (record_sickness, then confirm_return; a fit note from the 8th day) and payroll reads the facts for statutory pay, never the "
"money. HR records family leave, closure days come out of everyone's allowance, an adjustment of a balance needs a second person, and the year end "
"carries what the cap allows. Quantities are exact strings in days or hours."))
# The process view (BRICKS.md): the workflows as BPMN, GET /process/bpmn.
TOF.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text or "duplicate key" in text or "UNIQUE constraint" in text
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a racer's duplicate key (two bookings of one half day, two approvals of one request) can surface at
COMMIT, after the use case has returned - so it is caught here, around the commit, and answered as the retry-safe 409 a conflict
seen mid-use-case gets, never a 500."""
from brick_time_off.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else did the same a moment ago - load the request again and retry", gate="day_free") from None
repository = _repository_default # tests and a composition swap in their own
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words, through both doors."""
try:
yield
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
except seams.Unavailable as down:
raise Refused(down.http, down.reason) from None
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else did the same a moment ago - load the request again and retry", gate="day_free") from None
def _who(caller: dict) -> str:
return (caller.get("email") or caller.get("sub") or "").strip().lower()
def _me(caller: dict) -> rq.Identity:
"""Both of the caller's names: the same person under either is the same person."""
return rq.Identity((caller.get("email") or "").strip().lower(), (caller.get("sub") or "").strip())
def _now() -> datetime:
return work.now()
def _today(repo) -> date:
"""Business today: the date NOW (work.now()) in business_timezone - read from ONE place - never the UTC date."""
return _now().astimezone(ZoneInfo(repo.settings.timezone())).date()
def _need(caller: dict, perm: str) -> None:
if not auth.has_permission(caller, perm):
raise Refused(403, f"missing permission: {perm}")
def _bounds(reason: str, fix: str = "check the value and send it again") -> Refused:
return Refused(422, reason, gate="bounds", fix=fix)
def _date(v: Optional[str], what: str) -> date:
try:
d = date.fromisoformat(str(v).strip())
except ValueError:
raise _bounds(f"{what} is a date like 2026-12-21") from None
if not date(2000, 1, 1) <= d <= date(2100, 1, 1):
raise _bounds(f"{what} {d} is outside 2000-01-01 .. 2100-01-01")
return d
def _int(repo, key: str, on: date) -> int:
return int(Decimal(repo.settings.on(key, on)))
# ── the person, the year, the balance ─────────────────────────────────────────────────────────────
def _person(me: rq.Identity, actor: str) -> dict:
person = seams.person_by_sub(me.sub, actor) if me.sub else None
if person is None:
raise Refused(409, "hr holds no person with your sign-in, so there is no time off for you", gate="employed",
fix="ask HR to link your login to your person record")
return person
@dataclass
class Year:
"""Everything about one person's leave year of one type that a balance and a request need: hr's days for the year and the policy."""
label: int
first: date
last: date
days: list
policy: Optional[dict]
last_day: Optional[date]
ent: Optional[cal.Entitlement]
company: str = ""
def _year(repo, pid: int, company: str, ltype: dict, on: date, actor: str) -> Year:
"""The leave year `on` falls in for a type that counts against a policy: ONE hr call for the whole year, the policy picked by FTE on the
first employed day, the entitlement derived (never stored). A leaver's entitlement runs to the last day hr gave (tof_leaver)."""
policies = repo.catalog.policies(company, ltype["counts_against"], on)
start = policies[0]["leave_year_start"] if policies else "01-01"
label, first, last = cal.leave_year(on, start)
days = seams.work_pattern(pid, first, last, actor)
last_day = repo.leavers.last_day(pid)
live = [d for d in days if d.employed and (last_day is None or d.date <= last_day)]
policy = cal.pick_policy(policies, live[0].fte if live else Decimal(0)) if policies else None
ent = cal.entitlement(policy, days, first, last, last_day) if policy else None
return Year(label, first, last, days, policy, last_day, ent, company)
def _balance(repo, pid: int, ltype_code: str, y: Year, on: date) -> tuple:
"""(Balance, entries, claim rows): the derived entitlement plus the sum of the ledger, less what is pending. Four queries, whatever
the number of requests, days or other people."""
entries = repo.entries.of_person_year(pid, ltype_code, y.label)
reqs = {r["request_id"]: r["status"] for r in repo.requests.of_person_year(pid, ltype_code, y.label)}
held = repo.days.in_range(pid, y.first, y.last)
rows = [c for c in held if c["absence_ref"] in reqs]
claims = [(c["absence_date"], cal.q4(c["quantity"]), reqs[c["absence_ref"]]) for c in rows]
closure_rows = repo.closures.in_range(y.company, y.first, y.last, ltype_code) if y.company else []
closure_cost = cal.closure_deductions(closure_rows, y.days, {(c["absence_date"], c["part"]): c["absence_ref"] for c in held}, y.ent.unit, y.last_day) \
if closure_rows else (cal.ZERO, ())
bal = cal.balance(y.ent, [(e["kind"], cal.q4(e["quantity"])) for e in entries], claims, on, closure_cost)
return bal, entries, rows
def _balance_view(pid: int, ltype_code: str, y: Year, bal: cal.Balance, on: date, entries: list) -> dict:
u = bal.unit
return {"person_id": str(pid), "on": on.isoformat(), "leave_type_code": ltype_code, "leave_year": y.label, "unit": u,
"from": y.first.isoformat(), "to": y.last.isoformat(),
"entitled": cal.show(bal.entitled, u), "carried": cal.show(bal.carried, u), "lapsed": cal.show(bal.lapsed, u), "expired": cal.show(bal.expired, u),
"taken": cal.show(bal.taken, u), "booked": cal.show(bal.booked, u), "pending": cal.show(bal.pending, u), "adjustments": cal.show(bal.adjustments, u),
"bank_holidays_deducted": cal.show(bal.holiday_deduction, u), "closure_days": cal.show(bal.closures, u),
"closures": [{"date": c["date"].isoformat(), "name": c["name"], "cost": cal.show(c["cost"], u), "covered_by": c["covered_by"]} for c in bal.closure_details],
"available": cal.show(bal.available, u), "below_zero": bal.available < 0,
"day_minutes": y.ent.day_minutes, "policy_code": y.ent.policy_code,
"calculation": {"employed_days": y.ent.employed_days, "share_of_year": f"{y.ent.share:.6f}", "public_holidays": y.ent.holidays},
"entries": [{"entry_ref": e["entry_ref"], "kind": e["kind"], "quantity": cal.show(e["quantity"], u), "on_date": e["on_date"].isoformat(),
"reason": e["reason"]} for e in entries]}
def _company(pid: int, on: date, actor: str, fallback: str = "*") -> str:
return seams.employment(pid, on, actor).get("company_code") or fallback
# ── visibility: which rows ────────────────────────────────────────────────────────────────────
def _can_see(caller: dict, r: rq.Request) -> bool:
me = _me(caller)
if auth.has_permission(caller, READ_ALL):
return True
if auth.has_permission(caller, READ_OWN) and me.same_as(r.requester):
return True
return (auth.has_permission(caller, READ_TEAM) or auth.has_permission(caller, APPROVE)) and r.is_approver(me)
def _visible(repo, caller: dict, request_id: str) -> rq.Request:
r = repo.requests.get(request_id)
if r is None or not _can_see(caller, r):
raise Refused(404, f"no leave request {request_id}")
return r
def _own(repo, caller: dict, request_id: str) -> rq.Request:
"""A request of the caller's own (and only theirs): another person's, even HR's, is a 404."""
r = repo.requests.get(request_id)
if r is None or not _me(caller).same_as(r.requester):
raise Refused(404, f"no leave request {request_id}")
return r
def _view(r: rq.Request, slots: Optional[list] = None) -> dict:
out = {"request_id": r.request_id, "person_id": str(r.person_id), "company_code": r.company_code, "leave_type_code": r.leave_type_code,
"policy_code": r.policy_code or None, "leave_year": r.leave_year, "from_date": r.from_date.isoformat(), "from_part": r.from_part,
"to_date": r.to_date.isoformat(), "to_part": r.to_part, "cost": cal.show(r.cost, r.unit), "unit": r.unit, "status": r.status,
"revision": r.revision, "approver_person_id": None if r.approver_person_id is None else str(r.approver_person_id),
"approved_by": r.approved_by or None, "decided_at": r.decided_at.isoformat() if r.decided_at else None, "reason": r.reason or None,
"note": r.note or None, "stale": r.stale,
"stale_cost": None if r.stale_cost is None else cal.show(r.stale_cost, r.unit),
"cancellation_asked": r.cancel_asked_at is not None, "cancellation_refused": r.cancel_refused_reason or None}
if slots is not None:
out["days"] = [{"date": s.date.isoformat(), "part": s.part, "quantity": cal.show(s.quantity, r.unit)} for s in slots]
return out
def _summary(row: dict) -> dict:
return {"request_id": row["request_id"], "status": row["status"], "leave_type_code": row["leave_type_code"], "from_date": row["from_date"].isoformat(),
"to_date": row["to_date"].isoformat(), "cost": cal.show(row["cost"], row["unit"]), "unit": row["unit"], "company_code": row["company_code"]}
def _slots_of(repo, r: rq.Request) -> list:
return repo.days.slots(repo.days.in_range(r.person_id, r.from_date, r.to_date), r.request_id)
# ── reading ─────────────────────────────────────────────────────────────────────────────────
@TOF.use_case(kind="query", route="GET /leave-requests")
def list_time_off_requests(caller: dict, scope: Literal["mine", "waiting", "all"] = "mine") -> dict:
"""The leave requests the caller may see. scope=mine: their own. scope=waiting: the requests waiting for THEM to decide (a manager's team).
scope=all: every request (HR and the owner). Summaries only; get_time_off_request gives the days."""
with repository() as repo:
me = _me(caller)
if scope == "all":
_need(caller, READ_ALL)
rows = repo.requests.summaries(everything=True)
elif scope == "waiting":
if not (auth.has_permission(caller, READ_TEAM) or auth.has_permission(caller, APPROVE)):
raise Refused(403, f"missing permission: {READ_TEAM} or {APPROVE}")
rows = repo.requests.summaries(waiting_for=me)
else:
_need(caller, READ_OWN)
rows = repo.requests.summaries(mine=me)
return {"requests": [_summary(r) for r in rows]}
@TOF.use_case(kind="query", route="GET /leave-requests/{request_id}")
def get_time_off_request(caller: dict, request_id: Ref) -> dict:
"""One request, whole: its dates, what it costs, who decides it, each half day it holds, and whether a cancellation was asked. A request the
caller may not see is a 404."""
with repository() as repo:
r = _visible(repo, caller, request_id)
return _view(r, _slots_of(repo, r))
@TOF.use_case(kind="query", route="GET /balance")
def get_time_off_balance(caller: dict, leave_type_code: Code = "ANNUAL", on_date: Optional[DateText] = None, person_id: Optional[BigId] = None) -> dict:
"""How much time off a person has left and what it is made of, as at a business date (default today): the entitlement derived from the
policy and hr's employment (never stored), carried days, what was taken and what is booked, what is still pending approval, adjustments,
and the ledger entries behind each figure. Bank holidays of the person's nation are never deducted from a full-timer's days; a part-timer
books in hours and loses only the bank holidays that fall on a day they work. Your own, or (with tof:request:read_all) another person's."""
_need(caller, READ_OWN)
when = _date(on_date, "on_date") if on_date else None
with repository() as repo, _decide():
today = _today(repo)
on = when or today
actor, me = _who(caller), _me(caller)
if person_id is not None:
if not auth.has_permission(caller, READ_ALL):
raise Refused(404, f"no person {person_id}")
pid = int(person_id)
else:
pid = int(_person(me, actor)["person_id"])
company = _company(pid, on, actor)
lt = repo.catalog.leave_type(company, leave_type_code, on)
if lt is None or lt["counts_against"] in (None, "none"):
raise Refused(404, f"{leave_type_code} has no balance for {company}: it is not counted against a policy")
y = _year(repo, pid, company, lt, on, actor)
if y.ent is None:
raise Refused(409, f"no policy for {leave_type_code} is in force for {company} on {on}", gate="policy")
bal, entries, _ = _balance(repo, pid, leave_type_code, y, on)
return _balance_view(pid, leave_type_code, y, bal, on, entries)
@TOF.use_case(kind="query", route="GET /persons/{person_id}/balance", permission=PAYROLL_BALANCE, tool=False, aliases={"on_date": "on"})
def get_tof_balance_for_payroll(caller: dict, person_id: BigId, on_date: DateText, leave_type_code: Code = "ANNUAL") -> dict:
"""Payroll reads a person's balance as at a date (a leaver's untaken holiday, paid in lieu once): the same derivation as the person's own
screen. time-off never pays it. Held by payroll's service client alone."""
on = _date(on_date, "on")
with repository() as repo, _decide():
actor, pid = _who(caller), int(person_id)
company = _company(pid, on, actor)
lt = repo.catalog.leave_type(company, leave_type_code, on)
if lt is None or lt["counts_against"] in (None, "none"):
raise Refused(404, f"{leave_type_code} has no balance for {company}")
y = _year(repo, pid, company, lt, on, actor)
if y.ent is None:
raise Refused(409, f"no policy for {leave_type_code} is in force for {company} on {on}", gate="policy")
bal, _, _ = _balance(repo, pid, leave_type_code, y, on)
u = bal.unit
return {"person_id": str(pid), "on": on.isoformat(), "leave_type_code": leave_type_code, "unit": u, "entitled": cal.show(bal.entitled, u),
"carried": cal.show(bal.carried, u), "taken": cal.show(bal.taken, u), "booked": cal.show(bal.booked, u),
"adjustments": cal.show(bal.adjustments, u), "available": cal.show(bal.available, u), "day_minutes": y.ent.day_minutes}
# ── asking ─────────────────────────────────────────────────────────────────────────────────
def _cost_the_ask(repo, pid: int, company: str, lt: dict, first: date, last: date, from_part: str, to_part: str, actor: str, today: date):
"""The halves an ask takes and what they cost, from hr's days for the request's own leave year (ONE call, or two when the ask runs past the
year end). Returns (unit, label, year | None, slots). Refuses in the words a person reads: not employed, nothing to take, over a year end."""
counts = lt["counts_against"] not in (None, "none")
if counts:
y = _year(repo, pid, company, lt, first, actor)
if y.ent is None:
raise Refused(409, f"no policy for {lt['leave_type_code']} is in force for {company} on {first}", gate="policy",
fix="HR adds a policy for the leave type")
days = y.days if last <= y.last else seams.work_pattern(pid, first, last, actor)
unit, label = y.ent.unit, y.label
else:
y = None
days = seams.work_pattern(pid, first, last, actor)
live = [d for d in days if d.employed]
unit, label = ("hours" if live and live[0].fte < 1 else "days"), first.year
gone = cal.unemployed_day(days, first, last)
if gone:
raise Refused(409, f"not employed on {gone} (hr)", gate="employed", fix="HR checks the start or last day")
slots = cal.slots_of(days, first, last, from_part, to_part, unit)
if not slots:
raise Refused(422, "there is no working day in these dates: weekends, days you do not work and bank holidays cost nothing", gate="empty",
fix="pick days you work")
if y is not None and any(s.date > y.last for s in slots):
raise Refused(422, f"these dates run into the next leave year ({y.last + timedelta(days=1)}), which has its own balance", gate="leave_year",
fix=f"ask for the days up to {y.last} and the rest separately")
return unit, label, y, slots
def _check_free(repo, pid: int, slots: list, first: date, last: date, exclude: str = "") -> None:
held = {(c["absence_date"], c["part"]): c["absence_ref"] for c in repo.days.in_range(pid, first, last) if c["absence_ref"] != exclude}
for s in slots:
if (s.date, s.part) in held:
raise Refused(409, f"{s.date} {s.part} is already in request {held[(s.date, s.part)]}", gate="day_free",
fix="change the dates, or withdraw the other request")
def _check_balance(repo, pid: int, lt: dict, y: Optional[Year], cost: Decimal, on: date, today: date, giving_back: Decimal = Decimal(0), slots: tuple = ()) -> None:
"""The ask must fit in what is available. `slots` are the NEW halves asked for: a half on a company closure day is already paid by the allowance
(derived), so booking it is not a second cost - it is added back for the check, exactly as the balance will show it once the request holds the day."""
if y is None:
return
bal, _, _ = _balance(repo, pid, lt["leave_type_code"], y, on)
closed = {c["date"] for c in bal.closure_details if c["cost"] > 0}
giving_back = giving_back + sum((s.quantity for s in slots if s.date in closed), Decimal(0))
allow = Decimal(repo.settings.on("allow_negative_days", today))
if cost > bal.available + giving_back + allow:
have = bal.available + giving_back
raise Refused(422, f"{cal.show(cost, bal.unit)} asked, {cal.show(have, bal.unit)} available", gate="balance",
fix="ask for fewer days, or for unpaid leave")
@TOF.use_case(kind="command", route="POST /leave-requests", permission=WRITE_OWN, status=201)
def request_leave(caller: dict, from_date: DateText, to_date: Optional[DateText] = None, from_part: Part = "full", to_part: Optional[Part] = None,
leave_type_code: Code = "ANNUAL", note: Optional[Note] = None) -> dict:
"""Ask for time off (leave_request: none -> requested). Give the first day, the last day (omit it for one day) and, for a half day, which
half (am or pm); a longer ask starts whole or from lunchtime (from_part pm) and ends whole or at lunchtime (to_part am). The cost is worked out
from hr: weekends, days you do not work and your location's bank holidays cost nothing, a part-timer books in hours. Refused with the gate
named: not employed (409), a half day already in another request (409 day_free), no working day (422), or more than your balance (422
balance). Each half day is claimed in one row, so the same day is never booked twice. Your manager from hr on the first day is fixed as the
approver (your own manager's manager when you are the manager). The request is numbered UK01-TOF-2026-00381 from your company's year."""
first = _date(from_date, "from_date")
last = _date(to_date, "to_date") if to_date else first
to_part = to_part or ("full" if last != first else from_part)
with repository() as repo, _decide():
today = _today(repo)
try:
cal.check_parts(first, last, from_part, to_part, repo.settings.on("min_booking", today))
except ValueError as e:
raise _bounds(str(e)) from None
if first > today + timedelta(days=_int(repo, "max_days_ahead", today)):
raise Refused(422, f"{first} is more than {_int(repo, 'max_days_ahead', today)} days ahead", gate="max_days_ahead",
fix="ask nearer the time")
me, actor = _me(caller), _who(caller)
person = _person(me, actor)
pid = int(person["person_id"])
emp = seams.employment(pid, first, actor)
company = emp.get("company_code") or person.get("company_code")
if not company:
raise Refused(409, "hr gives no company for you on this date, so the request cannot be numbered", gate="employed")
lt = repo.catalog.leave_type(company, leave_type_code, first)
if lt is None:
raise Refused(409, f"No leave type {leave_type_code} is in force for {company} on {first}", gate="leave_type", fix="pick one of the company's types")
if lt["recorded_or_requested"] != "requested":
raise Refused(409, f"{leave_type_code} is recorded by your manager or HR, not requested", gate="leave_type")
unit, label, y, slots = _cost_the_ask(repo, pid, company, lt, first, last, from_part, to_part, actor, today)
_check_free(repo, pid, slots, first, last)
cost = cal.cost_of(slots)
_check_balance(repo, pid, lt, y, cost, first, today, slots=slots)
approver_pid, approver = rq.derive_approver(me, pid, lambda lvl: seams.manager(pid, first, lvl, actor))
number = repo.claims.take("series", f"{company}-{today.year}", _now(), actor)
r = rq.Request(f"{company}-TOF-{today.year}-{number:05d}", pid, company, me, leave_type_code, y.ent.policy_code if y else "", label, first,
from_part, last, to_part, cost, unit, approver_pid, approver, note=(note or "").strip())
repo.requests.create(r, _now())
repo.days.add(pid, r.request_id, slots, _now())
return _view(r, slots)
@TOF.use_case(kind="command", route="POST /leave-requests/{request_id}/withdraw", permission=WRITE_OWN)
def withdraw_leave(caller: dict, request_id: Ref) -> dict:
"""Withdraw your own request before it is decided (requested -> withdrawn, terminal). Its half days are released in the same transaction.
An approved request is not withdrawn: ask to cancel it (ask_leave_cancellation)."""
with repository() as repo, _decide():
r = _own(repo, caller, request_id)
r.move("withdraw_leave")
repo.claims.once("decide", f"{r.request_id}:r{r.revision}", _now(), _who(caller))
repo.days.release(r.person_id, r.request_id, r.from_date, r.to_date)
repo.requests.save(r, _now())
return _view(r, [])
# ── deciding ───────────────────────────────────────────────────────────────────────────────
def _decider(repo, caller: dict, request_id: str) -> rq.Request:
"""The request the caller decides. Order: the row is theirs to see (404), the permission (403), then the rule that nobody decides their own time off, under either login (409), then that they ARE its approver - or HR, who sees
every request (403)."""
r = _visible(repo, caller, request_id)
_need(caller, APPROVE)
me = _me(caller)
if me.same_as(r.requester):
raise Refused(409, "You asked for this time off, so you cannot decide it", gate="second_person", fix="it goes to your manager")
if not (r.is_approver(me) or auth.has_permission(caller, READ_ALL)):
raise Refused(403, "You are not the approver of this request", gate="approver", fix="it waits for the requester's manager, or HR")
return r
def _past_day(repo, r: rq.Request, today: date) -> Optional[date]:
held = [s.date for s in _slots_of(repo, r)]
return next((d for d in sorted(held) if d < today), None)
def _fresh(repo, r: rq.Request, actor: str) -> tuple:
"""hr's answer today for this request's own dates: (the halves it would take now, whether hr still says employed on every day)."""
days = seams.work_pattern(r.person_id, r.from_date, r.to_date, actor)
gone = cal.unemployed_day(days, r.from_date, r.to_date)
return cal.slots_of(days, r.from_date, r.to_date, r.from_part, r.to_part, r.unit), gone
@TOF.use_case(kind="command", route="POST /leave-requests/{request_id}/approve")
def approve_leave(caller: dict, request_id: Ref, acknowledge_overlap: bool = False) -> dict:
"""Approve a request (requested -> approved): only its approver (the requester's manager from hr, fixed at the request) or HR, and never the
requester - matched on email OR sub, so a second login of the same person is refused. Writes the ledger entry take, keyed
req:<request_id>:r<revision>, in the same transaction; payroll and timesheets can read the absence after it commits. If hr's calendar or
pattern changed since the request, the cost is recomputed, the request flagged stale and approval refused (409 stale) until
recompute_leave_request is run."""
stale = None
with repository() as repo, _decide():
r = _decider(repo, caller, request_id)
if r.status == "requested":
if r.stale:
raise Refused(409, "hr's calendar or pattern changed for these days: recompute the cost first", gate="stale",
fix="run recompute_leave_request, then approve")
slots, gone = _fresh(repo, r, _who(caller))
held = {(s.date, s.part, s.quantity) for s in _slots_of(repo, r)}
if gone or {(s.date, s.part, s.quantity) for s in slots} != held:
stale = (r, cal.cost_of(slots))
if stale is None:
me = _me(caller)
r.move("approve_leave")
repo.claims.once("decide", f"{r.request_id}:r{r.revision}", _now(), _who(caller))
r.approved_by, r.approved_sub, r.decided_at = me.email, me.sub, _now()
r.overlap_ack = "acknowledged" if acknowledge_overlap else ""
repo.entries.add(f"req:{r.request_id}:r{r.revision}", r.person_id, r.leave_type_code, r.leave_year, "take", -r.cost, r.unit, r.from_date,
me.email, me.sub, "", _now())
repo.requests.save(r, _now())
return _view(r, _slots_of(repo, r))
with repository() as repo: # the flag is committed even though this approval is refused
r, cost = stale
r.stale, r.stale_cost = True, cost
repo.requests.save(r, _now())
raise Refused(409, f"hr's calendar or pattern changed for these days: the cost is now {cal.show(cost, r.unit)}, it was {cal.show(r.cost, r.unit)}",
gate="stale", fix="run recompute_leave_request, then approve")
@TOF.use_case(kind="command", route="POST /leave-requests/{request_id}/recompute", permission=WRITE_OWN)
def recompute_leave_request(caller: dict, request_id: Ref) -> dict:
"""Recompute a pending request against hr's calendar and pattern as they are now (the requester's own request only): its half days and cost are
replaced, the stale flag cleared, and the balance gate runs again. A request already decided is not recomputed."""
with repository() as repo, _decide():
r = _own(repo, caller, request_id)
if r.status != "requested":
raise RuleBroken(f"Request {r.request_id} is {r.status}: only a request that waits for a decision is recomputed")
today, actor = _today(repo), _who(caller)
lt = repo.catalog.leave_type(r.company_code, r.leave_type_code, r.from_date)
unit, label, y, slots = _cost_the_ask(repo, r.person_id, r.company_code, lt, r.from_date, r.to_date, r.from_part, r.to_part, actor, today)
_check_free(repo, r.person_id, slots, r.from_date, r.to_date, exclude=r.request_id)
cost = cal.cost_of(slots)
_check_balance(repo, r.person_id, lt, y, cost, r.from_date, today, giving_back=r.cost)
repo.days.release(r.person_id, r.request_id, r.from_date, r.to_date)
repo.days.add(r.person_id, r.request_id, slots, _now())
r.cost, r.unit, r.stale, r.stale_cost = cost, unit, False, None
repo.requests.save(r, _now())
return _view(r, slots)
@TOF.use_case(kind="command", route="POST /leave-requests/{request_id}/reject", destructive=True)
def reject_leave(caller: dict, request_id: Ref, reason: Reason) -> dict:
"""Reject a request with a reason (requested -> rejected, terminal). Its half days are released in the same transaction. Decided by its approver
or HR, never by the requester (same rules as approve_leave)."""
if not reason.strip():
raise Refused(422, "a reason is required to reject a request", gate="bounds")
with repository() as repo, _decide():
r = _decider(repo, caller, request_id)
me = _me(caller)
r.move("reject_leave")
repo.claims.once("decide", f"{r.request_id}:r{r.revision}", _now(), _who(caller))
r.reason, r.approved_by, r.approved_sub, r.decided_at = reason.strip(), me.email, me.sub, _now()
repo.days.release(r.person_id, r.request_id, r.from_date, r.to_date)
repo.requests.save(r, _now())
return _view(r, [])
# ── cancelling approved leave ───────────────────────────────────────────────────────────────────
@TOF.use_case(kind="command", route="POST /leave-requests/{request_id}/ask-cancellation", permission=WRITE_OWN)
def ask_leave_cancellation(caller: dict, request_id: Ref) -> dict:
"""Ask to cancel your own approved leave. The ask is recorded on the request and the leave stands until it is decided: your manager confirms
(cancel_leave) or refuses (refuse_leave_cancellation). Refused when any day of it is already past - HR corrects a day already taken - or when
an ask is already open."""
with repository() as repo, _decide():
r = _own(repo, caller, request_id)
r.move("ask_leave_cancellation")
today = _today(repo)
past = _past_day(repo, r, today)
if past:
raise Refused(409, f"{past} is already past: ask HR to correct a day already taken", gate="past_day", fix="HR corrects it with an adjustment")
if r.cancel_asked_at is not None:
raise Refused(409, "a cancellation of this request is already asked and waits for your manager", gate="open_ask")
repo.claims.take("cancel_ask", r.request_id, _now(), _who(caller))
r.cancel_asked_at, r.cancel_asked_by, r.cancel_refused_reason = _now(), _who(caller), ""
repo.requests.save(r, _now())
return _view(r, _slots_of(repo, r))
@TOF.use_case(kind="command", route="POST /leave-requests/{request_id}/cancel", destructive=True)
def cancel_leave(caller: dict, request_id: Ref, reason: Reason) -> dict:
"""Cancel approved leave (approved -> cancelled, terminal): confirmed by its approver or HR, never by the requester. Writes the reversal entry
req:<request_id>:r<revision>:cancel and releases the half days in the same transaction, so payroll and timesheets read no absence for them. Refused
when any day is already past (HR corrects it with an adjustment). Nothing moves back to requested."""
if not reason.strip():
raise Refused(422, "a reason is required to cancel approved leave", gate="bounds")
with repository() as repo, _decide():
r = _decider(repo, caller, request_id)
r.move("cancel_leave")
past = _past_day(repo, r, _today(repo))
if past:
raise Refused(409, f"{past} is already past: HR corrects a day already taken with an adjustment", gate="past_day")
me = _me(caller)
repo.claims.once("cancel", f"{r.request_id}:r{r.revision}", _now(), _who(caller))
repo.entries.add(f"req:{r.request_id}:r{r.revision}:cancel", r.person_id, r.leave_type_code, r.leave_year, "reverse", r.cost, r.unit,
r.from_date, me.email, me.sub, reason.strip(), _now())
repo.days.release(r.person_id, r.request_id, r.from_date, r.to_date)
r.reason, r.cancel_asked_at = reason.strip(), None
repo.requests.save(r, _now())
return _view(r, [])
@TOF.use_case(kind="command", route="POST /leave-requests/{request_id}/refuse-cancellation")
def refuse_leave_cancellation(caller: dict, request_id: Ref, reason: Reason) -> dict:
"""Refuse the cancellation ask of approved leave, with a reason: the ask is closed and the leave stands (no ledger entry, no state change).
Decided by its approver or HR, never by the requester."""
if not reason.strip():
raise Refused(422, "a reason is required to refuse a cancellation", gate="bounds")
with repository() as repo, _decide():
r = _decider(repo, caller, request_id)
r.move("refuse_leave_cancellation")
if r.cancel_asked_at is None:
raise Refused(409, "nobody asked to cancel this request", gate="open_ask")
repo.claims.once("cancel_refuse", f"{r.request_id}:{repo.claims.count('cancel_ask', r.request_id)}", _now(), _who(caller))
r.cancel_asked_at, r.cancel_refused_reason = None, reason.strip()
repo.requests.save(r, _now())
return _view(r, _slots_of(repo, r))
# ── hr: a person left ──────────────────────────────────────────────────────────────────────────
LEAVER_REF = re.compile(r"^hr:left:(\d+):(\d+)$")
def _kept_shape(r: rq.Request, kept: list) -> tuple:
"""to_date and to_part of an approved request cut back to the halves that are kept."""
last = max(s.date for s in kept)
parts = {s.part for s in kept if s.date == last}
if last == r.from_date:
return last, r.from_part
return last, "am" if parts == {"am"} else "full"
def _apply_leaver(repo, r: rq.Request, held: list, last_day: date, who: str, me: rq.Identity) -> Optional[dict]:
"""What a leaver's last day does to one open request that has days after it. Nothing is deleted:
pending, any day after -> rejected (reason left), halves released
approved, every day after -> cancelled (reason left), reversal entry, halves released
approved, some after -> the days up to the last day are kept as revision n+1: revision n is reversed, n+1 taken, later halves released."""
after = [s for s in held if s.date > last_day]
if not after:
return None
at = _now()
if r.status == "requested":
r.move("reject_leave")
repo.claims.once("decide", f"{r.request_id}:r{r.revision}", at, who)
r.reason = "left"
repo.days.release(r.person_id, r.request_id, r.from_date, r.to_date)
repo.requests.save(r, at)
return {"request_id": r.request_id, "action": "rejected"}
kept = [s for s in held if s.date <= last_day]
repo.claims.once("cancel", f"{r.request_id}:r{r.revision}", at, who)
repo.entries.add(f"req:{r.request_id}:r{r.revision}:cancel", r.person_id, r.leave_type_code, r.leave_year, "reverse", r.cost, r.unit, r.from_date,
me.email, me.sub, "left", at)
if not kept:
r.move("cancel_leave")
r.reason = "left"
repo.days.release(r.person_id, r.request_id, r.from_date, r.to_date)
repo.requests.save(r, at)
return {"request_id": r.request_id, "action": "cancelled"}
repo.days.release_after(r.person_id, r.request_id, last_day, r.to_date)
old = r.revision
r.revision, r.reason = old + 1, "left"
r.to_date, r.to_part = _kept_shape(r, kept)
r.cost = cal.cost_of(kept)
repo.entries.add(f"req:{r.request_id}:r{r.revision}", r.person_id, r.leave_type_code, r.leave_year, "take", -r.cost, r.unit, r.from_date,
me.email, me.sub, "left", at)
repo.requests.save(r, at)
return {"request_id": r.request_id, "action": "shortened", "revision": r.revision}
def _leaver_answer(row: dict) -> dict:
done = json.loads(row["cancelled_json"] or "[]")
return {"person_id": str(row["person_id"]), "last_day": row["last_day"].isoformat(),
"cancelled_requests": [d["request_id"] for d in done if d["action"] in ("cancelled", "rejected")],
"changed_requests": done, "balance": json.loads(row["balance_json"]) if row["balance_json"] else None}
@TOF.use_case(kind="command", route="POST /employees/{person_id}/leaver", permission=LEAVER_INTAKE, idempotent=True, tool=False)
def intake_tof_leaver(caller: dict, person_id: BigId, last_day: DateText, source_ref: Ref) -> dict:
"""hr tells us a person left (after its Left event committed): leave booked after the last day is cancelled (reason left, its half days released,
a reversal entry written; a request that runs past the last day is cut back to it as the next revision; a pending one is rejected) and the
balance is derived to the last day - entitlement 25 x 289 / 365 = 19.79, never rounded up - so payroll can read what is owed. Nothing is
deleted. Idempotent on source_ref (hr:left:<person_id>:<n>): the same ref answers the same. A moved last day arrives as n+1. Held by hr's
service client alone."""
when = _date(last_day, "last_day")
m = LEAVER_REF.match(source_ref)
if not m or int(m.group(1)) != int(person_id):
raise _bounds(f"source_ref is hr:left:{person_id}:<n>", "send the reference hr minted for this person")
pid, actor = int(person_id), _who(caller)
me = _me(caller)
with repository() as repo, _decide():
row = repo.leavers.get(source_ref)
if row is None:
today = _today(repo)
done = []
if repo.settings.on("leaver_cancel_future", today).lower() == "true":
open_rows = [r for r in repo.requests.open_of_person(pid) if r["to_date"] > when]
if open_rows:
held = repo.days.in_range(pid, min(r["from_date"] for r in open_rows), max(r["to_date"] for r in open_rows))
for raw in open_rows:
r = repo.requests._make(raw)
out = _apply_leaver(repo, r, repo.days.slots(held, r.request_id), when, actor, me)
if out:
done.append(out)
repo.claims.once("leaver", source_ref, _now(), actor)
row = {"source_ref": source_ref, "person_id": pid, "n": int(m.group(2)), "last_day": when, "cancelled_json": json.dumps(done),
"balance_json": None}
repo.leavers.add(row, _now())
row["created_at"] = _now()
if row["balance_json"] is None: # second transaction: the balance is read after the cancellations committed
with repository() as repo, _decide():
row = repo.leavers.get(source_ref)
company = _company(pid, when, actor)
lt = repo.catalog.leave_type(company, "ANNUAL", when)
y = _year(repo, pid, company, lt, when, actor) if lt else None
if y is not None and y.ent is not None:
bal, _, _ = _balance(repo, pid, "ANNUAL", y, when)
balance = {"leave_type_code": "ANNUAL", "unit": bal.unit, "available": cal.show(bal.available, bal.unit)}
else:
balance = {"leave_type_code": "ANNUAL", "unit": None, "available": None}
repo.leavers.set_balance(row, balance, _now())
row["balance_json"] = json.dumps(balance)
return _leaver_answer(row)
# The sickness slice declares its use cases on the same brick; imported last so every helper above exists (a composition that loads TOF from here gets them too).
from brick_time_off.application import adjustments, closures, family, migration, settings, sickness, year_end # noqa: E402,F401