dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
src/brick_ar/infrastructure/seams.py
"""The neighbours this brick asks, behind one adapter each. Declared HERE and nowhere else, so the domain never knows how a
neighbour is reached. Each port is shaped from THE NEIGHBOUR'S spec contract (module.yaml seams), never from its tables, and
every answer is a reply, never an exception: a neighbour that is not installed, refuses, fails or does not answer is a
reason a person can read.
tax compute / check_contents / tax_point_filed tax computes every line and the per-rate summary; ar freezes it
payments settlement / allocate / credit_ok payments owns every allocation; ar never writes pay_ tables
gl post_journal / period_open the ledger's route, journal_id derived from the document
business-partner check_partner a customer is a bp_id from the shared record
sales-order order_lines the order lines an invoice line names (sales_order_id, so_line_no)
product get_product / mark_unit_used description, unit_code and tax category copied at line time
any brick migration_map (odoo db, model, id) -> natural key, written by that brick's migration
The OTHER direction (sales-order asks raise_draft / billed_per_source, payments reads invoice totals, sales reads the narrow
status route) is not here: those bricks call THIS brick's routes.
"""
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.seam import Seam, derived_id
from brick_ar.domain.tax import TaxedLine, TaxOutcome, TaxQuote, TaxRowIn
# Seam names are the composition's member names (module ids: tax, payments, gl, business-partner, sales-order, product): a seam reaches the neighbour
# registered under ITS name, so "payments" or "sales_order" would never find one.
TAX = Seam("tax", "TAX_BASE_URL", roles=["service"], permissions=["tax:document_tax:compute", "tax:code:read", "tax:tax_point:read", "tax:migration:read"])
# payments reads the invoices back FROM this brick as whoever asked it (its settlement and open items are row-scoped by the
# caller): asked by this seam, that is this seam - so it carries ar:invoice:read too, or invoicing would refuse itself.
PAYMENTS = Seam("payments", "PAYMENTS_BASE_URL", roles=["service"], permissions=["pay:allocation:from_source", "pay:receipt:read",
"pay:credit:read", "ar:invoice:read"])
GL = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post", "gl:period:read", "gl:company:read"])
PURCHASING = Seam("purchasing", "PURCHASING_BASE_URL", roles=["service"], permissions=["purchasing:bill:write"])
BP = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=["bp:migration:read"])
SALES_ORDER = Seam("sales-order", "SALES_ORDER_BASE_URL", roles=["service"], permissions=["so:order:read", "so:sales_order:read_all"])
PRODUCT = Seam("product", "PRODUCT_BASE_URL", roles=["service"], permissions=["product:product:read", "product:unit:mark_used", "product:migration:read"])
MAPS = {"business-partner": BP, "tax": TAX, "product": PRODUCT, "sales-order": SALES_ORDER, "payments": PAYMENTS}
def journal_id(document_id: str, event: str) -> int:
"""A BIGINT derived from the document, so re-posting changes the ledger once (blake2b, 7 bytes: the spec's rule)."""
return derived_id(document_id, event)
def _d(v) -> Decimal:
return Decimal(str(v))
# ── tax ────────────────────────────────────────────────────────────────────────────────────────
def _tax_line(ln) -> dict:
"""One line as tax's compute_tax LineIn names it: {line_no, description, category_code, amount, quantity, unit_price}."""
out = {"line_no": ln.line_no, "amount": str(ln.net_amount)}
if ln.pinned_from: # a deposit deduction / a settlement-discount credit: the earlier document's frozen code and rate
out["pinned_from"], out["tax_code"] = ln.pinned_from, ln.tax_code
if ln.description:
out["description"] = ln.description[:500]
if ln.tax_category:
out["category_code"] = ln.tax_category
if ln.quantity is not None:
out["quantity"] = str(ln.quantity)
if ln.unit_price is not None and ln.unit_price >= 0: # tax's unit price is never negative (a deduction line has none)
out["unit_price"] = str(ln.unit_price)
return out
def tax_compute(lines: list, *, bp_id: int, tax_point: date, currency: str, document_ref: Optional[str] = None,
fx_rate=None) -> TaxOutcome:
"""tax's compute_tax: POST /documents/compute {lines, tax_point, bp_id, currency, document_ref} -> {complete, lines [{line_no,
tax_code, rate_pct, vat, refusal}], summary [{tax_code, rate_pct, net, vat}], rounding_method}. `document_ref` (the invoice or
credit note id) keeps the tax on the document in tax, so the content check and the freeze at issue can name it. A line no
rule covers is refused there (complete false) and comes back as the reason. Not installed / refused / down -> the reason,
and no quote: the document stays a draft with its tax pending."""
if not TAX.installed:
return TaxOutcome(reason="the tax module is not installed, so no tax can be computed")
body = {"direction": "sale", "bp_id": str(bp_id), "tax_point": tax_point.isoformat(), "currency": currency, "lines": [_tax_line(ln) for ln in lines]}
if document_ref:
body["document_ref"] = document_ref
if fx_rate:
body["fx_rate"] = str(fx_rate)
r = TAX.post("/documents/compute", body)
if not r.ok:
return TaxOutcome(reason=f"the tax module did not compute the tax ({r.status or 'unreachable'}: {r.reason or 'no reason given'})")
d = r.data
if not d.get("complete"):
why = "; ".join(f"line {x['line_no']}: {x['refusal']}" for x in d.get("lines", []) if x.get("refusal")) or \
"; ".join(g.get("reason", "") for g in d.get("gates", []) if not g.get("met")) or "the tax is not complete"
return TaxOutcome(reason=f"the tax module could not tax every line ({why})", gates=list(d.get("gates", [])))
try:
return TaxOutcome(quote=TaxQuote(
lines=tuple(TaxedLine(int(x["line_no"]), x["tax_code"], _d(x.get("rate_pct") or 0), _d(x["vat"])) for x in d["lines"]),
summary=tuple(TaxRowIn(x["tax_code"], _d(x.get("rate_pct") or 0), _d(x["net"]), _d(x["vat"])) for x in d["summary"]),
rounding_method=d.get("rounding_method", ""), tax_point=tax_point))
except (KeyError, TypeError, ValueError) as exc:
return TaxOutcome(reason=f"the tax module answered in a shape this brick cannot read ({exc})")
def _contents(bill_to_json: str, issue_date: Optional[date], invoice_number: str, supplier: dict) -> dict:
"""The invoice facts that are not the tax's own (tax's ContentsIn): the number, the dates, our name and address (ar's
supplier settings), the customer's bill-to name and address (the profile copied at issue)."""
import json
try:
bill = json.loads(bill_to_json or "{}")
except ValueError:
bill = {}
out = {"title": "VAT invoice", "invoice_number": invoice_number or None, "issue_date": issue_date.isoformat() if issue_date else None,
"supplier_name": supplier.get("name") or None, "supplier_address": supplier.get("address") or None,
"customer_name": bill.get("bill_to_name") or None, "customer_address": bill.get("bill_to_address") or None}
return {k: v for k, v in out.items() if v is not None}
def tax_check_contents(*, document_ref: str, bill_to_json: str, issue_date: date, supplier: dict, invoice_number: str = "") -> list:
"""TAX-12: tax's check_contents (POST /documents/{document_ref}/check-contents {contents}) -> every para 16.3.1 gate as
{gate, met, reason, fix}. Before issue the number is not yet claimed: invoicing's own number_series gate answers for it,
so tax's invoice_number gate is left out (and contents_complete judged without it). [] when tax is not installed."""
if not TAX.installed:
return []
r = TAX.post(f"/documents/{document_ref}/check-contents", {"contents": _contents(bill_to_json, issue_date, invoice_number, supplier)})
if not r.ok:
return [{"gate": "tax_content", "met": False, "reason": f"tax could not check the contents ({r.status or 'unreachable'}: "
f"{r.reason or 'no reason'})", "fix": "try again, or ask the tax administrator"}]
gates = [g for g in r.data.get("gates", []) if invoice_number or g.get("gate") != "invoice_number"]
if not invoice_number:
rest = [g for g in gates if g.get("gate") != "contents_complete"]
gates = rest + [g | {"met": all(x.get("met") for x in rest)} for g in gates if g.get("gate") == "contents_complete"]
return gates
def tax_freeze(*, document_ref: str, bill_to_json: str, issue_date: date, supplier: dict, invoice_number: str) -> tuple:
"""At issue: tax's freeze_document_tax (POST /documents/{document_ref}/freeze {contents}) - the tax the customer received,
forever; a later rate change changes nothing and recomputing is refused. (ok, reason). Freezing twice is a no-op there."""
if not TAX.installed:
return True, ""
r = TAX.post(f"/documents/{document_ref}/freeze", {"contents": _contents(bill_to_json, issue_date, invoice_number, supplier)})
return (True, "") if r.ok else (False, f"tax refused to freeze the document's tax ({r.status or 'unreachable'}: {r.reason or 'no reason'})")
@dataclass
class TaxPoint:
"""tax's answer for a tax point (TAX-33). `known` is False when tax did not answer 200: the gate is then met false, so nothing
issues into a period nobody can vouch for."""
known: bool
filed: bool = False
status: str = ""
period_key: str = ""
gate_met: bool = False
reason: str = ""
fix: str = ""
def tax_point_status(tax_point: date) -> Optional[TaxPoint]:
"""GET /vat-returns/tax-point-status?date= -> {date, period_key | null, status: none | open | draft | locked | submitted, filed,
gate: {gate: tax_point_open, met, reason, fix}}. None when tax is not installed (nothing to ask)."""
if not TAX.installed:
return None
r = TAX.get("/vat-returns/tax-point-status", params={"date": tax_point.isoformat()})
if not r.ok:
return TaxPoint(False, reason=f"tax did not say whether {tax_point} is in a filed VAT return ({r.status or 'unreachable'})",
fix="try again, or ask the tax administrator")
g = r.data.get("gate") or {}
return TaxPoint(True, bool(r.data.get("filed")), r.data.get("status", ""), r.data.get("period_key") or "", bool(g.get("met")),
g.get("reason", ""), g.get("fix", ""))
def tax_point_filed(tax_point: date) -> Optional[bool]:
"""True when the tax point is in a filed VAT return, False when it is not, None when tax cannot say (or is not installed)."""
s = tax_point_status(tax_point)
return s.filed if s is not None and s.known else None
def party_treatment(bp_id: int) -> Optional[str]:
"""tax's treatment code for a party (TAX-23: GET /parties/{bp_id} -> {current: {treatment_code} | null}); None when unknown."""
if not TAX.installed:
return None
r = TAX.get(f"/parties/{int(bp_id)}")
cur = (r.data or {}).get("current") if r.ok else None
return (cur or {}).get("treatment_code")
def late_allocations(invoice_ids: list) -> Optional[list]:
"""payments' PAY-46: GET /allocations?invoice_ids= -> the allocations with their received_on (the money, not the allocation)."""
if not PAYMENTS.installed:
return None
r = PAYMENTS.get("/allocations", params={"invoice_ids": ",".join(invoice_ids)})
return list(r.data.get("allocations", [])) if r.ok else None
def company_known(company_code: str) -> Optional[bool]:
"""gl's GET /companies/{company_code}: is the partner company in this installation? None when gl is not installed."""
if not GL.installed:
return None
return GL.get(f"/companies/{company_code}").ok
def hand_bill(body: dict, actor: str) -> "Allocated":
"""purchasing's POST /bills/intercompany (PUR-64): one supplier bill keyed "ic:<invoice_id>"; the same bill_id answers 200."""
if not PURCHASING.installed:
return Allocated(False, "purchasing is not installed, so the partner company's bill cannot be recorded")
r = PURCHASING.post("/bills/intercompany", body, actor=actor)
return Allocated(r.ok, "" if r.ok else f"purchasing refused the bill ({r.status or 'unreachable'}: {r.reason or 'no reason'})")
# ── payments ───────────────────────────────────────────────────────────────────────────────────
def aging(as_at: date, company_code: str = "") -> tuple:
"""payments' PAY-40 aging AS AT a date: GET /collections/aging?as_at= -> {as_at, buckets: [{bucket, invoices: [{invoice_id, bp_id,
outstanding, days_past_due, disputed}]}]}. (answer, reason): the answer is None when payments cannot say."""
if not PAYMENTS.installed:
return None, "payments is not installed, so the aging is not known"
params = {"as_at": as_at.isoformat(), **({"company_code": company_code} if company_code else {})}
r = PAYMENTS.get("/collections/aging", params=params)
if not r.ok:
return None, f"payments did not answer the aging ({r.status or 'unreachable'}: {r.reason or 'no reason given'})"
return r.data, ""
@dataclass
class EinvoiceSent:
ok: bool
message_id: str = ""
reason: str = ""
retryable: bool = False
def einvoice_send(provider: str, key: str, *, document_ref: str, document_type: str, receiver_id: str, xml: str, actor: str) -> EinvoiceSent:
"""Hand a validated UBL to the access-point provider composed beside the brick (capability convention: POST
/capabilities/einvoice.send {idempotency_key, dry_run, input}, permission <provider>:capability:invoke; the provider claims the key,
so a retry sends once). Its answer is the RESULT shape: done with the message id in output.message_id (or provider_ref). A provider
that is not wired, is down or answers connection_required is a reason a person can read, never an exception."""
from erp_platform.capabilities import base_env
seam = Seam(provider, base_env(provider), roles=["service"], permissions=[f"{provider}:capability:invoke"])
if not seam.installed:
return EinvoiceSent(False, reason=f"the access point {provider} is not reachable: {base_env(provider)} is not set", retryable=True)
r = seam.post("/capabilities/einvoice.send", {"idempotency_key": key, "dry_run": False, "input": {
"format": "ubl", "document_type": document_type, "document_ref": document_ref, "receiver_id": receiver_id, "xml": xml}}, actor=actor)
if not r.ok:
return EinvoiceSent(False, reason=f"the access point refused ({r.status or 'unreachable'}: {r.reason or 'no reason'})",
retryable=not r.status or r.status >= 500)
d = r.data or {}
if d.get("status") != "done":
return EinvoiceSent(False, reason=f"the access point answered {d.get('status')}: {d.get('message') or 'no message'}",
retryable=d.get("status") in ("in_progress", "connection_required") or bool(d.get("retryable")))
mid = (d.get("output") or {}).get("message_id") or d.get("provider_ref") or ""
return EinvoiceSent(bool(mid), mid, "" if mid else "the access point did not return a message id")
@dataclass
class Settlement:
outstanding: Optional[Decimal]
allocated: Decimal = Decimal(0)
state: str = ""
age_days: Optional[int] = None
@dataclass
class SettlementReply:
by_invoice: dict
reason: str = "" # why there is no answer (payments not installed, down)
@property
def known(self) -> bool:
return not self.reason
def settlement(invoice_ids: list) -> SettlementReply:
"""payments' settlement (PAY-40): GET /settlement?invoice_ids= -> {invoices: [{invoice_id, total, outstanding, state, age_days}],
unknown: [...]}. invoicing stores no paid flag."""
if not invoice_ids:
return SettlementReply({})
if not PAYMENTS.installed:
return SettlementReply({}, "payments is not installed, so what is outstanding is not known")
out = {}
for i in range(0, len(invoice_ids), 200): # payments reads at most 200 ids a call
r = PAYMENTS.get("/settlement", params={"invoice_ids": ",".join(invoice_ids[i:i + 200])})
if not r.ok:
return SettlementReply({}, f"payments did not answer ({r.status or 'unreachable'}: {r.reason or 'no reason given'})")
for v in r.data.get("invoices") or []:
outstanding = None if v.get("outstanding") is None else _d(v["outstanding"])
allocated = _d(v["total"]) - outstanding if outstanding is not None and v.get("total") is not None else Decimal(0)
out[str(v["invoice_id"])] = Settlement(outstanding, allocated, v.get("state", ""), v.get("age_days"))
return SettlementReply(out)
def credit_unallocated(credits: list) -> dict:
"""credit note id -> what is still unallocated on it, from payments' open items (GET /customers/{bp_id}/open-items: a credit
note with something left is listed under credits as a negative amount; one fully allocated is not listed, so 0). `credits`
is [(credit_id, bp_id)]. {} when payments cannot say."""
if not credits or not PAYMENTS.installed:
return {}
out = {}
for bp_id in sorted({int(b) for _c, b in credits}):
r = PAYMENTS.get(f"/customers/{bp_id}/open-items")
if not r.ok:
return {}
left = {str(c["id"]): -_d(c["amount"]) for c in r.data.get("credits") or [] if c.get("kind") == "credit_note"}
for cid, b in credits:
if int(b) == bp_id:
out[cid] = left.get(cid, Decimal("0.00"))
return out
@dataclass
class Allocated:
ok: bool
reason: str = ""
def allocate(source_kind: str, source_id: str, invoice_id: str, amount: Decimal, on: date, actor: str) -> Allocated:
"""payments records an allocation (source_kind credit_note | writeoff); calling twice with the same (source, invoice) adds
nothing. invoicing never writes pay_ tables."""
if not PAYMENTS.installed:
return Allocated(False, "payments is not installed, so the balance cannot be reduced")
r = PAYMENTS.post("/allocations/from-source", {"source_kind": source_kind, "source_id": source_id, "invoice_id": invoice_id,
"amount": str(amount)}, actor=actor) # payments dates it today itself
return Allocated(r.ok, "" if r.ok else f"payments refused the allocation ({r.status or 'unreachable'}: {r.reason or 'no reason'})")
@dataclass
class CreditStatus:
known: bool
hold: bool = False
reason: str = ""
balance: Optional[Decimal] = None
def credit_status(bp_id: int) -> CreditStatus:
"""payments' PAY-40 readiness gate GET /customers/{bp_id}/credit-ok -> {gate: credit_ok, met, reason, fix, hold_mode, bp_id}:
a hold is the gate not met. (The spec lists no balance on it, so none is read.)"""
if not PAYMENTS.installed:
return CreditStatus(False, reason="payments is not installed")
r = PAYMENTS.get(f"/customers/{int(bp_id)}/credit-ok")
if not r.ok:
return CreditStatus(False, reason=f"payments did not answer ({r.status or 'unreachable'})")
return CreditStatus(True, not r.data.get("met", True), r.data.get("reason", ""))
# ── gl ─────────────────────────────────────────────────────────────────────────────────────────
@dataclass
class Posted:
ok: bool
reason: str = ""
def gl_installed() -> bool:
return GL.installed
def post_journal(body: dict, actor: str) -> Posted:
"""POST /journal-entry/journals with a balanced journal (gl:journal:post). The ids are derived by invoicing, so posting twice
changes the ledger once. A closed period is refused by the ledger and named in its reason."""
if not GL.installed:
return Posted(False, "the ledger (gl) is not installed")
r = GL.post("/journal-entry/journals", body, actor=actor)
return Posted(r.ok, "" if r.ok else f"the ledger refused the journal ({r.status or 'unreachable'}): {r.reason or 'no reason given'}")
def period_open(on: date) -> Optional[bool]:
if not GL.installed:
return None
r = GL.get("/periods/open", params={"date": on.isoformat()})
return bool(r.data.get("open")) if r.ok else None
# ── business-partner, sales-order, product ───────────────────────────────────────────────────────
@dataclass(frozen=True)
class Checked:
status: str # known | unknown | unchecked
note: str = ""
def check_partner(bp_id: int, caller: dict) -> Checked:
"""Does the partner register have this customer, as the caller sees it? invoicing reads partners and never writes them."""
if not BP.installed:
return Checked("unchecked", "the partner register (business-partner) is not part of this installation")
r = BP.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")
if r.ok:
return Checked("known")
return Checked("unchecked", f"the partner register did not answer ({r.status or 'unreachable'}: {r.reason or 'no reason'})")
def order_lines(sales_order_id: str) -> Optional[dict]:
"""(line_no) -> {unit_code, product_id} of an order's lines, from sales-order's get_sales_order (GET /sales-orders/{id},
asked as this seam: so:sales_order:read_all), or None when sales-order cannot say."""
if not SALES_ORDER.installed:
return None
r = SALES_ORDER.get(f"/sales-orders/{sales_order_id}")
if not r.ok:
return None
return {int(x["line_no"]): x for x in r.data.get("lines", [])}
def get_product(product_id: str) -> Optional[dict]:
if not PRODUCT.installed:
return None
r = PRODUCT.get(f"/products/{product_id}")
return r.data if r.ok else None
def mark_unit_used(product_id: str, unit_code: str, invoice_id: str) -> None:
"""A document used this unit: its conversion is frozen from now (best effort; a failure never undoes an issue)."""
if PRODUCT.installed and product_id and unit_code:
PRODUCT.post(f"/products/{product_id}/units/{unit_code}/used", {"document_ref": invoice_id})
def migration_map(brick: str, db: str, model: str, ids: list) -> Optional[dict]:
"""Odoo id -> natural key, read from the brick whose migration mapped it: GET /migration-map/{db}/{model}?ids= ->
[{odoo_model, odoo_id, ours}] (shared.yaml conventions.odoo_migration; an unmapped id is left out). None when that brick
cannot answer."""
seam = MAPS[brick]
if not seam.installed:
return None
out = {}
ids = [str(i) for i in ids]
for i in range(0, len(ids), 200):
r = seam.get(f"/migration-map/{db}/{model}", params={"ids": ",".join(ids[i:i + 200])})
if not r.ok:
return None
out |= {str(x["odoo_id"]): x["ours"] for x in r.data.get("body") or []}
return out