dodil/erp-brick-time-off

DODIL ERP brick: time off

src/brick_time_off/infrastructure/seams.py

"""The neighbour this brick asks, behind ONE adapter - the domain never knows how it is reached.

  hr   the person (by the caller's sub), their employment on a date, their derived manager on a date, and their work pattern over a
       range (contracted minutes, public holidays of THEIR location, employed or not, FTE - per day). Time-off keeps no copy of the
       employee, a calendar or a pattern: no person in hr, no leave.

What hr serves (its signed spec, erp-method modules/hr/module.yaml seams; tests/conftest.py checks every call):
  GET /persons?pool_user_sub=                       -> {person_id, bp_id, company_code}; 404 when no person carries the sub        (HR-58)
  GET /persons/{person_id}/employment?on=           -> {employed, company_code, worker_type, location_code, ...}                (HR-56)
  GET /persons/{person_id}/manager?on=&level=       -> {manager: {person_id, bp_id, pool_user_sub, email} | null}               (HR-57)
  GET /persons/{person_id}/work-pattern?from=&to=   -> {days: [{date, employed, contracted_minutes, public_holiday, location_code,
                                                        holiday_calendar, fte}]}; 422 {gate: bounds} for more than 400 days      (HR-62)
Every answer is a reply, never an exception: not installed (409), failed (502), did not answer (503).
Tests register an in-memory fake with `erp_platform.seam.register_local("hr", app)`.
"""
from datetime import date
from decimal import Decimal, InvalidOperation
from typing import Optional

from erp_platform.seam import Seam

from brick_time_off.domain.calendar import Day

HR = Seam("hr", "HR_BASE_URL", roles=["service"], permissions=["hr:employment:read", "hr:manager:read", "hr:work_pattern:read"])


class Unavailable(Exception):
    """A neighbour could not answer. `http` is what this brick answers: 409 not installed, 502 failed, 503 silent."""

    def __init__(self, http: int, reason: str):
        super().__init__(reason)
        self.http, self.reason = http, reason


def _hr(path: str, params: dict, actor: str) -> Optional[dict]:
    """A 200 body, None for a 404, or Unavailable."""
    if not HR.installed:
        raise Unavailable(409, "hr is not part of this installation, so nobody can book time off: the person, their pattern and manager come from hr")
    r = HR.get(path, params=params, actor=actor)
    if r.ok:
        return r.data
    if r.status == 404:
        return None
    if r.status and r.status >= 500:
        raise Unavailable(502, f"hr failed ({r.status}: {r.reason or 'no reason given'}) - try again")
    if r.status:
        raise Unavailable(502, f"hr answered {r.status} ({r.reason or 'no reason given'})")
    raise Unavailable(503, f"hr did not answer ({r.reason or 'no reason given'}) - try again")


def person_by_sub(sub: str, actor: str = "") -> Optional[dict]:
    """The signed-in user's person: {person_id, bp_id, company_code}, or None when hr holds nobody with that sub."""
    return _hr("/persons", {"pool_user_sub": sub}, actor)


def employment(person_id: int, on: date, actor: str = "") -> dict:
    """The person's employment on a date: {employed, company_code, worker_type, ...} (employed False when hr has no such person)."""
    return _hr(f"/persons/{person_id}/employment", {"on": on.isoformat()}, actor) or {"employed": False}


def manager(person_id: int, on: date, level: int, actor: str = "") -> Optional[dict]:
    """hr's derived manager `level` steps up on a date: {person_id, bp_id, pool_user_sub, email}, or None."""
    got = _hr(f"/persons/{person_id}/manager", {"on": on.isoformat(), "level": level}, actor)
    return (got or {}).get("manager")


def work_pattern(person_id: int, first: date, last: date, actor: str = "") -> list:
    """The person's days from `first` to `last` (at most 400 apart): [Day], or [] when hr has no such person. ONE call, however long."""
    got = _hr(f"/persons/{person_id}/work-pattern", {"from": first.isoformat(), "to": last.isoformat()}, actor)
    if got is None:
        return []
    try:
        return [Day(date.fromisoformat(d["date"]), bool(d["employed"]), int(d.get("contracted_minutes") or 0), d.get("public_holiday") or None,
                    Decimal(str(d.get("fte") or "0"))) for d in got["days"]]
    except (KeyError, TypeError, ValueError, InvalidOperation):
        raise Unavailable(502, "hr's work pattern did not have the shape its contract promises - try again") from None