dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/infrastructure/seams.py
"""The neighbours this brick asks, each by THEIR spec's exact route (module.yaml seams), behind one small adapter each.
product GET /products/{id}, GET /products/{id}/convert, POST /products/{id}/units/{unit}/used, GET /gtin/{gtin},
GET /product-sites, GET /products/{id}/round-proposal
so POST /sales-orders/{id}/deliveries, GET /sales-orders/{id}, GET /sales-orders/{id}/fulfilment,
POST /sales-orders/{id}/return-inspections
pur POST /purchase-orders/{id}/receipts, GET /purchase-orders/{id}/expected, GET /expected-receipts,
POST /purchase-requests/from-replenishment, POST /consignment-withdrawals
gl POST /journal-entry/journals, GET /periods/open, GET /cost-centres/{cost_centre}
Nothing here is imported by the domain, and no table of a neighbour is read. A neighbour that is not part of the installation,
refuses, fails or does not answer is a `Missing` carrying its real reason and status - never a guess, never an exception from
the transport. Tests register spec-shaped apps with `erp_platform.seam.register_local(name, app)`.
"""
import json
from datetime import date
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, status: int = 0, installed: bool = True):
super().__init__(f"{what}: {reason}")
self.what, self.reason, self.status, self.installed = what, reason, status, installed
@property
def text(self) -> str:
if isinstance(self.reason, dict):
return str(self.reason.get("reason") or self.reason.get("detail") or self.reason)
return str(self.reason)
@property
def gate(self) -> Optional[str]:
return self.reason.get("gate") if isinstance(self.reason, dict) else None
def _reason(raw: str):
try:
d = json.loads(raw)
except (ValueError, TypeError):
return str(raw)[:300]
if isinstance(d, dict):
inner = d.get("detail", d)
if isinstance(inner, dict):
return inner
return str(inner)[:300]
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)
PRODUCT = Seam("product", "PRODUCT_BASE_URL", roles=["service"], permissions=["product:product:read", "product:unit:mark_used"])
BP = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=["bp:migration:read"])
SO = Seam("sales-order", "SALES_ORDER_BASE_URL", roles=["service"], permissions=["so:order:read", "so:order:deliver", "so:return:inspect"])
PUR = Seam("purchasing", "PURCHASING_BASE_URL", roles=["service"], permissions=[
"purchasing:receipt:write", "purchasing:order:read_expected", "purchasing:request:intake", "purchasing:consignment:intake"])
MFG = Seam("manufacturing", "MANUFACTURING_BASE_URL", roles=["service"], permissions=["mfg:plan:read_requirements"])
GL = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post", "gl:period:read", "gl:cost_centre:read", "gl:company:read"])
def product_installed() -> bool:
return PRODUCT.installed
def sales_installed() -> bool:
return SO.installed
def purchasing_installed() -> bool:
return PUR.installed
def ledger_installed() -> bool:
return GL.installed
def manufacturing_installed() -> bool:
return MFG.installed
# ── product ─────────────────────────────────────────────────────────────────────────────────
def product(product_id: str, on: Optional[date] = None) -> dict:
"""GET /products/{product_id}?on= : kind, base_unit_code, roles, units with decimals and measures, valuation_class."""
params = {"on": on.isoformat()} if on else None
return _ask(PRODUCT.get(f"/products/{product_id}", params=params), f"product {product_id}")
def convert(product_id: str, quantity, from_unit: str, to_unit: str) -> dict:
"""GET /products/{product_id}/convert : {exact: "n/d", in_base_unit, base_unit_code, ...}; 422 when the unit is unknown."""
return _ask(PRODUCT.get(f"/products/{product_id}/convert",
params={"quantity": str(quantity), "from_unit": from_unit, "to_unit": to_unit}),
f"product {product_id}")
def mark_unit_used(product_id: str, unit_code: str, document_ref: str) -> dict:
return _ask(PRODUCT.post(f"/products/{product_id}/units/{unit_code}/used", {"document_ref": document_ref}),
f"product {product_id}")
def gtin(code: str) -> dict:
return _ask(PRODUCT.get(f"/gtin/{code}"), f"gtin {code}")
def product_sites(site_id: str, on: date) -> list:
d = _ask(PRODUCT.get("/product-sites", params={"site_id": site_id, "planning": "active", "on": on.isoformat()}), "product sites")
return d.get("body", d) if isinstance(d, dict) else d
def round_proposal(product_id: str, site_id: str, quantity, unit_code: str, on: date) -> dict:
return _ask(PRODUCT.get(f"/products/{product_id}/round-proposal",
params={"site_id": site_id, "quantity": str(quantity), "unit_code": unit_code, "on": on.isoformat()}),
f"product {product_id}")
# ── sales order ─────────────────────────────────────────────────────────────────────────────
def record_delivery(sales_order_id: str, body: dict) -> dict:
"""POST /sales-orders/{id}/deliveries {line_no, quantity, source: inventory, delivery_ref, delivered_on, corrects_ref}."""
return _ask(SO.post(f"/sales-orders/{sales_order_id}/deliveries", body), f"sales order {sales_order_id}")
def sales_order(sales_order_id: str) -> dict:
return _ask(SO.get(f"/sales-orders/{sales_order_id}"), f"sales order {sales_order_id}")
def fulfilment(sales_order_id: str) -> dict:
return _ask(SO.get(f"/sales-orders/{sales_order_id}/fulfilment"), f"sales order {sales_order_id}")
def return_inspection(sales_order_id: str, body: dict) -> dict:
return _ask(SO.post(f"/sales-orders/{sales_order_id}/return-inspections", body), f"sales order {sales_order_id}")
# ── purchasing ──────────────────────────────────────────────────────────────────────────────
def expected(purchase_order_id: str) -> dict:
"""GET /purchase-orders/{id}/expected : lines [{line_no, product_id, unit_code, ordered, received, outstanding, ...}]."""
return _ask(PUR.get(f"/purchase-orders/{purchase_order_id}/expected"), f"purchase order {purchase_order_id}")
def record_receipt(purchase_order_id: str, body: dict) -> dict:
"""POST /purchase-orders/{id}/receipts {delivery_note, received_on, location_id, lines}: the ONE way a receipt is recorded."""
return _ask(PUR.post(f"/purchase-orders/{purchase_order_id}/receipts", body), f"purchase order {purchase_order_id}")
def expected_receipts(product_id: str, warehouse_id: str) -> list:
d = _ask(PUR.get("/expected-receipts", params={"product_id": product_id, "warehouse_id": warehouse_id}), "expected receipts")
return d.get("body", []) if isinstance(d, dict) else d
def purchase_request(body: dict) -> dict:
return _ask(PUR.post("/purchase-requests/from-replenishment", body), "purchasing")
def consignment_withdrawal(body: dict) -> dict:
return _ask(PUR.post("/consignment-withdrawals", body), "purchasing")
# ── manufacturing ───────────────────────────────────────────────────────────────────────────
def dependent_requirements(product_id: str, warehouse_id: str, frm: date, to: date) -> list:
"""GET /dependent-requirements?product_id=&warehouse_id=&from=&to= (INV-61, manufacturing MFG-27): [{required_on, quantity (base unit),
source: planned | released, production_order_id}]. Inventory never reads mfg_ tables."""
d = _ask(MFG.get("/dependent-requirements", params={"product_id": product_id, "warehouse_id": warehouse_id, "from": frm.isoformat(),
"to": to.isoformat()}), "manufacturing")
return d.get("requirements", []) if isinstance(d, dict) else d
# ── the ledger ──────────────────────────────────────────────────────────────────────────────
def post_journal(body: dict) -> dict:
return _ask(GL.post("/journal-entry/journals", body), "the ledger")
def company_currency(company_code: str) -> str:
"""The company's functional currency, the ledger's (shared.yaml functional_currency: every module reads it through gl's
GET /companies/{company_code} and keeps no row of its own)."""
got = _ask(GL.get(f"/companies/{company_code}"), f"company {company_code}")
if not got.get("functional_currency"):
raise Missing(f"company {company_code}", f"the ledger names no functional currency for company {company_code}", 422)
return str(got["functional_currency"])
def period_open(on: date, company_code: str) -> dict:
return _ask(GL.get("/periods/open", params={"date": on.isoformat(), "company_code": company_code}), "the ledger")
def cost_centre(code: str) -> dict:
return _ask(GL.get(f"/cost-centres/{code}"), f"cost centre {code}")