dodil/erp-brick-expenses
DODIL ERP brick: expenses
src/brick_expenses/infrastructure/seams.py
"""The neighbours this brick asks, each behind ONE adapter - the domain never knows how they are reached.
hr the person (by the caller's sub), their employment on a date, and their derived manager on a date.
Expenses keeps no copy of the employee: no person in hr, no claim.
gl the ledger: an approved claim posts ONE journal whose id is derived from the report, revision and event.
payments the payout: the employee is paid back as a payout to their person partner; payments calls us back on the statement.
POST /payouts/{payout_ref}/cancel {reason} -> 200 {payout_ref, status: cancelled} (a repeat the same); 409 {gate: not_exported} once exported
gl (fx) GET /fx-rates/{currency}?on=&type=spot&to= -> {currency, to, type, on, rate_date, rate, source}; 404 {gate: fx_rate} (GL-33)
ar POST /invoices/raise {source_kind: expense, source_id, bp_id, currency, lines} -> 201 the draft invoice; 200 the same for the same source_id (AR-20)
tax POST /documents/compute -> {document_ref, status, lines: [{line_no, tax_code, rate_pct, net, vat, gross,
vat_reclaimable, vat_non_reclaimable}], totals, gates, version} (TAX-53)
POST /documents/{document_ref}/check-contents -> {document_ref, status, ready, title, gates: [{gate, met, reason, fix}]} (TAX-50)
What they serve (the callee's signed spec, erp-method modules/<id>/module.yaml seams; tests/conftest.py checks every call):
hr 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, cost_centre, bp_id, ...} (HR-56)
GET /persons/{person_id}/manager?on=&level= -> {manager: {person_id, bp_id, pool_user_sub, email} | null} (HR-57)
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: requested}; 200 the same; 409 other amount; 422 {gate}
Every answer is a reply, never an exception: not installed (409), failed (502), did not answer (503).
Tests register in-memory fakes with `erp_platform.seam.register_local("hr", app)`.
"""
import json
from datetime import date
from typing import Optional
from erp_platform.capabilities import CapabilitySeam, ManualTask
from erp_platform.seam import Seam, derived_id
MAIL = CapabilitySeam("mail.send") # a reminder is an email through the composition's mail provider, or a task for a person when it has none
HR = Seam("hr", "HR_BASE_URL", roles=["service"], permissions=["hr:employment:read", "hr:manager:read"])
LEDGER = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post", "gl:fx:read"])
AR = Seam("ar", "AR_BASE_URL", roles=["service"], permissions=["ar:invoice:raise"])
TAX = Seam("tax", "TAX_BASE_URL", roles=["service"], permissions=["tax:document_tax:compute"])
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) -> str:
"""The journal's id, DERIVED (gl checks it): posting twice is gl's replay, never a second journal."""
return str(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 _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 claim: the person, their company 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, cost_centre, bp_id, ...} (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 _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 post_journal(body: dict, actor: str) -> dict:
"""POST the stored journal body to gl, unchanged. A 409/422 is a refusal (the period is closed: gl's gate and reason); no answer is a retry."""
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:
"""POST the stored payout request to payments, unchanged (payments replays the same payout_ref; another amount is its 409)."""
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))
def fx_rate(currency: str, on: date, to: str, actor: str = "") -> Optional[dict]:
"""gl's spot rate for a currency on a date, converted TO the company's currency: {rate, rate_date, source, ...}; None when gl holds no rate
on or before the date (the line is refused with the currency and date named). Never silently 1: a foreign line without a rate is not a line."""
if not LEDGER.installed:
raise Unavailable(409, f"gl is not part of this installation, so there is no rate to convert {currency} to {to}")
r = LEDGER.get(f"/fx-rates/{currency}", params={"on": on.isoformat(), "type": "spot", "to": to}, 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"gl failed ({r.status}: {r.reason or 'no reason given'}) - try again")
if r.status:
raise Unavailable(502, f"gl answered {r.status} ({r.reason or 'no reason given'})")
raise Unavailable(503, f"gl did not answer ({r.reason or 'no reason given'}) - try again")
def tax_installed() -> bool:
return TAX.installed
def compute_tax(body: dict, actor: str) -> dict:
"""POST one receipt to tax's compute (purchase, no partner: the merchant is a treatment). The `_outcome` shape: ok with data, a refusal
(tax's gate and reason: no rule for the pair), or a transient failure to retry."""
return _outcome(TAX.post("/documents/compute", body, actor=actor))
def check_contents(document_ref: str, body: dict, actor: str) -> dict:
"""Ask tax whether the receipt is a valid VAT invoice (full, or a less detailed one under the threshold)."""
return _outcome(TAX.post(f"/documents/{document_ref}/check-contents", body, actor=actor))
def cancel_payout(payout_ref: str, reason: str, actor: str) -> dict:
"""Ask payments to cancel a payout not yet in an exported run. A 409 {gate: not_exported} is its refusal: the payment stands."""
if not PAYMENTS.installed:
return {"ok": False, "retry": True, "status": 0, "gate": "", "reason": "payments is not part of this installation: finance cancels by hand",
"fix": "install payments"}
return _outcome(PAYMENTS.post(f"/payouts/{payout_ref}/cancel", {"reason": reason}, actor=actor))
def send_mail(key: str, from_email: str, to: list, subject: str, body: str, actor: str) -> dict:
"""Hand one reminder to the mail capability (idempotent on `key`: a retried pass is the stored answer). {ok, manual, retry, reason}: with no provider
it is a task for a person (manual), a provider that is down is retried, and one that answers 4xx is our bug and raises."""
res = MAIL.call(key, {"from_email": from_email, "to": to, "cc": [], "subject": subject[:200], "body_text": body, "body_html": None}, actor=actor)
if isinstance(res, ManualTask):
return {"ok": True, "manual": True, "retry": False, "reason": res.reason}
if res.status == "done":
return {"ok": True, "manual": False, "retry": False, "reason": ""}
return {"ok": False, "manual": False, "retry": bool(res.retryable), "reason": f"mail capability {res.status}: {res.message or ''}".strip()}
def ar_installed() -> bool:
return AR.installed
def raise_invoice(body: dict, actor: str) -> dict:
"""POST one re-billed line to ar, unchanged: ONE draft invoice per source_id (ar replays the same), never issued here. A 422 is ar's refusal."""
if not AR.installed:
return {"ok": False, "retry": True, "status": 0, "gate": "", "reason": "ar is not part of this installation: re-billing needs it", "fix": "install ar"}
return _outcome(AR.post("/invoices/raise", body, actor=actor))