dodil/erp-brick-manufacturing
DODIL ERP brick: manufacturing
src/brick_manufacturing/infrastructure/seams.py
"""The neighbours this brick asks, each behind ONE adapter - the domain never knows how they are reached.
product (reads) GET /products/{product_id}, GET /products/{product_id}/convert: the base unit, and an EXACT conversion into it
inventory (reads) GET /availability/{product_id}, GET /valuation/products/{product_id}: what is there, what a unit costs
(writes) POST /goods-issues/from-source (source_kind production_issue), POST /receipts (production_receipt |
by_product_receipt), POST /valuation-adjustments (production_settlement, an average-cost company)
gl (writes) POST /journal-entry/journals (source_module manufacturing, event settle)
Writes are NEVER called from a use case: they are the payloads of outbox rows, sent after the local commit by
application/dispatch.py. Every call is checked in tests/conftest.py against the CALLEE's signed spec (erp-method).
Every answer is a reply, never an exception: not installed (409), failed (502), did not answer (503).
"""
from decimal import Decimal
from typing import Optional
from erp_platform.seam import Seam
PRODUCT = Seam("product", "PRODUCT_BASE_URL", roles=["service"], permissions=["product:product:read"])
INVENTORY = Seam("inventory", "INVENTORY_BASE_URL", roles=["service"],
permissions=["inv:availability:read", "inv:valuation:read", "inv:stock:read", "inv:issue:intake",
"inv:receipt:intake", "inv:valuation:post", "inv:standard_price:intake"])
LEDGER = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post", "gl:cost_centre:read", "gl:period:read"])
SALES = Seam("sales-order", "SALES_ORDER_BASE_URL", roles=["service"], permissions=["so:demand:read"])
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 _ask(seam: Seam, label: str, path: str, params: Optional[dict] = None) -> dict:
if not seam.installed:
raise Unavailable(409, f"{label} is not part of this installation: production needs it to know products, stock and value")
r = seam.get(path, params={k: v for k, v in (params or {}).items() if v not in (None, "")})
if r.ok:
return r.data
if r.status == 404:
raise Unavailable(404, f"{label} has no {path}")
if r.status and r.status < 500:
raise Unavailable(422 if r.status == 422 else 502, f"{label} answered {r.status} ({r.reason or 'no reason given'}) for {path}")
if r.status:
raise Unavailable(502, f"{label} failed ({r.status}: {r.reason or 'no reason given'}) - try again")
raise Unavailable(503, f"{label} did not answer ({r.reason or 'no reason given'}) - try again")
def product(product_id: str, on: str) -> dict:
"""{base_unit_code, kind, roles, units, ...}: the product's own answer."""
return _ask(PRODUCT, "product", f"/products/{product_id}", {"on": on})
def to_base(product_id: str, quantity: Decimal, unit_code: str, base_unit_code: str) -> Decimal:
"""`quantity` of `unit_code` in the product's base unit, EXACT (product refuses, 422, what does not fit - never rounds)."""
if unit_code == base_unit_code:
return quantity
got = _ask(PRODUCT, "product", f"/products/{product_id}/convert",
{"quantity": f"{quantity:f}", "from_unit": unit_code, "to_unit": base_unit_code})
return Decimal(str(got["converted"]))
def available(product_id: str, warehouse_id: str) -> Decimal:
"""What is free to issue in the warehouse (inventory's availability), in the product's base unit."""
got = _ask(INVENTORY, "inventory", f"/availability/{product_id}", {"warehouse_id": warehouse_id})
rows = [w for w in got.get("warehouses", []) if w.get("warehouse_id") == warehouse_id]
return Decimal(str(rows[0]["available"])) if rows else Decimal("0")
def unit_cost(product_id: str, company_code: str, as_of: str) -> Optional[Decimal]:
"""The product's cost per base unit as inventory values it on `as_of`: the standard price on a standard-cost company, the
average on an average-cost one (inventory's `average_cost`). None when inventory has no value for it."""
got = _ask(INVENTORY, "inventory", f"/valuation/products/{product_id}", {"company_code": company_code, "as_of": as_of})
v = got.get("average_cost")
return Decimal(str(v)) if v not in (None, "") else None
def round_up(product_id: str, site_id: str, quantity: Decimal, unit_code: str, on: str) -> Decimal:
"""`quantity` rounded the way product's profile for the production role rounds a proposal (UP to a lot): product answers, we never round."""
got = _ask(PRODUCT, "product", f"/products/{product_id}/round-proposal",
{"site_id": site_id, "quantity": f"{quantity:f}", "unit_code": unit_code, "on": on})
return Decimal(str(got["rounded"]))
def product_sites(site_id: str, on: str) -> dict:
"""{product_id: site row} of the products planned at the warehouse (safety stock, procurement type, lead days)."""
got = _ask(PRODUCT, "product", "/product-sites", {"site_id": site_id, "planning": "active", "on": on})
rows = got.get("body", []) if isinstance(got.get("body"), list) else got.get("rows", got.get("sites", [])) # a list body arrives as {"body": [...]}
return {r["product_id"]: r for r in rows}
def sales_demand(product_id: str, warehouse_id: str, start: str, end: str) -> list:
"""[{needed_on, open_base}] open sales-order lines (no prices). Sales is OPTIONAL: not installed means no demand from it."""
if not SALES.installed:
return []
out, offset = [], 0
while True:
got = _ask(SALES, "sales-order", "/demand", {"product_id": product_id, "warehouse_id": warehouse_id, "from": start, "to": end,
"limit": 500, "offset": offset})
rows = got.get("demand", [])
out += [{"needed_on": r["needed_on"], "open_base": Decimal(str(r["open_base"]))} for r in rows]
if not got.get("more") or not rows:
return out
offset += len(rows)
def cost_centre(code: str) -> Optional[dict]:
"""gl's cost centre {cost_centre, status, ...}; None when gl is not installed (activities are then not posted); a 404 when unknown."""
if not LEDGER.installed:
return None
return _ask(LEDGER, "gl", f"/cost-centres/{code}")
def lot_trace(product_id: str, lot_code: str) -> dict:
"""Where inventory says a lot went (delivered_to, on_hand, history); {} when inventory is not installed or does not know the lot."""
if not INVENTORY.installed:
return {}
try:
return _ask(INVENTORY, "inventory", f"/lots/{product_id}/{lot_code}/trace")
except Unavailable as e:
if e.http == 404:
return {}
raise
# The writes: (seam, method, path) per outbox target, used by application/dispatch.py.
TARGETS = {
"inventory.issue": (INVENTORY, "/goods-issues/from-source"),
"inventory.receipt": (INVENTORY, "/receipts"),
"inventory.adjustment": (INVENTORY, "/valuation-adjustments"),
"inventory.standard_price": (INVENTORY, "/standard-prices"),
"gl.journal": (LEDGER, "/journal-entry/journals"),
}