dodil/erp-brick-time-off

DODIL ERP brick: time off

src/brick_time_off/domain/request.py

"""The leave request: one ask for time off by one person, decided by their manager (module.yaml workflows.leave_request).

Plain Python - no SQLAlchemy, FastAPI, MCP or identity. The rules an HR manager checks first live here:

  - the state machine only moves FORWARD: requested -> approved | rejected | withdrawn, approved -> cancelled; "taken" is never a
    state (an approved day in the past), and a cancellation ASK is a fact on the approved request, never a state
  - nobody decides their own time off: matched on email OR sub (either one is the same person)
  - the approver is hr's derived manager on the first day, FIXED at the request; a manager's own request goes one level up
"""
from dataclasses import dataclass
from datetime import date, datetime
from decimal import Decimal
from typing import Callable, Optional

from brick_time_off.domain.errors import RuleBroken

STATES = ("requested", "approved", "rejected", "withdrawn", "cancelled")
TERMINAL = ("rejected", "withdrawn", "cancelled")

# transition key -> (states it may start from, the state it ends in). Every arrow goes forward; tests try the reverse of each.
# ask_leave_cancellation and refuse_leave_cancellation are approved -> approved: a recorded fact, not a move.
MOVES = {
    "approve_leave": (("requested",), "approved"),
    "reject_leave": (("requested",), "rejected"),
    "withdraw_leave": (("requested",), "withdrawn"),
    "cancel_leave": (("approved",), "cancelled"),
    "ask_leave_cancellation": (("approved",), "approved"),
    "refuse_leave_cancellation": (("approved",), "approved"),
}


@dataclass(frozen=True)
class Identity:
    """Who a person is to this brick: both of their logins' names. The same person under either is the same person."""
    email: str = ""
    sub: str = ""

    def same_as(self, other: "Identity") -> bool:
        a, b = _norm(self.email), _norm(other.email)
        s, t = (self.sub or "").strip(), (other.sub or "").strip()
        return bool((a and a == b) or (s and s == t))


def _norm(v: Optional[str]) -> str:
    return (v or "").strip().lower()


@dataclass
class Request:
    request_id: str
    person_id: int
    company_code: str
    requester: Identity
    leave_type_code: str
    policy_code: str
    leave_year: int
    from_date: date
    from_part: str
    to_date: date
    to_part: str
    cost: Decimal
    unit: str
    approver_person_id: Optional[int] = None
    approver: Identity = Identity()
    status: str = "requested"
    revision: int = 1
    approved_by: str = ""
    approved_sub: str = ""
    decided_at: Optional[datetime] = None
    reason: str = ""
    note: str = ""
    overlap_ack: str = ""
    stale: bool = False
    stale_cost: Optional[Decimal] = None
    cancel_asked_at: Optional[datetime] = None
    cancel_asked_by: str = ""
    cancel_refused_reason: str = ""
    decided_by_system: bool = False
    created_at: Optional[datetime] = None

    def move(self, key: str) -> None:
        """Take a transition's arrow; a state that has no such arrow is a broken rule in the rule's own words."""
        starts, end = MOVES[key]
        if self.status not in starts:
            raise RuleBroken(f"Request {self.request_id} is {self.status}: {key.replace('_', ' ')} needs it to be {' or '.join(starts)}")
        self.status = end

    # ── who decides ────────────────────────────────────────────────────────────────────────
    def decider_ok(self, me: Identity) -> None:
        """Nobody decides their own time off, under either login; a request not routed to this person is not theirs to decide."""
        if me.same_as(self.requester):
            raise RuleBroken("You asked for this time off, so you cannot decide it: it goes to your manager")

    def is_approver(self, me: Identity) -> bool:
        return bool(self.approver.email or self.approver.sub) and me.same_as(self.approver)


def derive_approver(requester: Identity, requester_person_id: int, ask: Callable[[int], Optional[dict]]) -> tuple:
    """hr's derived manager on the first day (level 1), or level 2 when level 1 is the requester themselves. `ask(level)` is hr's
    {person_id, pool_user_sub, email} or None. Returns (person_id | None, Identity): None when hr names nobody (HR decides)."""
    for level in (1, 2):
        m = ask(level)
        if not m:
            return None, Identity()
        who = Identity((m.get("email") or "").strip().lower(), (m.get("pool_user_sub") or "").strip())
        pid = m.get("person_id")
        if who.same_as(requester) or (pid is not None and int(pid) == int(requester_person_id)):
            continue
        return (int(pid) if pid is not None else None), who
    return None, Identity()