dodil/erp-brick-pos
DODIL ERP brick: point of sale
src/brick_pos/infrastructure/seams.py
"""The neighbours this brick talks to, each behind ONE adapter: the domain never knows how they are reached.
product GET /price-lists/{price_list_id}/prices?on=&limit=&offset= the shop's prices on the business date (PRD-40)
GET /products/{product_id}?on= the product's tax category and name (PRD-28)
GET /products/{product_id}/convert?quantity=&from_unit=&to_unit= a recipe quantity converts exactly (PRD-21)
manufacturing GET /boms/{product_id}/explode?usage=sale&quantity=&unit_code=&modifiers=&on= the recipe, when manufacturing is the source (MFG-30)
tax GET /rates?country=&categories=&on=&consumption= the rate by category, eat-in or takeaway (TAX-52)
inventory POST /goods-issues/from-source stock leaves the shop's location (INV-58)
POST /receipts a restocked refund line comes back (INV-59)
GET /warehouses/{warehouse_id}, GET /locations?usage=&warehouse_id= the shop's warehouse and location exist (POS-01, INV-01)
gl GET /companies/{company_code} the shop's company exists (POS-01, GL-61)
POST /journal-entry/journals a session's summary journal (GL-30, source_module pos)
payments POST /deposit-batches/expected a sealed cash bag expected at the bank (PAY-49)
Every answer is a reply, never an exception: not installed (409), failed (502), did not answer (503). pos NEVER writes a
neighbour's table. Reads happen while a sale is rung (they have no side effect); the one WRITE, the goods issue, is sent only
from a pending outbox row after the receipt's own commit (application/use_cases.py: dispatch_outbox).
KNOWN SPEC GAP (reported with this slice): product's spec lists GET /products/{product_id}?on= as a seam for inventory, not
for pos, so tests/conftest.py allows exactly that route and test_seams.py fails the day product lists it."""
from datetime import date
from decimal import Decimal
from typing import Optional
from erp_platform.seam import Seam
from brick_pos.domain.errors import RuleBroken
from brick_pos.domain.pricing import Price, Rate
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:code:read", "tax:document_tax:compute"])
MANUFACTURING = Seam("manufacturing", "MANUFACTURING_BASE_URL", roles=["service"], permissions=["mfg:bom:explode"])
INVENTORY = Seam("inventory", "INVENTORY_BASE_URL", roles=["service"], permissions=["inv:issue:intake", "inv:receipt:intake", "inv:stock:read"])
LEDGER = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post", "gl:company:read"])
PAYMENTS = Seam("payments", "PAYMENTS_BASE_URL", roles=["service"], permissions=["pay:deposit_batch:intake"])
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, what: str, path: str, params: dict) -> dict:
if not seam.installed:
raise Unavailable(409, f"{what} is not part of this installation, and a till cannot sell without it")
r = seam.get(path, params={k: v for k, v in params.items() if v not in (None, "")})
if r.ok:
return r.data
if r.status and r.status >= 500:
raise Unavailable(502, f"{what} failed ({r.status}: {r.reason or 'no reason given'}) - try again")
if r.status == 404:
raise Unavailable(409, f"{what} does not know {path}: {r.reason or 'not found'}")
if r.status:
raise Unavailable(502, f"{what} answered {r.status} ({r.reason or 'no reason given'})")
raise Unavailable(503, f"{what} did not answer ({r.reason or 'no reason given'}) - try again")
def price_list(price_list_id: str, on: date, wanted: set) -> tuple:
"""(currency, prices_include_tax, [Price]) for the products asked for: pages of 1,000 until every product is seen or
the list ends. One read for the whole list, not one per button."""
prices, offset, currency, incl = [], 0, "", True
while True:
data = _ask(PRODUCT, "the price list (product)", f"/price-lists/{price_list_id}/prices",
{"on": on.isoformat(), "limit": 1000, "offset": offset})
currency, incl = str(data.get("currency") or currency).upper(), bool(data.get("prices_include_tax", incl))
rows = data.get("prices") or []
prices += [Price(product_id=str(p["product_id"]), unit_code=str(p["unit_code"]),
from_quantity=Decimal(str(p.get("from_quantity") or "1")), unit_price=Decimal(str(p["unit_price"])),
pricing_quantity=int(p.get("pricing_quantity") or 1)) for p in rows]
if not data.get("more") or wanted <= {p.product_id for p in prices}:
return currency, incl, prices
offset += 1000
def price_list_all(price_list_id: str, on: date) -> tuple:
"""(currency, prices_include_tax, [Price], next_change_on): EVERY page of the list, for the register's snapshot - read once when a
session opens, so a sale reads nothing."""
prices, offset, currency, incl, nxt = [], 0, "", True, None
while True:
data = _ask(PRODUCT, "the price list (product)", f"/price-lists/{price_list_id}/prices",
{"on": on.isoformat(), "limit": 1000, "offset": offset})
currency, incl = str(data.get("currency") or currency).upper(), bool(data.get("prices_include_tax", incl))
nxt = data.get("next_change_on") or nxt
prices += [Price(product_id=str(p["product_id"]), unit_code=str(p["unit_code"]),
from_quantity=Decimal(str(p.get("from_quantity") or "1")), unit_price=Decimal(str(p["unit_price"])),
pricing_quantity=int(p.get("pricing_quantity") or 1)) for p in data.get("prices") or []]
if not data.get("more"):
return currency, incl, prices, date.fromisoformat(nxt) if nxt else None
offset += 1000
def catalogue(ids: list, on: date) -> dict:
"""product_id -> {tax_category, name} for the products the item master knows and gives a tax category; the others are left out (not
sellable), a neighbour that fails raises. One read per distinct product."""
out = {}
for pid in sorted(set(ids)):
try:
data = _ask(PRODUCT, "the item master (product)", f"/products/{pid}", {"on": on.isoformat()})
except Unavailable as e:
if e.http == 409 and PRODUCT.installed and "does not know" in e.reason:
continue
raise
if data.get("tax_category"):
out[pid] = {"tax_category": str(data["tax_category"]), "name": str(data.get("name") or "")}
return out
def rate_table(country: str, categories: list, on: date) -> tuple:
"""([Rate], valid_until): tax's rates for the categories on the date, every consumption variant (the register picks eat-in or takeaway
itself); `valid_until` is the day its rules next change (TAX-52)."""
if not categories:
return [], None
data = _ask(TAX, "the tax rules (tax)", "/rates", {"country": country, "categories": ",".join(sorted(set(categories))), "on": on.isoformat()})
rows = [Rate(category_code=str(r["category_code"]), supply=str(r.get("supply") or "goods"), consumption=str(r.get("consumption") or "any"),
tax_code=str(r["tax_code"]), rate_pct=Decimal(str(r["rate_pct"]))) for r in data.get("rates") or []]
until = data.get("valid_until")
return rows, date.fromisoformat(until) if until else None
def price_list_header(price_list_id: str, on: date) -> tuple:
"""(currency, prices_include_tax): one row asked for, to learn the list's currency when a session opens."""
data = _ask(PRODUCT, "the price list (product)", f"/price-lists/{price_list_id}/prices",
{"on": on.isoformat(), "limit": 1, "offset": 0})
return str(data.get("currency") or "").upper(), bool(data.get("prices_include_tax", True))
def products(ids: list, on: date) -> dict:
"""product_id -> {tax_category, name}, one read per DISTINCT product rung."""
out = {}
for pid in sorted(set(ids)):
data = _ask(PRODUCT, "the item master (product)", f"/products/{pid}", {"on": on.isoformat()})
if not data.get("tax_category"):
raise Unavailable(409, f"product {pid} has no tax category, so no VAT can be found for it")
out[pid] = {"tax_category": str(data["tax_category"]), "name": str(data.get("name") or "")}
return out
def rates(country: str, categories: list, on: date, eat_in: Optional[bool]) -> list:
"""tax's rates for the categories on the business date. A category tax does not know is `missing` (the caller says which)."""
data = _ask(TAX, "the tax rules (tax)", "/rates",
{"country": country, "categories": ",".join(sorted(set(categories))), "on": on.isoformat(),
"consumption": None if eat_in is None else ("eat_in" if eat_in else "takeaway")})
return [Rate(category_code=str(r["category_code"]), supply=str(r.get("supply") or "goods"),
consumption=str(r.get("consumption") or "any"), tax_code=str(r["tax_code"]), rate_pct=Decimal(str(r["rate_pct"])))
for r in data.get("rates") or []]
def _send(seam: Seam, what: str, path: str, body: dict) -> tuple:
"""One write to a neighbour. (ok, permanent, reason): ok on 200/201 (a replay is 200); a 4xx is permanent (the neighbour refused
the request: a closed period, an unknown bank account), a 5xx or silence is not (the row stays pending and is tried again)."""
if not seam.installed:
return False, False, f"{what} is not part of this installation"
r = seam.post(path, body)
if r.ok:
return True, False, ""
return False, bool(r.status and 400 <= r.status < 500), f"{what} answered {r.status or 'nothing'}: {r.reason or 'no reason given'}"
def stocked_product(product_id: str, on: date) -> dict:
"""{product_id, stocked, base_unit_code}: what a recipe needs to know of a product it names (GET /products/{id}, PRD-28)."""
data = _ask(PRODUCT, "the item master (product)", f"/products/{product_id}", {"on": on.isoformat()})
return {"product_id": product_id, "stocked": bool(data.get("stocked_default")), "base_unit_code": str(data.get("base_unit_code") or "")}
def convert_exactly(product_id: str, quantity: Decimal, from_unit: str, to_unit: str) -> None:
"""A recipe quantity converts EXACTLY to the component's base unit or is refused, never rounded (GET /products/{id}/convert, PRD-21)."""
if not PRODUCT.installed:
raise Unavailable(409, "the item master (product) is not part of this installation, and a till cannot sell without it")
r = PRODUCT.get(f"/products/{product_id}/convert", params={"quantity": format(quantity, "f"), "from_unit": from_unit, "to_unit": to_unit})
if r.ok:
return
if r.status == 422:
raise RuleBroken(f"{format(quantity, 'f')} {from_unit} of {product_id} does not convert exactly to {to_unit}: {r.reason or 'refused'}",
gate="units_exact", fix="write the quantity in a unit it converts to exactly")
if r.status == 404:
raise Unavailable(409, f"the item master (product) does not know {product_id} or its unit {from_unit}: {r.reason or 'not found'}")
raise Unavailable(502, f"the item master (product) answered {r.status or 'nothing'}")
def explode(product_id: str, quantity: Decimal, unit_code: str, modifiers: list, on: date) -> Optional[dict]:
"""manufacturing's usage-sale bill of materials for `quantity` of the product (MFG-30): {bom_id, version, components: [(product_id,
quantity, unit_code)]}, or None when no released version is valid on the date (the product is then consumed itself). The spec lists
unit_code and modifiers as part of the request, so both are always sent (modifiers empty when there are none)."""
if not MANUFACTURING.installed:
raise Unavailable(409, "recipes come from manufacturing (recipe_source) and it is not part of this installation")
r = MANUFACTURING.get(f"/boms/{product_id}/explode", params={"usage": "sale", "quantity": format(quantity, "f"), "unit_code": unit_code,
"modifiers": ",".join(sorted(modifiers)), "on": on.isoformat()})
if r.ok:
return {"bom_id": str(r.data["bom_id"]), "version": int(r.data["version"]),
"components": [(str(c["product_id"]), Decimal(str(c["quantity"])), str(c["unit_code"])) for c in r.data.get("components") or []]}
if r.status == 404:
return None
if r.status:
raise Unavailable(502, f"manufacturing answered {r.status} ({r.reason or 'no reason given'})")
raise Unavailable(503, f"manufacturing did not answer ({r.reason or 'no reason given'}) - try again")
def _failed(seam: Seam, what: str, r) -> Unavailable:
if r.status and r.status >= 500:
return Unavailable(502, f"{what} failed ({r.status}: {r.reason or 'no reason given'}) - try again")
if r.status:
return Unavailable(502, f"{what} answered {r.status} ({r.reason or 'no reason given'})")
return Unavailable(503, f"{what} did not answer ({r.reason or 'no reason given'}) - try again")
def company(company_code: str) -> Optional[dict]:
"""gl's company (GL-61: name, country, functional_currency), or None when gl is not part of this installation (nothing to check against)."""
if not LEDGER.installed:
return None
r = LEDGER.get(f"/companies/{company_code}")
if r.ok:
return r.data
if r.status == 404:
raise RuleBroken(f"company {company_code} does not exist in gl, the books", gate="company_known", fix="create the company in gl first")
raise _failed(LEDGER, "gl", r)
def warehouse(warehouse_id: str) -> Optional[dict]:
"""inventory's warehouse (INV-01: warehouse_id, company_code, status), or None when inventory is not part of this installation."""
if not INVENTORY.installed:
return None
r = INVENTORY.get(f"/warehouses/{warehouse_id}")
if r.ok:
return r.data
if r.status == 404:
raise RuleBroken(f"warehouse {warehouse_id} does not exist in inventory", gate="warehouse_known", fix="create the warehouse in inventory first")
raise _failed(INVENTORY, "inventory", r)
def internal_locations(warehouse_id: str) -> Optional[list]:
"""The ids of the internal locations of a warehouse (INV-01 GET /locations), or None when inventory is not part of this installation."""
if not INVENTORY.installed:
return None
r = INVENTORY.get("/locations", params={"usage": "internal", "warehouse_id": warehouse_id})
if r.ok:
return [str(x["location_id"]) for x in r.data.get("locations") or []]
raise _failed(INVENTORY, "inventory", r)
def goods_issue(body: dict) -> tuple:
"""Send one goods issue to inventory (INV-58): idempotent on each line's move_ref."""
return _send(INVENTORY, "inventory", "/goods-issues/from-source", body)
def goods_receipt(body: dict) -> tuple:
"""Send one restocked refund line to inventory (INV-59): idempotent on move_ref."""
return _send(INVENTORY, "inventory", "/receipts", body)
def journal(body: dict) -> tuple:
"""Post a session's summary journal to gl (GL-30): journal_id is derived from (source_document, event), so a resend is gl's replay."""
return _send(LEDGER, "gl", "/journal-entry/journals", body)
def deposit_batch(body: dict) -> tuple:
"""Tell payments a cash bag is on its way (PAY-49): idempotent on batch_ref, so a resend is its replay."""
return _send(PAYMENTS, "payments", "/deposit-batches/expected", body)