dodil/erp-brick-payments
DODIL ERP brick: payments
src/brick_payments/infrastructure/seams.py
"""The neighbours this brick asks, each behind ONE adapter - the domain never knows how they are reached.
ar (invoicing) the documents being settled: invoices and credit notes (their totals, currency, dates), and
a write-off PROPOSAL. Read-only for us: payments never writes ar_ tables (PAY-41).
bp (business-partner) does this customer exist (as the caller may see it)?
gl (the ledger) optional: money in and out posts as a journal whose id is derived from the payment/refund id.
What ar serves (its use cases, read in the wiring pass 2026-10-22):
GET /invoice-totals?invoice_ids=a,b,c | ?bp_id= -> {"invoices": {id: {bp_id, lifecycle, total, currency, issue_date, due_date, invoice_no}}}
GET /credit-notes?bp_id=&status=approved -> {"credit_notes": [{credit_id, credit_no, bp_id, invoice_id, credit_date, currency, status, total}]}
POST /writeoffs <- {invoice_id, amount, reason} (AR-09's threshold applies there)
GET /migration/open-balances?db=<odoo db> -> {"balances": [{bp_id, currency, open_balance}]} (ar's reconcile figure)
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("ar", app)`.
"""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.seam import Seam, derived_id
AR = Seam("ar", "AR_BASE_URL", roles=["service"], permissions=["ar:invoice:read", "ar:writeoff:write", "ar:migration:read",
"ar:invoice:settle_discount", "ar:payment_request:settle"])
TAX = Seam("tax", "TAX_BASE_URL", roles=["service"], permissions=["tax:vat_return:read", "tax:document_tax:compute", "tax:cis:record_suffered"])
PARTNERS = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=[])
LEDGER = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post", "gl:company:read", "gl:report:read"])
PURCHASING = Seam("purchasing", "PURCHASING_BASE_URL", roles=["service"], permissions=["purchasing:bill:read", "purchasing:payment:read", "purchasing:payment:confirm"])
EXPENSES = Seam("expenses", "EXPENSES_BASE_URL", roles=["service"], permissions=["exp:report:mark_paid"])
HR = Seam("hr", "HR_BASE_URL", roles=["service"], permissions=["hr:bank_account:read_for_payout"])
UNSETTLEABLE = ("draft", "void", "voided", "cancelled", "canceled")
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
@dataclass(frozen=True)
class InvoiceDoc:
invoice_id: str
number: str
bp_id: int
currency: str
total: Decimal
issued_on: Optional[date]
due_date: Optional[date]
state: str = "issued"
discount_pct: Optional[Decimal] = None # v4: frozen from the terms at issue (2/10 net 30: 2 until the 10th day)
discount_until: Optional[date] = None
fx_rate: Optional[Decimal] = None # v4 (PAY-33): the rate FROZEN at issue, functional currency per unit of the invoice's currency
@dataclass(frozen=True)
class CreditNoteDoc:
credit_note_id: str
number: str
bp_id: int
currency: str
total: Decimal
issued_on: Optional[date]
invoice_id: str = ""
def _d(v) -> Optional[date]:
return date.fromisoformat(str(v)[:10]) if v else None
def _ask(path: str, params: dict, caller: Optional[dict], actor: str = "") -> dict:
if not AR.installed:
raise Unavailable(409, "invoicing (ar) is not part of this installation, so there is nothing to settle against. "
"Payments without invoicing records cash only")
r = AR.get(path, params={k: v for k, v in params.items() if v not in (None, "")}, actor=actor, caller=caller)
if r.ok:
return r.data
if r.status and r.status >= 500:
raise Unavailable(502, f"invoicing failed ({r.status}: {r.reason or 'no reason given'}), so its documents cannot be read - try again")
if r.status:
raise Unavailable(502, f"invoicing answered {r.status} ({r.reason or 'no reason given'}) when asked for its documents")
raise Unavailable(503, f"invoicing did not answer ({r.reason or 'no reason given'}), so its documents cannot be read - try again")
def _rows(data: dict, key: str) -> list:
got = data.get(key, data.get("body", []))
return got if isinstance(got, list) else []
def _invoice_doc(iid: str, r: dict) -> InvoiceDoc:
return InvoiceDoc(str(iid), str(r.get("invoice_no") or iid), int(r["bp_id"]), str(r["currency"]).upper(), Decimal(str(r["total"])),
_d(r.get("issue_date")), _d(r.get("due_date")), str(r.get("lifecycle") or "issued"),
Decimal(str(r["settlement_discount_pct"])) if r.get("settlement_discount_pct") not in (None, "") else None,
_d(r.get("settlement_discount_until")),
Decimal(str(r["fx_rate"])) if r.get("fx_rate") not in (None, "") else None)
def invoices(*, bp_id: Optional[int] = None, ids: Optional[list] = None, caller: Optional[dict] = None) -> list:
"""The settleable invoices of a customer, these ids, or (neither) every issued one (a draft, void or cancelled one cannot be
settled): ar's invoice_totals, 200 ids a call, or a page of 1000 at a time."""
rows: dict = {}
if ids:
for i in range(0, len(ids), 200):
rows |= _ask("/invoice-totals", {"invoice_ids": ",".join(ids[i:i + 200])}, caller).get("invoices") or {}
else:
offset = 0
while True:
page = _ask("/invoice-totals", {"bp_id": bp_id, "limit": 1000, "offset": offset}, caller)
rows |= page.get("invoices") or {}
if not page.get("more"):
break
offset += 1000
out = [_invoice_doc(iid, r) for iid, r in rows.items() if str(r.get("lifecycle", "issued")).lower() not in UNSETTLEABLE]
if bp_id is not None:
out = [d for d in out if d.bp_id == int(bp_id)]
return out
def invoice(invoice_id: str, caller: Optional[dict] = None) -> Optional[InvoiceDoc]:
got = invoices(ids=[invoice_id], caller=caller)
return got[0] if got else None
def credit_notes(*, bp_id: Optional[int] = None, caller: Optional[dict] = None) -> list:
"""The APPROVED credit notes of a customer (or every one), from ar's list_credit_notes, a page of 1000 at a time."""
out, offset = [], 0
while True:
page = _rows(_ask("/credit-notes", {"bp_id": bp_id, "status": "approved", "limit": 1000, "offset": offset}, caller), "credit_notes")
out += [CreditNoteDoc(str(r["credit_id"]), str(r.get("credit_no") or r["credit_id"]), int(r["bp_id"]),
str(r["currency"]).upper(), Decimal(str(r["total"])), _d(r.get("credit_date")),
str(r.get("invoice_id") or "")) for r in page if r.get("status", "approved") == "approved"]
if len(page) < 1000:
return out
offset += 1000
def ar_open_balances(db: str, caller: Optional[dict] = None) -> dict:
"""ar's own reconciled open balance per customer and currency for the Odoo migration (its reconcile figure):
GET /migration/open-balances?db=<db> -> {"balances": [{bp_id, currency, open_balance}]}. Raises Unavailable."""
data = _ask("/migration/open-balances", {"db": db}, caller, "odoo-migration")
return {(int(r["bp_id"]), str(r["currency"]).upper()): Decimal(str(r["open_balance"])) for r in _rows(data, "balances")}
def propose_writeoff(invoice_id: str, amount: Decimal, currency: str, reason: str, source: str, actor: str) -> dict:
"""Ask invoicing to consider writing off a small difference (AR-09): ar's propose_writeoff, POST /writeoffs. Returns
{proposed, reason}."""
if not AR.installed:
return {"proposed": False, "reason": "invoicing (ar) is not part of this installation"}
r = AR.post("/writeoffs", {"invoice_id": invoice_id, "amount": str(amount), "reason": f"{reason} ({source})"[:500]}, actor=actor)
return {"proposed": True, "writeoff_id": r.data.get("writeoff_id")} if r.ok else {"proposed": False, **r.problem()}
# ── business-partner ────────────────────────────────────────────────────────────────────────────────────
@dataclass(frozen=True)
class Checked:
status: str # known | unknown | unchecked
note: str = ""
http: int = 200
def check_customer(bp_id: int, caller: dict) -> Checked:
"""Does the partner register have this customer, as the caller sees it? Asked AS THE CALLER."""
if not PARTNERS.installed:
return Checked("unchecked", "the partner register (business-partner) is not part of this installation, so the "
"customer cannot be confirmed", 409)
r = PARTNERS.get(f"/partners/{int(bp_id)}", actor=caller.get("email", ""), caller=caller)
if r.status == 404:
return Checked("unknown", f"there is no business partner {bp_id} - add the customer to the partner register first", 422)
if r.ok:
return Checked("known")
why = r.reason or "no reason given"
if r.status in (401, 403):
return Checked("unchecked", f"the partner register refused the question ({r.status}: {why}): you may not look that "
"customer up", 403)
if r.status and r.status >= 500:
return Checked("unchecked", f"the partner register failed ({r.status}: {why}), so the customer cannot be checked - try again", 502)
if r.status:
return Checked("unchecked", f"the partner register answered {r.status} ({why})", 502)
return Checked("unchecked", f"the partner register did not answer ({why}), so the customer cannot be checked - try again", 503)
# ── the ledger ──────────────────────────────────────────────────────────────────────────────────────────
def account_line(gl_account: str) -> dict:
"""A journal line naming the bank account's ledger account: account_id when it is a number, else a purpose."""
v = str(gl_account).strip()
return {"account_id": v} if v.isdigit() else {"purpose": v}
def receivable_line(bp_id) -> dict:
"""The customer's side: the ledger's receivable purpose, naming the customer when there is one."""
return {"purpose": "receivable", **({"bp_id": str(bp_id)} if bp_id is not None else {})}
def post_journal(source_document: str, event: str, on: date, currency: str, memo: str, lines: list, actor: str,
company_code: str = "", fx_rate: Optional[Decimal] = None) -> dict:
"""Post a balanced journal in gl's GL-30 shape for payments: {journal_id, source_module: "payments", source_document, event
(receipt | receipt_reversal | refund | bank_line | bank_line_reversal | ...), company_code, journal_date, currency, memo,
lines[{account_id | purpose, tax_code?, debit?, credit?, bp_id?}]}. journal_id is a BIGINT DERIVED from
(source_document, event) - the payment or refund id and what happened - so posting twice changes nothing (gl answers
200 replay). Optional: no ledger installed is a reply ('not posted'), never an error - the business still works."""
if not LEDGER.installed:
return {"posted": False, "reason": "no ledger is installed, so nothing was posted"}
if not company_code:
return {"posted": False, "reason": "the bank account has no company_code, so the ledger cannot be told which company to post in"}
journal_id = str(derived_id(source_document, event))
r = LEDGER.post("/journal-entry/journals", {"journal_id": journal_id, "source_module": "payments", "source_document": source_document,
"event": event, "company_code": company_code, "journal_date": on.isoformat(), "currency": currency,
"memo": memo, "lines": lines, **({"fx_rate": str(fx_rate)} if fx_rate else {})}, actor=actor)
return {"posted": True, "journal_id": journal_id} if r.ok else {"posted": False, **r.problem()}
# ── invoicing: the settlement-discount credit (PAY-16, ar AR-38) ────────────────────────────────────────
def _body(text: str) -> dict:
import json
try:
got = json.loads(text)
return got if isinstance(got, dict) else {}
except (ValueError, TypeError):
return {}
def settlement_discount(invoice_id: str, payment_id: str, received_on: date, discount_amount: Decimal, actor: str) -> dict:
"""Ask invoicing to credit the early-settlement discount taken: ar's POST /invoices/{invoice_id}/settlement-discount
{payment_id, received_on, discount_amount}. ar raises the credit at the invoice's own rates and allocates it through
POST /allocations/from-source, so the invoice is paid; the same payment_id again answers 200 with the same credit.
Returns {ok, credit} or {ok: False, gate, reason, fix} (422 within_discount_window | discount_matches_terms) or raises Unavailable."""
if not AR.installed:
raise Unavailable(409, "invoicing (ar) is not part of this installation, so no discount credit can be raised")
r = AR.post(f"/invoices/{invoice_id}/settlement-discount",
{"payment_id": payment_id, "received_on": received_on.isoformat(), "discount_amount": str(discount_amount)}, actor=actor)
if r.ok:
return {"ok": True, "credit": r.data}
if r.status == 422:
body = _body(r.reason)
return {"ok": False, "gate": body.get("gate", ""), "reason": body.get("reason") or r.reason, "fix": body.get("fix", "")}
if r.status and r.status >= 500:
raise Unavailable(502, f"invoicing failed ({r.status}: {r.reason or 'no reason given'}), so no discount credit was raised - try again")
if r.status:
raise Unavailable(502, f"invoicing answered {r.status} ({r.reason or 'no reason given'}) to the discount credit")
raise Unavailable(503, f"invoicing did not answer ({r.reason or 'no reason given'}), so no discount credit was raised - try again")
# ── tax ──────────────────────────────────────────────────────────────────────────────────────────────────
def vat_liabilities(actor: str = "") -> dict:
"""tax's GET /vat-returns/liabilities (TAX-32): {liabilities: [{period_key, charge_type, outstanding_amount, due_date}], cis_due: [...]}."""
if not TAX.installed:
raise Unavailable(409, "tax is not part of this installation, so an HMRC payment cannot be matched to a liability")
r = TAX.get("/vat-returns/liabilities", actor=actor)
if not r.ok:
raise Unavailable(502 if r.status else 503, f"tax did not give its liabilities ({r.status or 'no answer'}: {r.reason or 'no reason given'}) - try again")
return r.data
def tax_compute(document_ref: str, tax_point: date, currency: str, treatment: str, gross: Decimal, description: str, actor: str) -> dict:
"""tax's POST /documents/compute with direction purchase (a statement line is money paid OUT, prices include tax):
returns the single line's {vat, vat_reclaimable, vat_non_reclaimable}. Raises Unavailable when tax is absent or silent."""
if not TAX.installed:
raise Unavailable(409, "tax is not part of this installation, so the VAT on this payment cannot be computed")
r = TAX.post("/documents/compute", {"document_ref": document_ref, "direction": "purchase", "tax_point": tax_point.isoformat(),
"bp_id": "0", "currency": currency, "treatment_code": treatment, "prices_include_tax": True,
"lines": [{"line_no": 1, "description": description[:200], "supply": "services",
"amount": str(gross), "quantity": "1", "unit_price": str(gross)}]}, actor=actor)
if not r.ok:
raise Unavailable(502 if r.status else 503, f"tax did not compute the VAT ({r.status or 'no answer'}: {r.reason or 'no reason given'})")
line = (r.data.get("lines") or [{}])[0]
return {"vat": Decimal(str(line.get("vat", "0"))), "vat_reclaimable": Decimal(str(line.get("vat_reclaimable", line.get("vat", "0")))),
"vat_non_reclaimable": Decimal(str(line.get("vat_non_reclaimable", "0")))}
# ── the ledger's company register (PAY-27) and purchasing's open bills (PAY-29) ──────────────────────────
def company(company_code: str, actor: str = "") -> Optional[dict]:
"""gl's GET /companies/{company_code} (GL-61): {company_code, name, country, functional_currency, ...}; None when gl says 404 (unknown or
not visible). No ledger installed returns {} (the company cannot be checked). Raises Unavailable when gl fails or is silent."""
if not LEDGER.installed:
return {}
r = LEDGER.get(f"/companies/{company_code}", actor=actor)
if r.ok:
return r.data
if r.status == 404:
return None
raise Unavailable(502 if r.status else 503, f"the ledger could not confirm company {company_code} ({r.status or 'no answer'}: "
f"{r.reason or 'no reason given'}) - try again")
def open_bills(as_of: date, actor: str = "") -> list:
"""purchasing's GET /bills/open?as_of= (PUR-14): [{bill_id, bill_no, bp_id, currency, total, outstanding, due_date}]. Raises Unavailable
(409 not installed, 502 failed, 503 silent): the forecast then says its payables side is missing."""
if not PURCHASING.installed:
raise Unavailable(409, "purchasing is not part of this installation, so supplier bills are not in the forecast")
r = PURCHASING.get("/bills/open", params={"as_of": as_of.isoformat()}, actor=actor)
if not r.ok:
raise Unavailable(502 if r.status else 503, f"purchasing did not give its open bills ({r.status or 'no answer'}: {r.reason or 'no reason given'})")
return [i for i in (r.data.get("items") or []) if isinstance(i, dict)]
# ── v4 slice C: the book balance (PAY-34), payment requests (PAY-36), CIS (PAY-45), supplier payments (PAY-48) ──────────
def _refusal(r) -> dict:
"""A neighbour's 4xx as {gate, reason, fix}: the body when it is JSON, else the words it gave."""
body = _body(r.reason)
text = body.get("reason") or body.get("detail") or r.reason or "no reason given"
return {"gate": body.get("gate", ""), "reason": str(text), "fix": body.get("fix", "")}
def _down(who: str, r, doing: str) -> Unavailable:
if r.status and r.status >= 500:
return Unavailable(502, f"{who} failed ({r.status}: {r.reason or 'no reason given'}), so {doing} - try again")
if r.status:
return Unavailable(502, f"{who} answered {r.status} ({r.reason or 'no reason given'}) when {doing}")
return Unavailable(503, f"{who} did not answer ({r.reason or 'no reason given'}), so {doing} - try again")
def account_balance(account_id: str, as_of: date, company_code: str = "", actor: str = "") -> Optional[dict]:
"""gl's GET /accounts/{account_id}/balance?as_of= (GL-61): {account_id, company_code, as_of, balance, currency}; None on 404 (unknown or
not visible). Raises Unavailable (409 no ledger installed, 502 failed, 503 silent): the book side of book_equals_bank cannot be proven."""
if not LEDGER.installed:
raise Unavailable(409, "no ledger is installed, so the book balance of the bank account cannot be compared with the statement")
r = LEDGER.get(f"/accounts/{account_id}/balance", params={"as_of": as_of.isoformat(), **({"company_code": company_code} if company_code else {})},
actor=actor)
if r.ok:
return r.data
if r.status == 404:
return None
raise _down("the ledger", r, f"the balance of account {account_id} could not be read")
def payment_request(request_id: str, actor: str) -> dict:
"""ar's GET /payment-requests/{request_id} (read-only): {request_id, bp_id, currency, remaining, ...}. Returns {ok: True, **request} or
{ok: False, status, reason, fix} for ar's 4xx (404 unknown request); raises Unavailable. Read BEFORE settling, so a receipt that cannot
be allocated is refused before invoicing is told anything."""
if not AR.installed:
raise Unavailable(409, "invoicing (ar) is not part of this installation, so a payment request cannot be read")
r = AR.get(f"/payment-requests/{request_id}", params={}, actor=actor)
if r.ok:
return {"ok": True, **r.data}
if r.status and 400 <= r.status < 500:
return {"ok": False, "status": r.status, **_refusal(r)}
raise _down("invoicing", r, "the payment request could not be read")
def settle_payment_request(request_id: str, payment_id: str, received_on: date, amount: Decimal, currency: str, actor: str) -> dict:
"""ar's POST /payment-requests/{request_id}/settle {payment_id, received_on, amount, currency} (AR-36): 200 {request_id, deposit_invoice_id,
invoice_no, amount, remaining, status}; the same payment_id again answers 200 with the same invoice. Returns {ok: True, **answer}, or
{ok: False, gate, reason, fix} for ar's 4xx (422 within_request, 404 unknown request); raises Unavailable."""
if not AR.installed:
raise Unavailable(409, "invoicing (ar) is not part of this installation, so a payment request cannot be settled")
r = AR.post(f"/payment-requests/{request_id}/settle", {"payment_id": payment_id, "received_on": received_on.isoformat(),
"amount": str(amount), "currency": currency}, actor=actor)
if r.ok:
return {"ok": True, **r.data}
if r.status and 400 <= r.status < 500:
return {"ok": False, "status": r.status, **_refusal(r)}
raise _down("invoicing", r, "the payment request could not be settled")
def record_cis_suffered(contractor_bp_id: int, statement_ref: str, tax_month: date, invoice_id: str, payment_id: str, gross: Decimal,
materials: Decimal, deduction: Decimal, actor: str) -> dict:
"""tax's POST /cis/suffered (TAX-41): 201 {recorded: true}; 200 {replay: true} for the same statement; 422 {gate: within_cis_rate, reason, fix}.
Returns {ok: True, replay} or {ok: False, gate, reason, fix}; raises Unavailable."""
if not TAX.installed:
raise Unavailable(409, "tax is not part of this installation, so a CIS deduction cannot be recorded")
r = TAX.post("/cis/suffered", {"contractor_bp_id": str(contractor_bp_id), "statement_ref": statement_ref, "tax_month": tax_month.isoformat(),
"invoice_id": invoice_id, "payment_id": payment_id, "gross": str(gross), "materials": str(materials),
"deduction": str(deduction)}, actor=actor)
if r.ok:
return {"ok": True, "replay": bool(r.data.get("replay"))}
if r.status and 400 <= r.status < 500:
return {"ok": False, "status": r.status, **_refusal(r)}
raise _down("tax", r, "the CIS deduction could not be recorded")
def supplier_payments(bank_account_id: str, status: str, actor: str = "") -> list:
"""purchasing's GET /supplier-payments?bank_account_id=&status= (PUR-66): [{payment_id, run_id, bp_id, bank_account_id, amount, currency,
reference, status: sent | confirmed | rejected, paid_on, bills}]. Raises Unavailable."""
if not PURCHASING.installed:
raise Unavailable(409, "purchasing is not part of this installation, so a statement debit cannot be matched to a supplier payment")
r = PURCHASING.get("/supplier-payments", params={"bank_account_id": bank_account_id, "status": status}, actor=actor)
if not r.ok:
raise _down("purchasing", r, "its supplier payments could not be read")
return [p for p in (r.data.get("payments") or []) if isinstance(p, dict)]
def confirm_supplier_payment(payment_id: str, bank_ref: str, paid_on: date, actor: str) -> dict:
"""purchasing's POST /supplier-payments/{payment_id}/confirm {bank_ref, paid_on} (PUR-66): 200 {payment_id, status: confirmed, replay}; 409
{gate: payment_open}. Returns {ok: True, replay} or {ok: False, gate, reason, fix}; raises Unavailable."""
if not PURCHASING.installed:
raise Unavailable(409, "purchasing is not part of this installation, so the supplier payment cannot be confirmed")
r = PURCHASING.post(f"/supplier-payments/{payment_id}/confirm", {"bank_ref": bank_ref, "paid_on": paid_on.isoformat()}, actor=actor)
if r.ok:
return {"ok": True, "replay": bool(r.data.get("replay"))}
if r.status and 400 <= r.status < 500:
return {"ok": False, "status": r.status, **_refusal(r)}
raise _down("purchasing", r, "the supplier payment could not be confirmed")
# ── expenses and hr: the payout (PAY-51) ─────────────────────────────────────────────────────────────
def tell_expenses_paid(source_document: str, payout_ref: str, payment_id: str, paid_on: date, amount: Decimal) -> dict:
"""POST /reports/{report_id}/paid {payout_ref, payment_id, paid_on, amount} (expenses EXP-13): 200 the report, status paid; 200 again for a repeated
payout_ref; 409 {gate: amount_matches} for another amount. -> {ok} | {ok: False, reason, status}. The caller retries until ok."""
if not EXPENSES.installed:
return {"ok": False, "reason": "expenses is not part of this installation", "status": 0}
r = EXPENSES.post(f"/reports/{source_document}/paid", {"payout_ref": payout_ref, "payment_id": payment_id, "paid_on": paid_on.isoformat(),
"amount": f"{amount:.2f}"}, actor="payments")
return {"ok": True} if r.ok else {"ok": False, **r.problem()}
def payee_bank_account(person_id: str, on: date) -> Optional[dict]:
"""GET /persons/{person_id}/bank-account?on= (hr HR-59, payments' client only, audited): {account_holder, sort_code | null, account_number | null,
iban | null, bic | null, valid_from}; None when hr holds none (404) or is not part of the installation."""
if not HR.installed:
return None
r = HR.get(f"/persons/{person_id}/bank-account", params={"on": on.isoformat()}, actor="payments")
return r.data if r.ok else None