dodil/erp-brick-payroll
DODIL ERP brick: payroll
src/brick_payroll/infrastructure/seams.py
"""The neighbours this brick asks, each behind ONE adapter - the domain never knows how they are reached.
hr the payroll extract (pay, starters, leavers, HR-60), the employment on a date (company, cost centre, the person
partner who is paid, HR-56), the signed-in person (HR-58) and the work pattern for proration (HR-62).
Payroll keeps no copy of salaries: no extract, no run.
time-off approved absences in the period (TOF-20) and a leaver's holiday balance on the last day (TOF-22).
gl the ledger: an approved run posts ONE journal whose id is derived from the run and the event (GL-30).
payments the payouts: one per payslip, batched by the run, paid by Bacs on the pay date (PAY-52). Payroll never
sees a bank account (payments reads it from hr, HR-59).
What they serve (the callee's signed spec, erp-method modules/<id>/module.yaml seams; tests/conftest.py checks every call):
hr POST /payroll-extracts {legal_entity_ref, period} -> 201/200 {batch_id, state, produced_at, lines}
GET /payroll-extracts/{batch_id} -> {batch_id, state, starters, leavers, lines, bank_changed}
POST /payroll-extracts/{batch_id}/sent {sent_ref} -> {batch_id, state: sent, sent_ref}; 409 {gate: sent_once}
GET /persons/{person_id}/employment?on= -> {employed, company_code, cost_centre, bp_id, ...}
GET /persons?pool_user_sub= -> {person_id, bp_id, company_code}; 404 nobody
GET /persons/{person_id}/work-pattern?from=&to= -> {days: [{date, employed, contracted_minutes, public_holiday, ...}]}
GET /persons/{person_id}/payroll-identity?on= -> {legal_first_names, legal_last_name, date_of_birth, ni_number|null, home_address} (HR-61)
time-off GET /absences?company_code=&from=&to= -> {absences: [{absence_ref, revision, person_id, pay_treatment, status, days}]}
GET /persons/{person_id}/balance?leave_type_code=&on= -> {unit, available, ...}
gl POST /journal-entry/journals -> 201 {journal_id, replay: false}; 200 replay; 409 {gate: period_open}
payments POST /payouts -> 201 {payout_ref, payout_id, status}; 200 the same; 409 another amount; 422 {gate}
Every answer is a reply, never an exception: not installed (409), failed (502), did not answer (503).
"""
import json
from datetime import date
from typing import Optional
from urllib.parse import quote
from erp_platform.seam import Seam, derived_id
HR = Seam("hr", "HR_BASE_URL", roles=["service"],
permissions=["hr:employment:read", "hr:work_pattern:read", "hr:payroll_extract:read", "hr:payroll_extract:mark_sent",
"hr:payroll_identity:read"])
TIME_OFF = Seam("time-off", "TIME_OFF_BASE_URL", roles=["service"], permissions=["tof:absence:read_for_payroll", "tof:balance:read_for_payroll"])
LEDGER = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post"])
PAYMENTS = Seam("payments", "PAYMENTS_BASE_URL", roles=["service"], permissions=["pay:payout:intake"])
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 journal_id(source_document: str, event: str) -> int:
"""The journal's id, DERIVED (gl checks it): posting twice is gl's replay, never a second journal."""
return derived_id(source_document, event)
def _body(text: str) -> dict:
try:
got = json.loads(text)
return got if isinstance(got, dict) else {}
except (ValueError, TypeError):
return {}
def _read(seam: Seam, what: str, method: str, path: str, actor: str, params: Optional[dict] = None, body: Optional[dict] = None) -> Optional[dict]:
"""A 2xx body, None for a 404, or Unavailable."""
if not seam.installed:
raise Unavailable(409, f"{seam.name} is not part of this installation: {what}")
r = seam.get(path, params=params, actor=actor) if method == "GET" else seam.post(path, body, 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"{seam.name} failed ({r.status}: {r.reason or 'no reason given'}) - try again")
if r.status:
raise Unavailable(502, f"{seam.name} answered {r.status} ({r.reason or 'no reason given'})")
raise Unavailable(503, f"{seam.name} did not answer ({r.reason or 'no reason given'}) - try again")
NO_HR = "pay, starters and leavers come from hr's payroll extract, so no run can be calculated without it"
def batch_id(company_code: str, period: str) -> str:
return f"{company_code}|{period}"
def produce_extract(company_code: str, period: str, actor: str) -> dict:
"""POST /payroll-extracts: hr (re)produces the draft batch while it is a draft; a sent batch answers as it was."""
return _read(HR, NO_HR, "POST", "/payroll-extracts", actor, body={"legal_entity_ref": company_code, "period": period}) or {}
def extract(company_code: str, period: str, actor: str) -> Optional[dict]:
return _read(HR, NO_HR, "GET", f"/payroll-extracts/{quote(batch_id(company_code, period), safe='')}", actor)
def employment(person_id: int, on: date, actor: str) -> dict:
return _read(HR, NO_HR, "GET", f"/persons/{int(person_id)}/employment", actor, params={"on": on.isoformat()}) or {"employed": False}
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."""
if not sub:
return None
return _read(HR, NO_HR, "GET", "/persons", actor, params={"pool_user_sub": sub})
def work_pattern(person_id: int, frm: date, to: date, actor: str) -> list:
got = _read(HR, NO_HR, "GET", f"/persons/{int(person_id)}/work-pattern", actor, params={"from": frm.isoformat(), "to": to.isoformat()}) or {}
return [{**d, "date": date.fromisoformat(d["date"])} for d in got.get("days") or []]
def identity(person_id: int, on: date, actor: str) -> dict:
"""The identity HMRC needs (HR-61), read only when a P45, P60 or full payment submission is built and written only into that file."""
got = _read(HR, NO_HR, "GET", f"/persons/{int(person_id)}/payroll-identity", actor, params={"on": on.isoformat()})
if got is None:
raise Unavailable(502, f"hr holds no identity for person {person_id}: HMRC cannot be told who was paid")
return got
NO_TOF = "approved absences come from time-off"
def absences(company_code: str, frm: date, to: date, actor: str, person_id: Optional[int] = None) -> list:
"""Approved and recorded absences in the period (one person's with person_id), or [] when time-off is not installed (absences are then manual lines)."""
if not TIME_OFF.installed:
return []
params = {"company_code": company_code, "from": frm.isoformat(), "to": to.isoformat(), **({"person_id": int(person_id)} if person_id is not None else {})}
got = _read(TIME_OFF, NO_TOF, "GET", "/absences", actor, params=params) or {}
return got.get("absences") or []
def leave_balance(person_id: int, on: date, actor: str) -> Optional[dict]:
if not TIME_OFF.installed:
return None
return _read(TIME_OFF, NO_TOF, "GET", f"/persons/{int(person_id)}/balance", actor, params={"leave_type_code": "ANNUAL", "on": on.isoformat()})
def _outcome(r) -> dict:
"""What a neighbour said to a write: ok, or {gate, reason, fix} for a refusal, or a transient failure (retry)."""
if r.ok:
return {"ok": True, "data": r.data, "retry": False}
if r.status and 400 <= r.status < 500:
body = _body(r.reason)
return {"ok": False, "retry": False, "status": r.status, "gate": body.get("gate", ""), "reason": body.get("reason") or r.reason or f"answered {r.status}",
"fix": body.get("fix", "")}
why = f"failed ({r.status}: {r.reason or 'no reason given'})" if r.status else f"did not answer ({r.reason or 'no reason given'})"
return {"ok": False, "retry": True, "status": r.status or 0, "gate": "", "reason": why, "fix": "it is retried on the next pass"}
def mark_extract_sent(batch: str, sent_ref: str, actor: str) -> dict:
if not HR.installed:
return {"ok": False, "retry": True, "status": 0, "gate": "", "reason": "hr is not installed", "fix": "install hr"}
return _outcome(HR.post(f"/payroll-extracts/{quote(batch, safe='')}/sent", {"sent_ref": sent_ref}, actor=actor))
def post_journal(body: dict, actor: str) -> dict:
if not LEDGER.installed:
return {"ok": False, "retry": True, "status": 0, "gate": "", "reason": "no ledger is installed, so nothing was posted", "fix": "install gl"}
return _outcome(LEDGER.post("/journal-entry/journals", body, actor=actor))
def request_payout(body: dict, actor: str) -> dict:
if not PAYMENTS.installed:
return {"ok": False, "retry": True, "status": 0, "gate": "", "reason": "payments is not part of this installation: finance pays by hand",
"fix": "install payments"}
return _outcome(PAYMENTS.post("/payouts", body, actor=actor))