dodil/erp-brick-time-off
DODIL ERP brick: time off
src/brick_time_off/application/adjustments.py
"""Manual adjustments of a balance (TOF-12): a reason, and a second person.
HR proposes (a signed quantity in the person's unit, a reason); a DIFFERENT holder of tof:balance:adjust approves or rejects; only an approved one
is written - as the ledger entry `adj:<person>:<type>:<year>:<n>` (kind adjust) in the same transaction as the decision. Nobody adjusts their own
balance, proposes or decides alone: the proposer and the person whose balance it is are both refused as decider, under either login (email OR sub).
Entries are never edited or deleted - a wrong adjustment is answered by another.
GAPS (AGENTS.md): the spec says only 'my own balance is refused'; the two-person rule (proposer is not decider) is the brief's. No deadline is enforced on
HR's override of a lapsed carry-over."""
from typing import Annotated, Literal, Optional
from erp_platform import auth
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, Reason, Ref
from brick_time_off.domain import adjustment as adj
from brick_time_off.domain import calendar as cal
from brick_time_off.infrastructure import seams
TOF = hub.TOF
ADJUST = "tof:balance:adjust"
Quantity = Annotated[str, Field(max_length=24)]
def _view(a: adj.Adjustment) -> dict:
return {"adjustment_id": a.adjustment_id, "person_id": str(a.person_id), "company_code": a.company_code, "leave_type_code": a.leave_type_code,
"leave_year": a.leave_year, "quantity": cal.show(a.quantity, a.unit), "unit": a.unit, "on_date": a.on_date.isoformat(), "reason": a.reason,
"status": a.status, "proposed_by": a.proposed_by.email or a.proposed_by.sub, "decided_by": (a.decided_by.email or a.decided_by.sub) or None,
"decision_reason": a.decision_reason or None}
def _my_pid(caller: dict) -> Optional[int]:
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
def _someone_else(caller: dict, a_person: int) -> None:
"""The caller is not the person whose balance it is - under either login (their person in hr, found by sub)."""
if _my_pid(caller) == a_person:
raise Refused(409, "This is your own balance: another HR user adjusts it", gate="second_person", fix="ask another HR user")
@TOF.use_case(kind="command", route="POST /balance-adjustments", permission=ADJUST, status=201)
def propose_time_off_adjustment(caller: dict, person_id: BigId, quantity: Quantity, reason: Reason, leave_type_code: Code = "ANNUAL",
on_date: Optional[DateText] = None) -> dict:
"""Propose an adjustment of a person's balance (HR): a signed quantity in their unit (days or hours - see get_time_off_balance), say +1.00 to give a
day back or -0.5 to take half a day, with a reason. It changes nothing until ANOTHER holder of tof:balance:adjust approves it
(approve_time_off_adjustment). Never your own balance (409). The leave year is the one the date falls in (default today). Quantity: not zero, at most
4 decimals, at most 400 either way."""
if not reason.strip():
raise Refused(422, "an adjustment needs a reason", gate="bounds")
try:
q = adj.parse_quantity(quantity)
except ValueError as e:
raise hub._bounds(str(e)) from None
when = hub._date(on_date, "on_date") if on_date else None
pid, actor = int(person_id), hub._who(caller)
with hub.repository() as repo, hub._decide():
today = hub._today(repo)
on = when or today
_someone_else(caller, pid)
company = hub._company(pid, on, actor, fallback="")
lt = repo.catalog.leave_type(company or "*", leave_type_code, on)
if not company or lt is None or lt["counts_against"] in (None, "none"):
raise Refused(404, f"{leave_type_code} has no balance for person {pid} on {on}")
y = hub._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")
n = repo.claims.take("adj", f"{pid}:{leave_type_code}:{y.label}", hub._now(), actor)
a = adj.Adjustment(f"adj:{pid}:{leave_type_code}:{y.label}:{n}", pid, company, leave_type_code, y.label, q, y.ent.unit, on, reason.strip(), hub._me(caller))
repo.adjustments.create(a, hub._now())
return _view(a)
def _decide_adjustment(caller: dict, adjustment_id: str, key: str, note: str) -> tuple:
with hub.repository() as repo, hub._decide():
a = repo.adjustments.get(adjustment_id)
if a is None:
raise Refused(404, f"no adjustment {adjustment_id}")
me = hub._me(caller)
if me.same_as(a.proposed_by):
raise Refused(409, "You proposed this adjustment, so you cannot decide it", gate="second_person", fix="another HR user decides it")
_someone_else(caller, a.person_id)
a.move(key)
repo.claims.once("decide_adjustment", a.adjustment_id, hub._now(), hub._who(caller))
a.decided_by, a.decided_at, a.decision_reason = me, hub._now(), note.strip()
if key == "approve_adjustment":
repo.entries.add(a.adjustment_id, a.person_id, a.leave_type_code, a.leave_year, "adjust", a.quantity, a.unit, a.on_date,
a.proposed_by.email, a.proposed_by.sub, f"{a.reason} (approved by {me.email or me.sub})", hub._now())
repo.adjustments.save(a, hub._now())
return a, _view(a)
@TOF.use_case(kind="command", route="POST /balance-adjustments/{adjustment_id}/approve", permission=ADJUST)
def approve_time_off_adjustment(caller: dict, adjustment_id: Ref, note: Optional[Reason] = None) -> dict:
"""Approve a proposed adjustment (proposed -> approved): another HR user than its proposer, and never the person whose balance it is - under either
login. In the same transaction the ledger entry adj:<person>:<type>:<year>:<n> is written with the proposer's name and the reason; entries are never
edited or deleted. Deciding twice is refused."""
return _decide_adjustment(caller, adjustment_id, "approve_adjustment", note or "")[1]
@TOF.use_case(kind="command", route="POST /balance-adjustments/{adjustment_id}/reject", permission=ADJUST, destructive=True)
def reject_time_off_adjustment(caller: dict, adjustment_id: Ref, reason: Reason) -> dict:
"""Reject a proposed adjustment with a reason (proposed -> rejected, terminal): nothing is written to the ledger. Same second-person rules as approving."""
if not reason.strip():
raise Refused(422, "a reason is required to reject an adjustment", gate="bounds")
return _decide_adjustment(caller, adjustment_id, "reject_adjustment", reason)[1]
@TOF.use_case(kind="query", route="GET /balance-adjustments")
def list_time_off_adjustments(caller: dict, status: Optional[Literal["proposed", "approved", "rejected"]] = None, person_id: Optional[BigId] = None) -> dict:
"""The adjustments proposed, approved and rejected (HR and the owner), newest key last; filter by status or person. A person's approved ones are also
entries on their balance."""
if not (auth.has_permission(caller, ADJUST) or auth.has_permission(caller, hub.READ_ALL)):
raise Refused(403, f"missing permission: {ADJUST}")
with hub.repository() as repo:
return {"adjustments": [_view(repo.adjustments._make(r)) for r in repo.adjustments.listed(status, None if person_id is None else int(person_id))]}