dodil/erp-brick-sales-order
DODIL ERP brick: sales order
src/brick_sales_order/infrastructure/seams.py
"""The neighbours this brick asks, each behind a small PORT shaped from THEIR spec's contract (module.yaml seams).
product find_price(product_id, quantity, unit_code, bp_id, currency, priced_on) -> price + priced_from (SO-02)
tax compute(lines, bp_id, tax_point, currency) -> per-line code and VAT, the total, the method (TAX-20)
payments credit_status(bp_id) -> hold, reason, mode warn|block, as the credit_ok gate (SO-22)
invoicing raise_draft / plan_billing / billed_per_source / cancel_drafts (ar: SO-20, SO-21)
partners the customer exists and can transact (business-partner)
Nothing here is imported by the domain. Every call is idempotent on a key the order derives from its own work, so a
retried request asks the neighbour the same question and gets the same answer. A neighbour that is not part of the
installation, refuses, fails or does not answer is a `Missing` with its real reason — never a guess.
Tests replace `ports.<name>` with in-memory fakes (tests/fakes.py); a composition leaves the real adapters, which
reach the neighbour in process (tier 1) or at its BASE_URL (tiers 2 and 3) as erp_platform.seam does.
"""
import json
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.seam import Seam
class Missing(Exception):
"""A neighbour could not answer. `reason` is its own (or 'not installed'); `status` is what it said, 0 if nothing."""
def __init__(self, what: str, reason: str, status: int = 0, installed: bool = True):
super().__init__(f"{what}: {reason}")
self.what, self.reason, self.status, self.installed = what, reason, status, installed
def _reason(raw: str) -> str:
try:
d = json.loads(raw)
return str(d.get("detail") or d.get("reason") or raw)[:300] if isinstance(d, dict) else str(raw)[:300]
except (ValueError, TypeError):
return str(raw)[:300]
def _ask(reply, what: str) -> dict:
if reply.ok:
return reply.data
raise Missing(what, _reason(reply.reason), reply.status, installed="not installed" not in reply.reason)
# ── what the order takes from the neighbours ────────────────────────────────────────────────────
@dataclass(frozen=True)
class PriceAnswer:
"""A normalised find_price answer: the price per `unit_code`, the unit's exact conversion, and where it came from."""
unit_price: Decimal
unit_code: str
numerator: int
denominator: int
priced_from: str
price_list_id: str
description: str
tax_category: str
invoicing_policy: str
billing_kind: str
billing_plan: str
pricing_quantity: int = 1 # unit_price is per this many of pricing_unit_code (PRD-20)
pricing_unit_code: str = ""
pricing_numerator: int = 1
pricing_denominator: int = 1
product_type_code: str = "" # PRD-31: what the item category is determined from
category_code: str = "" # the product's category: conditions are found by it
gross_weight: Optional[Decimal] = None # per ONE of the line's unit, from the measures row in force on priced_on (freight)
weight_unit_code: str = ""
measure_valid_from: Optional[date] = None
def price_answer(a: dict) -> PriceAnswer:
"""Read product's find_price answer. priced_from names the list, the rule and the row the answer used."""
line, conv = a.get("line") or {}, a.get("conversion") or {}
m = line.get("measures") or {}
pf = a["priced_from"] if isinstance(a.get("priced_from"), dict) else {} # PRD-20: {price_list_id, rule_seq, row}
bits = [str(pf.get("price_list_id") or "")]
if pf.get("rule_seq") is not None:
bits.append(f"rule {pf['rule_seq']}")
row = pf.get("row") if isinstance(pf.get("row"), dict) else {}
if row:
bits.append(f"row {a.get('unit_code', '')}@{row.get('valid_from', '')}")
return PriceAnswer(Decimal(str(a["unit_price"])), a["unit_code"], int(conv.get("numerator", 1)), int(conv.get("denominator", 1)),
" ".join(b for b in bits if b) or "find_price", str(pf.get("price_list_id") or ""),
line.get("description") or line.get("name") or "", line.get("tax_category") or "",
line.get("invoicing_policy") or "ordered", line.get("billing_kind") or "", line.get("billing_plan") or "",
int(a.get("pricing_quantity") or 1), str(a.get("pricing_unit_code") or ""),
int(a.get("pricing_numerator") or 1), int(a.get("pricing_denominator") or 1),
str(line.get("product_type_code") or ""), str(line.get("category_code") or ""),
Decimal(str(m["gross_weight"])) if m.get("gross_weight") not in (None, "") else None, str(m.get("weight_unit_code") or ""),
date.fromisoformat(m["valid_from"]) if m.get("valid_from") else None)
class ProductPort:
def find_price(self, caller: dict, *, product_id: str, quantity: Decimal, unit_code: Optional[str], bp_id: int,
currency: str, priced_on: date) -> dict:
raise NotImplementedError
def rounding(self, caller: dict, *, product_id: str, quantity: Decimal, unit_code: str, on: date) -> dict:
"""GET /products/{id}/rounding?role=delivery&quantity=&unit_code=&on= (PRD-32): {product_id, role, rule | null, quantity_in_rule_unit, met,
next_down, next_up, reason}. A product with no delivery rule answers rule null and met true."""
raise NotImplementedError
def convert(self, caller: dict, *, product_id: str, quantity: Decimal, unit_code: str) -> dict:
"""GET /products/{id}/convert (unit -> the same unit): {in_base_unit (rounded half up to the base unit's decimals),
base_unit_code}. The order compares it with the exact fraction: equal = exact, never rounded (SO-26)."""
raise NotImplementedError
def price_lines(self, caller: dict, *, bp_id: int, currency: str, priced_on: date, lines: list) -> list:
"""POST /prices/batch (product v7 PRD-41): lines [{line_no, product_id, quantity, unit_code, price, rounding_role}] ->
[{line_no, product_id, price, rounding, conversion}], each part the one-line route's answer, {status, reason} for what it
refused, or None when not asked. This default asks the one-line routes line by line: a fake, or a product module that
predates PRD-41. Missing (the whole call) only when product is not installed or does not answer."""
out = []
for ln in lines:
def part(fn):
try:
return fn()
except Missing as m:
if not m.installed:
raise
return {"status": m.status, "reason": m.reason}
pid, q, unit = ln["product_id"], Decimal(str(ln["quantity"])), ln.get("unit_code")
out.append({"line_no": ln["line_no"], "product_id": pid,
"price": part(lambda: self.find_price(caller, product_id=pid, quantity=q, unit_code=unit, bp_id=bp_id, currency=currency,
priced_on=priced_on)) if ln.get("price", True) else None,
"rounding": part(lambda: self.rounding(caller, product_id=pid, quantity=q, unit_code=unit, on=priced_on))
if ln.get("rounding_role") else None,
"conversion": None})
return out
class TaxPort:
def compute(self, caller: dict, *, lines: list, bp_id: int, tax_point: date, currency: str, document_ref: str = "", ship_to_country: str = "") -> dict:
"""document_ref so:{sales_order_id}@r{revision} (TAX-20). lines: [{ref, product_id, tax_category, net_amount}] -> {lines: [{ref, tax_code, rate, tax_amount}], tax_total, method}"""
raise NotImplementedError
class PaymentsPort:
def credit_status(self, caller: dict, bp_id: int) -> dict:
"""-> {hold: bool, reason: str, mode: 'warn'|'block', balance}"""
raise NotImplementedError
class InvoicingPort:
def raise_draft(self, *, source_id: str, bp_id: int, currency: str, lines: list, kind: str, customer_po_ref: str,
sold_to_bp_id: int = 0, ship_to_bp_id: int = 0, bill_to_bp_id: int = 0, deductions: Optional[list] = None,
issue_hint: str = "") -> dict:
"""bp_id is the PAYER (whose receivable it is); sold_to, ship_to and bill_to ride beside it (bill-to's billing profile addresses the
invoice). deductions: [{deposit_invoice_id, net_amount, tax_category}], what this invoice takes off each deposit; issue_hint cash: issued at
once when invoicing's owner control says so. -> {invoice_id}; keyed on (source_kind sales_order, source_id): the second call returns the same draft"""
raise NotImplementedError
def plan_billing(self, *, schedule_id: str, bp_id: int, currency: str, line: dict) -> dict:
"""one schedule per repeating line, keyed '{sales_order_id}|{line_no}'"""
raise NotImplementedError
def billed(self, sales_order_id: str) -> dict:
"""-> {line_no: quantity on ISSUED invoices}; invoicing's lines carry sales_order_id and so_line_no"""
raise NotImplementedError
def billed_page(self, sales_order_ids: list) -> dict:
"""-> {sales_order_id: {line_no: qty}} for a page of the order list, in ONE call"""
return {i: self.billed(i) for i in sales_order_ids}
def cancel_drafts(self, source_ids: list) -> dict:
raise NotImplementedError
def issued_sources(self, invoice_ids: list) -> dict:
"""-> {invoice_id: True when that invoice is ISSUED}, keyed by ar's own invoice_id (a deposit has no order line, so `billed` cannot show it)."""
raise NotImplementedError
def billed_lines(self, *, bp_id: int, issued_from: date, issued_to: date) -> list:
"""GET /billed-lines (AR-20): [{document_id, kind (invoice | credit_note), line_no, issue_date, product_id, category_code, net_amount,
tax_category, sales_order_id, so_line_no}]: the rebate basis, read here and never from ar_ tables."""
raise NotImplementedError
def credit_note(self, *, source_id: str, bp_id: int, currency: str, reason: str, lines: list, account_purpose: str = "", invoice_id: str = "") -> dict:
"""POST /credit-notes (AR-04): idempotent on source_id (so:{sales_order_id}:{request_no} | rebate:{agreement}:{settlement_no}); invoice_id the
original invoice, each line's invoice_line_no against it; account_purpose rebate posts against accrued rebates, not revenue.
-> {credit_id, status: draft, ...}; the same source_id answers the same credit note."""
raise NotImplementedError
class GlPort:
def post_journal(self, *, journal_id: str, source_document: str, event: str, company_code: str, journal_date: date, currency: str,
memo: str, lines: list) -> dict:
"""POST /journal-entry/journals (GL-30): the ledger owns journals, idempotent on journal_id (a repeat answers replay true). A refusal
(409 period_open | journal_conflict, 422 balanced | purposes_mapped | ledger_assigned | bounds) is a Missing with the ledger's words."""
raise NotImplementedError
class InventoryPort:
def reserve(self, *, sales_order_id: str, revision: int, lines: list, schedule_version: int = 0, priority: str = "") -> dict:
"""POST /reservations/sales-orders/{id} (INV): keyed (order, revision, schedule_version), a repeat changes nothing; a re-promise
re-sends the same revision with a higher schedule version (lines may carry warehouse_id and schedule [{confirmed_on, quantity_base}]) ->
{lines: [{line_no, reserved_base, short_base, base_unit_code, readiness, ignored (a service line)}]}"""
raise NotImplementedError
def atp(self, *, product_id: str, warehouse_id: str, quantity: str, unit_code: str, requested_on: str, min_shelf_life_days: int = 0,
exclude_sales_order_id: str = "") -> dict:
"""GET /atp/{product_id} (INV, SO-35): nothing is reserved ->
{product_id, base_unit_code, available_now, confirmations: [{available_on, quantity_base}], shortfall_base, sources, pick_pack_days}"""
raise NotImplementedError
def release(self, *, sales_order_id: str, reason: str) -> dict:
"""POST /reservations/sales-orders/{id}/release: releases what is held and not delivered; a repeat releases nothing more."""
raise NotImplementedError
def expect_return(self, *, rma_ref: str, bp_id: int, warehouse_id: str, lines: list) -> dict:
"""POST /expected-returns (INV-14): idempotent on rma_ref -> {rma_ref, expected: [{line_no, quantity_base}]}; inventory receives a customer
return only against an open RMA."""
raise NotImplementedError
def cancel_expected_return(self, *, rma_ref: str) -> dict:
"""POST /expected-returns/{rma_ref}/cancel: withdraws the RMA while nothing is received."""
raise NotImplementedError
def consignment_stock(self, *, bp_id: int, product_id: str) -> dict:
"""GET /consignment-stock/customers/{bp_id}?product_id= (INV-45) -> {bp_id, products: [{product_id, base_unit_code, on_hand_base, location_id}]}:
stock at the customer that is still ours; read here, never kept here."""
raise NotImplementedError
class PurchasingPort:
def purchase_request(self, *, request_ref: str, product_id: str, quantity: str, unit_code: str, needed_by: str, ship_to_bp_id: int,
ship_to_address: dict, sales_order_id: str, so_line_no: int) -> dict:
"""POST /purchase-requests/from-sales-order (PUR-65): idempotent on request_ref ->
201|200 {request_ref, request_id, status: awaiting_approval}; 422 {gate: purchasable, reason, fix} is a Missing with purchasing's words."""
raise NotImplementedError
def cancel_purchase_request(self, *, request_id: str, reason: str) -> dict:
"""POST /purchase-requests/{request_id}/cancel: a refusal (the supplier already confirmed) is a Missing with purchasing's reason."""
raise NotImplementedError
class PartnersPort:
def customer(self, caller: dict, bp_id: int) -> dict:
"""-> {can_transact: bool}"""
raise NotImplementedError
def partner(self, caller: dict, bp_id: int) -> dict:
"""GET /partners/{bp_id} (BP): {bp_id, status, roles: [customer, ship_to, ...], can_transact, ...}. A partner function is accepted only for an
active partner with the role it plays (customer for sold-to, bill-to and payer; ship_to or customer for a ship-to)."""
raise NotImplementedError
class OdooPort:
"""The one-off Odoo migration's READER (read-only). Rows arrive flat: a many2one is its id, a currency its code.
orders: id, name, company_id, state, locked, partner_id, client_order_ref, date_order (ISO), user_id, currency,
pricelist_name, payment_term_id, incoterm, amount_untaxed, amount_tax, amount_total.
lines: id, order_id, sequence, display_type, is_downpayment, product_id, name, product_uom_qty, product_uom, price_unit,
discount, tax_ids, price_subtotal, price_tax, qty_delivered, qty_invoiced, qty_to_invoice, and on a down-payment line
invoiced_amount, deducted_amount and invoice_move_ids (the posted moves it links to)."""
db: str = ""
def orders(self) -> list:
raise NotImplementedError
def lines(self, order_ids: list) -> list:
raise NotImplementedError
class MapsPort:
"""The earlier bricks' Odoo id maps (shared.yaml conventions.odoo_migration): partner -> bp_id, product -> product_id +
invoicing policy, uom -> unit_code + exact conversion, tax -> tax_code, payment term -> terms id. None = unmapped."""
db: str = "" # the Odoo database being migrated; the plan sets it from the reader
def partner(self, odoo_id: int) -> Optional[int]:
raise NotImplementedError
def product(self, odoo_id: int) -> Optional[dict]:
raise NotImplementedError
def unit(self, odoo_uom_id: int, product_id: str) -> Optional[dict]:
raise NotImplementedError
def tax(self, odoo_id: int) -> Optional[str]:
raise NotImplementedError
def payment_term(self, odoo_id: int) -> Optional[str]:
raise NotImplementedError
# ── the real adapters: each neighbour's REAL route and shape (read from its use case, the wiring pass 2026-10-22) ────────
PRODUCT = Seam("product", "PRODUCT_BASE_URL", roles=["service"], permissions=["product:product:read", "product:price_list:read"])
TAX = Seam("tax", "TAX_BASE_URL", roles=["service"], permissions=["tax:document_tax:compute", "tax:code:read"])
PAY = Seam("payments", "PAYMENTS_BASE_URL", roles=["service"], permissions=["pay:credit:read"])
PUR = Seam("purchasing", "PURCHASING_BASE_URL", roles=["service"], permissions=["purchasing:request:intake"])
AR = Seam("ar", "AR_BASE_URL", roles=["service"], permissions=["ar:invoice:raise", "ar:schedule:plan", "ar:invoice:read",
"ar:invoice:write", "ar:migration:read", "ar:credit_note:write"])
BP = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=[])
INV = Seam("inventory", "INVENTORY_BASE_URL", roles=["service"], permissions=["inv:reservation:intake", "inv:availability:read", "inv:return:expect"])
GL = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post"])
# product's billing vocabulary (one_off | recurring | metered; single | periodic | arrears) in invoicing's words
_AR_KIND = {"recurring": "recurring", "metered": "usage", "instalment": "instalment"}
_AR_PLAN = {"monthly": "monthly", "quarterly": "quarterly", "annual": "annual", "periodic": "periodic", "arrears": "periodic"}
class RealProduct(ProductPort):
def find_price(self, caller, *, product_id, quantity, unit_code, bp_id, currency, priced_on):
params = {"currency": currency, "quantity": str(quantity), "bp_id": str(bp_id), "priced_on": priced_on.isoformat()}
if unit_code:
params["unit_code"] = unit_code
return _ask(PRODUCT.get(f"/products/{product_id}/price", params=params, actor=caller.get("email", ""), caller=caller),
f"product {product_id}")
def rounding(self, caller, *, product_id, quantity, unit_code, on):
return _ask(PRODUCT.get(f"/products/{product_id}/rounding", params={"role": "delivery", "quantity": str(quantity), "unit_code": unit_code,
"on": on.isoformat()}, actor=caller.get("email", ""), caller=caller),
f"product {product_id}")
def price_lines(self, caller, *, bp_id, currency, priced_on, lines):
body = {"currency": currency, "priced_on": priced_on.isoformat(), "lines": [
{**ln, "quantity": str(ln["quantity"])} for ln in lines]}
if bp_id is not None:
body["bp_id"] = str(bp_id)
r = PRODUCT.post("/prices/batch", body, actor=caller.get("email", ""), caller=caller)
if not r.ok and r.status in (404, 405): # a product module that predates PRD-41: the one-line routes
return super().price_lines(caller, bp_id=bp_id, currency=currency, priced_on=priced_on, lines=lines)
return _ask(r, "product (price lines)")["lines"]
def convert(self, caller, *, product_id, quantity, unit_code):
return _ask(PRODUCT.get(f"/products/{product_id}/convert", params={"quantity": str(quantity), "from_unit": unit_code, "to_unit": unit_code},
actor=caller.get("email", ""), caller=caller), f"product {product_id}")
class RealTax(TaxPort):
"""tax's compute_tax: POST /documents/compute {lines: [{line_no, category_code, amount}], tax_point, bp_id, currency} ->
{complete, lines: [{line_no, tax_code, rate_pct, vat, refusal}], totals: {vat, method}}. A line no rule covers comes back
refused (complete false): that is a Missing with tax's own reason, never a 0% guess."""
def compute(self, caller, *, lines, bp_id, tax_point, currency, document_ref="", ship_to_country=""):
body = {"lines": [{"line_no": int(ln["ref"]), "category_code": ln.get("tax_category") or None, "amount": str(ln["net_amount"])}
for ln in lines],
"document_ref": document_ref, "direction": "sale", "tax_point": tax_point.isoformat(), "bp_id": str(bp_id), "currency": currency,
**({"ship_to_country": ship_to_country} if ship_to_country else {})}
d = _ask(TAX.post("/documents/compute", body, actor=caller.get("email", "")), "tax")
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"
raise Missing("tax", why, 422)
totals = d.get("totals") or {}
return {"lines": [{"ref": x["line_no"], "tax_code": x["tax_code"], "rate": x.get("rate_pct"), "tax_amount": x["vat"]}
for x in d.get("lines", [])],
"tax_total": totals.get("vat"), "method": totals.get("method") or d.get("rounding_method", "")}
class RealPayments(PaymentsPort):
def credit_status(self, caller, bp_id):
"""payments' credit_ok (PAY-39): GET /customers/{bp_id}/credit-ok -> {gate: credit_ok, met, reason, fix, hold_mode, bp_id}.
A hold is the gate not met. (The route this brick called before, credit-status, no longer exists.)"""
d = _ask(PAY.get(f"/customers/{int(bp_id)}/credit-ok", actor=caller.get("email", "")), "payments")
return {"hold": not d.get("met", True), "reason": d.get("reason") or "", "mode": d.get("hold_mode") or "block"}
def _ar_line(x: dict, sales_order_id: str) -> dict:
"""One order line as invoicing's LineIn names it (extra fields are refused there): line_no -> so_line_no; the net is
invoicing's to compute (quantity_priced x unit_price / pricing_quantity x (100 - discount_pct)/100); the tax code is tax's to choose."""
out = {k: x[k] for k in ("description", "quantity", "unit_price", "discount_pct", "unit_code", "product_id", "tax_category",
"pricing_quantity", "pricing_unit_code", "quantity_priced")
if x.get(k) not in (None, "")}
if x.get("billed_net"): # a conditioned line: its net is its conditions' sum, which invoicing cannot re-derive from a price
out["net_amount"] = x["billed_net"]
if x.get("line_no"):
out |= {"sales_order_id": sales_order_id, "so_line_no": int(x["line_no"])}
return out
def raise_body(*, source_id, bp_id, currency, lines, kind, customer_po_ref, sold_to_bp_id=0, ship_to_bp_id=0, bill_to_bp_id=0, deductions=None, issue_hint="") -> dict:
"""POST /invoices/raise exactly as invoicing's spec lists it (AR-20); the fakes build the same body, so a drift shows in the contract check."""
sid = source_id[3:source_id.rindex(":")]
body = {"source_kind": "sales_order", "source_id": source_id, "bp_id": str(bp_id), "currency": currency, "kind": kind,
"lines": [_ar_line(x, sid) for x in lines]}
if deductions:
body["deductions"] = [{"deposit_invoice_id": d["deposit_invoice_id"], "net_amount": str(d["net_amount"]), "tax_category": d["tax_category"]} for d in deductions]
if issue_hint:
body["issue_hint"] = issue_hint
if customer_po_ref:
body["customer_po_ref"] = customer_po_ref
return body | _parties(sold_to_bp_id, ship_to_bp_id, bill_to_bp_id)
def credit_body(*, source_id, bp_id, currency, reason, lines, account_purpose="", invoice_id="") -> dict:
body = {"source_id": source_id, "bp_id": str(bp_id), "currency": currency, "reason": reason, "lines": lines}
if invoice_id:
body["invoice_id"] = invoice_id
if account_purpose:
body["account_purpose"] = account_purpose
return body
def expected_return_body(*, rma_ref, bp_id, warehouse_id, lines) -> dict:
return {"rma_ref": rma_ref, "bp_id": int(bp_id), "warehouse_id": warehouse_id, "lines": lines}
def _parties(sold_to: int, ship_to: int, bill_to: int) -> dict:
"""The order's partner functions beside the payer (SO-40): only the ones named."""
return {k: str(v) for k, v in (("sold_to_bp_id", sold_to), ("ship_to_bp_id", ship_to), ("bill_to_bp_id", bill_to)) if v}
class RealInvoicing(InvoicingPort):
def raise_draft(self, *, source_id, bp_id, currency, lines, kind, customer_po_ref, sold_to_bp_id=0, ship_to_bp_id=0, bill_to_bp_id=0,
deductions=None, issue_hint=""):
"""ar's raise_draft: POST /invoices/raise (keyed on source_id). A deposit is deducted BY AMOUNT per deposit (`deductions`, each in the
deposit's tax category), never beyond what that deposit has left (409 deposit_remaining)."""
return _ask(AR.post("/invoices/raise", raise_body(source_id=source_id, bp_id=bp_id, currency=currency, lines=lines, kind=kind,
customer_po_ref=customer_po_ref, sold_to_bp_id=sold_to_bp_id, ship_to_bp_id=ship_to_bp_id,
bill_to_bp_id=bill_to_bp_id, deductions=deductions, issue_hint=issue_hint), actor="sales-order"), "invoicing")
def plan_billing(self, *, schedule_id, bp_id, currency, line):
"""ar's plan_billing: POST /schedules, keyed "<source_id>|<line_id>". The order line's quantity is its number of
periods (12 periods at 120.00 = quantity 12, unit price 120.00); each period bills one."""
source_id, line_id = schedule_id.rsplit("|", 1)
qty = Decimal(str(line.get("quantity") or 1))
if qty != qty.to_integral() or qty < 1:
raise Missing("invoicing", f"a repeating line bills whole periods: quantity {qty} is not a number of periods", 422)
body = {"source_kind": "sales_order", "source_id": source_id, "line_id": line_id, "bp_id": str(bp_id), "currency": currency,
"starts_on": line["starts_on"], "periods": int(qty), "unit_price": str(line["unit_price"]),
"billing_kind": _AR_KIND.get(line.get("billing_kind") or "", "recurring"),
"billing_plan": _AR_PLAN.get(line.get("billing_plan") or "", "monthly")}
if line.get("billing_plan") == "arrears":
body["in_advance"] = False
for k in ("discount_pct", "unit_code", "product_id", "description", "tax_category"):
if line.get(k) not in (None, ""):
body[k] = str(line[k])
return _ask(AR.post("/schedules", body, actor="sales-order"), "invoicing")
def billed(self, sales_order_id):
"""ar's billed_per_source: GET /sources/sales_order/{id}/billed -> {lines: [{so_line_no, quantity_billed, net_billed}]}."""
data = _ask(AR.get(f"/sources/sales_order/{sales_order_id}/billed", actor="sales-order"), "invoicing")
return {int(x["so_line_no"]): Decimal(str(x["quantity_billed"])) for x in data.get("lines") or []}
def billed_page(self, sales_order_ids):
"""ar serves billed per order, not per page: one read per order on the page."""
return {i: self.billed(i) for i in sales_order_ids}
def cancel_drafts(self, source_ids):
"""ar's cancel_draft, one draft at a time: POST /invoices/{invoice_id}/cancel (a draft is keyed on its source)."""
done = []
for sid in source_ids:
r = AR.post(f"/invoices/{sid}/cancel", {}, actor="sales-order")
if not r.ok and r.status != 404:
raise Missing("invoicing", _reason(r.reason), r.status, installed="not installed" not in r.reason)
done.append(sid)
return {"cancelled": done}
def issued_sources(self, invoice_ids):
"""ar's invoice_totals: GET /invoice-totals?invoice_ids= -> {invoices: {id: {lifecycle, ...}}}, keyed by ar's OWN invoice_id (what
raise_draft answered), not by our source id."""
data = _ask(AR.get("/invoice-totals", params={"invoice_ids": ",".join(invoice_ids)}, actor="sales-order"), "invoicing")
got = data.get("invoices") or {}
return {i: (got.get(i) or {}).get("lifecycle") == "issued" for i in invoice_ids}
def billed_lines(self, *, bp_id, issued_from, issued_to):
return _ask(AR.get("/billed-lines", params={"bp_id": str(bp_id), "issued_from": issued_from.isoformat(), "issued_to": issued_to.isoformat()},
actor="sales-order"), "invoicing")
def credit_note(self, *, source_id, bp_id, currency, reason, lines, account_purpose="", invoice_id=""):
body = credit_body(source_id=source_id, bp_id=bp_id, currency=currency, reason=reason, lines=lines, account_purpose=account_purpose, invoice_id=invoice_id)
return _ask(AR.post("/credit-notes", body, actor="sales-order"), "invoicing")
class RealGl(GlPort):
def post_journal(self, *, journal_id, source_document, event, company_code, journal_date, currency, memo, lines):
body = {"journal_id": journal_id, "source_module": "sales-order", "source_document": source_document, "event": event,
"journal_date": journal_date.isoformat(), "currency": currency, "memo": memo, "lines": lines,
"company_code": company_code} # empty: the ledger's default company
return _ask(GL.post("/journal-entry/journals", body, actor="sales-order"), "the ledger")
class RealInventory(InventoryPort):
def reserve(self, *, sales_order_id, revision, lines, schedule_version=0, priority=""):
body = {"revision": int(revision), "lines": lines}
if schedule_version or priority == "rush": # a RUSH order says so even before it has a promise
body |= {"priority": priority or "normal"}
if schedule_version:
body |= {"schedule_version": int(schedule_version)}
return _ask(INV.post(f"/reservations/sales-orders/{sales_order_id}", body, actor="sales-order"), "inventory")
def atp(self, *, product_id, warehouse_id, quantity, unit_code, requested_on, min_shelf_life_days=0, exclude_sales_order_id=""):
params = {"warehouse_id": warehouse_id, "quantity": str(quantity), "unit_code": unit_code, "requested_on": requested_on,
"min_shelf_life_days": int(min_shelf_life_days or 0)}
if exclude_sales_order_id:
params["exclude_sales_order_id"] = exclude_sales_order_id
return _ask(INV.get(f"/atp/{product_id}", params=params, actor="sales-order"), "inventory")
def release(self, *, sales_order_id, reason):
return _ask(INV.post(f"/reservations/sales-orders/{sales_order_id}/release", {"reason": reason}, actor="sales-order"), "inventory")
def expect_return(self, *, rma_ref, bp_id, warehouse_id, lines):
return _ask(INV.post("/expected-returns", expected_return_body(rma_ref=rma_ref, bp_id=bp_id, warehouse_id=warehouse_id, lines=lines),
actor="sales-order"), "inventory")
def cancel_expected_return(self, *, rma_ref):
return _ask(INV.post(f"/expected-returns/{rma_ref}/cancel", {}, actor="sales-order"), "inventory")
def consignment_stock(self, *, bp_id, product_id):
return _ask(INV.get(f"/consignment-stock/customers/{int(bp_id)}", params={"product_id": product_id}, actor="sales-order"), "inventory")
class RealPurchasing(PurchasingPort):
def purchase_request(self, *, request_ref, product_id, quantity, unit_code, needed_by, ship_to_bp_id, ship_to_address, sales_order_id, so_line_no):
body = {"request_ref": request_ref, "product_id": product_id, "quantity": quantity, "unit_code": unit_code, "needed_by": needed_by,
"ship_to_bp_id": int(ship_to_bp_id), "ship_to_address": ship_to_address, "sales_order_id": sales_order_id, "so_line_no": int(so_line_no)}
return _ask(PUR.post("/purchase-requests/from-sales-order", body, actor="sales-order"), "purchasing")
def cancel_purchase_request(self, *, request_id, reason):
return _ask(PUR.post(f"/purchase-requests/{request_id}/cancel", {"reason": reason}, actor="sales-order"), "purchasing")
class RealPartners(PartnersPort):
def customer(self, caller, bp_id):
return _ask(BP.get(f"/partners/{int(bp_id)}", actor=caller.get("email", ""), caller=caller), f"partner {bp_id}")
partner = customer
class RealMaps(MapsPort):
"""Each earlier brick answers GET /migration-map/{db}/{odoo model}?ids= -> [{odoo_model, odoo_id, ours}] from its
<prefix>_migration_map (an unmapped id is left out). `db` is the Odoo database being migrated (set from the reader)."""
def _ours(self, seam: Seam, model: str, odoo_id: int) -> Optional[str]:
r = seam.get(f"/migration-map/{self.db}/{model}", params={"ids": str(int(odoo_id))}, actor="sales-order")
if r.ok:
return next((x.get("ours") for x in r.data.get("body") or [] if str(x.get("odoo_id")) == str(int(odoo_id))), None)
raise Missing(f"{seam.name} migration map", _reason(r.reason), r.status, installed="not installed" not in r.reason)
def partner(self, odoo_id):
d = self._ours(BP, "res.partner", odoo_id)
return int(d) if d else None
def product(self, odoo_id):
"""product_id from product's map, and the invoicing policy copied from the product itself."""
pid = self._ours(PRODUCT, "product.product", odoo_id)
if not pid:
return None
p = _ask(PRODUCT.get(f"/products/{pid}", actor="sales-order"), f"product {pid}")
return {"product_id": pid, "invoicing_policy": p.get("invoicing_policy") or "ordered"}
def unit(self, odoo_uom_id, product_id):
"""unit_code from product's map, its exact conversion to the product's base unit from product's convert."""
code = self._ours(PRODUCT, "uom.uom", odoo_uom_id)
if not code:
return None
p = _ask(PRODUCT.get(f"/products/{product_id}", actor="sales-order"), f"product {product_id}")
c = _ask(PRODUCT.get(f"/products/{product_id}/convert", params={"quantity": "1", "from_unit": code,
"to_unit": p["base_unit_code"]}, actor="sales-order"),
f"product {product_id} unit {code}")
n, _, d = str(c["exact"]).partition("/")
return {"unit_code": code, "numerator": int(n), "denominator": int(d or 1)}
def tax(self, odoo_id):
return self._ours(TAX, "account.tax", odoo_id)
def payment_term(self, odoo_id):
return self._ours(AR, "account.payment.term", odoo_id)
class OdooJsonRpc(OdooPort):
"""Reads a customer's Odoo through its JSON-RPC endpoint (/jsonrpc, object.execute_kw search_read). Read only: no
write, create or unlink is ever called. ODOO_URL, ODOO_DB, ODOO_LOGIN, ODOO_PASSWORD (an API key) come from the
migration's own environment, never from a request."""
FIELDS = {
"sale.order": ["id", "name", "company_id", "state", "locked", "partner_id", "client_order_ref", "date_order", "user_id", "currency_id",
"pricelist_id", "payment_term_id", "incoterm", "amount_untaxed", "amount_tax", "amount_total"],
"sale.order.line": ["id", "order_id", "sequence", "display_type", "is_downpayment", "product_id", "name", "product_uom_qty", "product_uom",
"price_unit", "discount", "tax_id", "price_subtotal", "price_tax", "qty_delivered", "qty_invoiced", "qty_to_invoice",
"invoice_lines"],
}
def __init__(self, url: str, db: str, login: str, password: str, transport=None):
import httpx
self.url, self.db, self.login, self.password = url.rstrip("/"), db, login, password
self.http, self.uid = httpx.Client(transport=transport, timeout=60), None
@classmethod
def from_env(cls) -> "OdooJsonRpc":
import os
need = [os.environ.get(k, "") for k in ("ODOO_URL", "ODOO_DB", "ODOO_LOGIN", "ODOO_PASSWORD")]
if not all(need):
raise Missing("odoo", "ODOO_URL, ODOO_DB, ODOO_LOGIN and ODOO_PASSWORD are not set in the migration's environment", 0, installed=False)
return cls(*need)
def _rpc(self, service: str, method: str, *args):
r = self.http.post(f"{self.url}/jsonrpc", json={"jsonrpc": "2.0", "method": "call", "id": 1, "params": {"service": service, "method": method, "args": list(args)}})
body = r.json()
if body.get("error"):
raise Missing("odoo", str((body["error"].get("data") or {}).get("message") or body["error"].get("message"))[:300], r.status_code)
return body["result"]
def _search_read(self, model: str, domain: list):
if self.uid is None:
self.uid = self._rpc("common", "login", self.db, self.login, self.password)
return self._rpc("object", "execute_kw", self.db, self.uid, self.password, model, "search_read", [domain], {"fields": self.FIELDS[model]})
@staticmethod
def _id(v):
return v[0] if isinstance(v, (list, tuple)) and v else (v or None)
def orders(self):
out = []
for r in self._search_read("sale.order", []):
out.append({**r, "company_id": self._id(r["company_id"]), "partner_id": self._id(r["partner_id"]), "user_id": self._id(r["user_id"]),
"currency": (r["currency_id"] or [0, ""])[1], "pricelist_name": (r["pricelist_id"] or [0, ""])[1],
"payment_term_id": self._id(r["payment_term_id"]), "incoterm": (r["incoterm"] or [0, ""])[1]})
return out
def lines(self, order_ids):
out = []
for r in self._search_read("sale.order.line", [["order_id", "in", list(order_ids)]]):
out.append({**r, "order_id": self._id(r["order_id"]), "product_id": self._id(r["product_id"]), "product_uom": self._id(r["product_uom"]),
"tax_ids": r.get("tax_id") or [], "invoice_move_ids": []})
return out
class Ports:
"""The one place a use case reaches a neighbour from. Replace an attribute to swap a neighbour (tests do)."""
def __init__(self):
self.product: ProductPort = RealProduct()
self.tax: TaxPort = RealTax()
self.payments: PaymentsPort = RealPayments()
self.invoicing: InvoicingPort = RealInvoicing()
self.partners: PartnersPort = RealPartners()
self.inventory: InventoryPort = RealInventory()
self.gl: GlPort = RealGl()
self.purchasing: PurchasingPort = RealPurchasing()
self.maps: MapsPort = RealMaps()
self.odoo: Optional[OdooPort] = None # None: read the customer's Odoo from the migration's environment
def invoicing_installed(self) -> bool:
return isinstance(self.invoicing, RealInvoicing) is False or AR.installed
def inventory_installed(self) -> bool:
return isinstance(self.inventory, RealInventory) is False or INV.installed
def gl_installed(self) -> bool:
return isinstance(self.gl, RealGl) is False or GL.installed
def purchasing_installed(self) -> bool:
return isinstance(self.purchasing, RealPurchasing) is False or PUR.installed
def tax_installed(self) -> bool:
return isinstance(self.tax, RealTax) is False or TAX.installed
def payments_installed(self) -> bool:
return isinstance(self.payments, RealPayments) is False or PAY.installed
ports = Ports()