dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/infrastructure/seams.py
"""The neighbours this brick asks, each behind a small PORT shaped from THEIR spec's contract (module.yaml seams).
business-partner promote(role supplier), get PUR-21
product get, convert, unit_used, rounding PUR-24, PUR-35, PUR-37
tax compute, check_contents, freeze, party, CIS, imports PUR-28, PUR-61, PUR-62
gl post_journal, period_open PUR-20
inventory receive, return_to_supplier, locations, subcontract_provision, valuation_adjustment PUR-25, 26, 57, 63
payments bank_accounts, account_details (unmasked, never stored) PUR-19
ar customer_status PUR-22
fixed-assets asset, classes, acquisition PUR-55, PUR-56
objects put/get by key (the bank file, a signed PDF)
odoo, maps the one-off migration's reader and the earlier bricks' id maps
Nothing here is imported by the domain. A neighbour that is not part of the installation, refuses, fails or does not answer is a
`Missing` with its real reason and status (and its gates, when it answered {gate, reason, fix}) — never a guess. Tests replace
`ports.<name>` with in-memory fakes (tests/fakes.py) that answer exactly what the callee's SPEC route answers.
"""
import json
import os
from datetime import date
from typing import Optional
from urllib.parse import quote
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, gates: Optional[list] = None, body: Optional[dict] = None):
super().__init__(f"{what}: {reason}")
self.what, self.reason, self.status, self.installed = what, reason, status, installed
self.gates, self.body = gates or [], body or {}
@property
def gate(self) -> str:
return (self.gates[0].get("gate") if self.gates else "") or (self.body.get("gate") or "")
def _parse(raw: str) -> dict:
try:
d = json.loads(raw)
except (ValueError, TypeError):
return {}
return d if isinstance(d, dict) else {}
def _reason(raw: str) -> str:
d = _parse(raw)
if d:
detail = d.get("detail")
if isinstance(detail, dict):
d = detail
elif detail:
return str(detail)[:300]
return str(d.get("reason") or raw)[:300]
return str(raw)[:300]
def _gates(raw: str) -> list:
d = _parse(raw)
d = d.get("detail") if isinstance(d.get("detail"), dict) else d
if d.get("gates"):
return [g for g in d["gates"] if isinstance(g, dict)]
if d.get("gate"):
return [{"gate": d["gate"], "met": False, "reason": d.get("reason", ""), "fix": d.get("fix", "")}]
return []
def ask(reply, what: str) -> dict:
"""The answer's data, with `_status` (200, 201...), or a Missing carrying the neighbour's own reason, gates and body."""
if reply.ok:
return {**reply.data, "_status": reply.status}
raw = reply.reason or ""
d = _parse(raw)
d = d.get("detail") if isinstance(d.get("detail"), dict) else d
raise Missing(what, _reason(raw), reply.status, installed="not installed" not in raw, gates=_gates(raw), body=d)
# ── ports ─────────────────────────────────────────────────────────────────────────────────────
class BpPort:
def promote(self, caller: dict, *, bp_key: str, legal_name: str, country: str = "", tax_id: str = "", domain: str = "") -> dict:
raise NotImplementedError
def get(self, caller: dict, bp_id: int) -> dict:
raise NotImplementedError
class ProductPort:
def get(self, caller: dict, product_id: str, on: Optional[date] = None) -> dict:
raise NotImplementedError
def convert(self, caller: dict, product_id: str, quantity: str, from_unit: str, to_unit: str) -> dict:
raise NotImplementedError
def unit_used(self, product_id: str, unit_code: str, document_ref: str) -> dict:
raise NotImplementedError
def rounding(self, caller: dict, product_id: str, quantity: str, unit_code: str, on: date, role: str = "purchase") -> dict:
raise NotImplementedError
class TaxPort:
def compute(self, body: dict) -> dict:
raise NotImplementedError
def check_contents(self, document_ref: str, contents: dict) -> dict:
raise NotImplementedError
def freeze(self, document_ref: str, contents: Optional[dict] = None) -> dict:
raise NotImplementedError
def party(self, bp_id: int) -> dict:
raise NotImplementedError
def cis_subcontractor(self, bp_id: int, on: date) -> dict:
raise NotImplementedError
def cis_deduct(self, body: dict) -> dict:
raise NotImplementedError
def cis_reverse(self, payment_ref: str, body: dict) -> dict:
raise NotImplementedError
def record_import(self, body: dict) -> dict:
raise NotImplementedError
class GlPort:
def post_journal(self, body: dict) -> dict:
raise NotImplementedError
def period_open(self, on: date, company_code: Optional[str] = None) -> dict:
raise NotImplementedError
def cost_centre(self, cost_centre: str) -> dict:
raise NotImplementedError
def company(self, company_code: str) -> dict:
raise NotImplementedError
def companies(self) -> list:
raise NotImplementedError
class InventoryPort:
def receive(self, body: dict) -> dict:
raise NotImplementedError
def return_to_supplier(self, body: dict) -> dict:
raise NotImplementedError
def locations(self, usage: Optional[str] = None, warehouse_id: Optional[str] = None) -> dict:
raise NotImplementedError
def subcontract_provision(self, body: dict) -> dict:
raise NotImplementedError
def valuation_adjustment(self, body: dict) -> dict:
raise NotImplementedError
class PaymentsPort:
def bank_accounts(self) -> dict:
raise NotImplementedError
def account_details(self, bank_account_id: str, purpose: str) -> dict:
"""OUR paying account, unmasked (PAY-27). Held in memory for one file; the caller never stores or logs it."""
raise NotImplementedError
class ArPort:
def customer_status(self, bp_id: int) -> dict:
raise NotImplementedError
class FixedAssetsPort:
def asset(self, asset_id: str) -> dict:
raise NotImplementedError
def classes(self, on: date) -> dict:
raise NotImplementedError
def acquisition(self, body: dict) -> dict:
raise NotImplementedError
class ObjectsPort:
def put(self, key: str, body: bytes, content_type: str) -> str:
raise NotImplementedError
def get(self, key: str) -> Optional[bytes]:
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. `db` names the
database; each method returns the rows of that Odoo model with the fields module.yaml migration.odoo.reads lists."""
db: str = ""
def read(self, model: str) -> list:
raise NotImplementedError
class MapsPort:
"""The earlier bricks' Odoo id maps (shared.yaml conventions.odoo_migration.map_route): partner -> bp_id, product -> product_id,
uom -> unit_code, tax -> tax_code. None = unmapped."""
db: str = ""
def partner(self, odoo_id: int) -> Optional[int]:
raise NotImplementedError
def product(self, odoo_id: int) -> Optional[str]:
raise NotImplementedError
def unit(self, odoo_uom_id: int) -> Optional[str]:
raise NotImplementedError
def tax(self, odoo_id: int) -> Optional[str]:
raise NotImplementedError
# ── the real adapters: each neighbour's REAL route, as its spec states it ─────────────────────────
BP = Seam("business-partner", "BUSINESS_PARTNER_BASE_URL", roles=["service"], permissions=["bp:partner:promote_supplier", "bp:partner:read"])
PRODUCT = Seam("product", "PRODUCT_BASE_URL", roles=["service"], permissions=["product:product:read", "product:unit:mark_used"])
TAX = Seam("tax", "TAX_BASE_URL", roles=["service"], permissions=["tax:document_tax:compute", "tax:code:read", "tax:cis:read", "tax:cis:deduct",
"tax:import_vat:record"])
GL = Seam("gl", "GL_BASE_URL", roles=["service"], permissions=["gl:journal:post", "gl:period:read", "gl:cost_centre:read", "gl:company:read"])
INV = Seam("inventory", "INVENTORY_BASE_URL", roles=["service"], permissions=["inv:receipt:intake", "inv:stock:read", "inv:valuation:post"])
PAY = Seam("payments", "PAYMENTS_BASE_URL", roles=["service"], permissions=["pay:receipt:read", "pay:bank_account:read_unmasked"])
AR = Seam("ar", "AR_BASE_URL", roles=["service"], permissions=["ar:status:read"])
FA = Seam("fixed-assets", "FIXED_ASSETS_BASE_URL", roles=["service"], permissions=["fa:asset:read", "fa:acquisition:intake"])
def _actor(caller) -> str:
return (caller or {}).get("email", "") if isinstance(caller, dict) else ""
class RealBp(BpPort):
def promote(self, caller, *, bp_key, legal_name, country="", tax_id="", domain=""):
body = {"bp_key": bp_key, "role": "supplier", "legal_name": legal_name, "country": country, "tax_id": tax_id, "domain": domain}
return ask(BP.post("/partners/promote", {k: v for k, v in body.items() if v != ""}, actor=_actor(caller)), "business-partner")
def get(self, caller, bp_id):
return ask(BP.get(f"/partners/{int(bp_id)}", actor=_actor(caller), caller=caller), f"partner {bp_id}")
class RealProduct(ProductPort):
def get(self, caller, product_id, on=None):
params = {"on": on.isoformat()} if on else None
return ask(PRODUCT.get(f"/products/{product_id}", params=params, actor=_actor(caller)), f"product {product_id}")
def convert(self, caller, product_id, quantity, from_unit, to_unit):
return ask(PRODUCT.get(f"/products/{product_id}/convert", params={"quantity": quantity, "from_unit": from_unit, "to_unit": to_unit},
actor=_actor(caller)), f"product {product_id}")
def unit_used(self, product_id, unit_code, document_ref):
return ask(PRODUCT.post(f"/products/{product_id}/units/{unit_code}/used", {"document_ref": document_ref}, actor="purchasing"),
f"product {product_id}")
def rounding(self, caller, product_id, quantity, unit_code, on, role="purchase"):
return ask(PRODUCT.get(f"/products/{product_id}/rounding", params={"role": role, "quantity": quantity, "unit_code": unit_code,
"on": on.isoformat()}, actor=_actor(caller)), f"product {product_id}")
class RealTax(TaxPort):
def compute(self, body):
return ask(TAX.post("/documents/compute", body, actor="purchasing"), "tax")
def check_contents(self, document_ref, contents):
return ask(TAX.post(f"/documents/{document_ref}/check-contents", {"direction": "purchase", "contents": contents}, actor="purchasing"), "tax")
def freeze(self, document_ref, contents=None):
return ask(TAX.post(f"/documents/{document_ref}/freeze", {"contents": contents} if contents else {}, actor="purchasing"), "tax")
def party(self, bp_id):
return ask(TAX.get(f"/parties/{int(bp_id)}", actor="purchasing"), "tax")
def cis_subcontractor(self, bp_id, on):
return ask(TAX.get(f"/cis/subcontractors/{int(bp_id)}", params={"on": on.isoformat()}, actor="purchasing"), "tax")
def cis_deduct(self, body):
return ask(TAX.post("/cis/deductions", body, actor="purchasing"), "tax")
def cis_reverse(self, payment_ref, body):
return ask(TAX.post(f"/cis/deductions/{payment_ref}/reverse", body, actor="purchasing"), "tax")
def record_import(self, body):
return ask(TAX.post("/imports", body, actor="purchasing"), "tax")
class RealGl(GlPort):
def post_journal(self, body):
return ask(GL.post("/journal-entry/journals", body, actor="purchasing"), "the ledger")
def period_open(self, on, company_code=None):
params = {"date": on.isoformat(), **({"company_code": company_code} if company_code else {})}
return ask(GL.get("/periods/open", params=params, actor="purchasing"), "the ledger")
def cost_centre(self, cost_centre):
return ask(GL.get(f"/cost-centres/{quote(str(cost_centre), safe='')}", actor="purchasing"), "the ledger")
def company(self, company_code):
return ask(GL.get(f"/companies/{quote(str(company_code), safe='')}", actor="purchasing"), "the ledger")
def companies(self):
reply = GL.get("/companies", actor="purchasing")
if not reply.ok:
ask(reply, "the ledger")
got = reply.data
return got if isinstance(got, list) else (got or {}).get("companies", [])
class RealInventory(InventoryPort):
def receive(self, body):
return ask(INV.post("/receipts", body, actor="purchasing"), "inventory")
def return_to_supplier(self, body):
return ask(INV.post("/returns-to-supplier", body, actor="purchasing"), "inventory")
def locations(self, usage=None, warehouse_id=None):
params = {k: v for k, v in (("usage", usage), ("warehouse_id", warehouse_id)) if v}
return ask(INV.get("/locations", params=params, actor="purchasing"), "inventory")
def subcontract_provision(self, body):
return ask(INV.post("/subcontract-provisions", body, actor="purchasing"), "inventory")
def valuation_adjustment(self, body):
return ask(INV.post("/valuation-adjustments", body, actor="purchasing"), "inventory")
class RealPayments(PaymentsPort):
def bank_accounts(self):
return ask(PAY.get("/bank-accounts", actor="purchasing"), "payments")
def account_details(self, bank_account_id, purpose):
return ask(PAY.get(f"/bank-accounts/{quote(str(bank_account_id), safe='')}/account-details", params={"purpose": purpose}, actor="purchasing"), "payments")
class RealAr(ArPort):
def customer_status(self, bp_id):
return ask(AR.get(f"/customers/{int(bp_id)}/status", actor="purchasing"), "invoicing")
class RealFixedAssets(FixedAssetsPort):
def asset(self, asset_id):
return ask(FA.get(f"/assets/{asset_id}", actor="purchasing"), "fixed-assets")
def classes(self, on):
return ask(FA.get("/asset-classes", params={"on": on.isoformat()}, actor="purchasing"), "fixed-assets")
def acquisition(self, body):
return ask(FA.post("/acquisitions", body, actor="purchasing"), "fixed-assets")
class TableObjects(ObjectsPort):
"""Objects kept in a table of this brick (pur_object): the default where no object store is configured. Kept once by key."""
def __init__(self, repository):
self.repository = repository
def put(self, key, body, content_type):
import base64
import hashlib
with self.repository() as repo:
if repo.objects.get(key) is None:
repo.objects.put({"object_key": key, "content_type": content_type, "body": base64.b64encode(body).decode(),
"sha256": hashlib.sha256(body).hexdigest()})
return key
def get(self, key):
import base64
with self.repository() as repo:
row = repo.objects.get(key)
return base64.b64decode(row["body"]) if row else None
class S3Objects(ObjectsPort):
"""DataK3's S3 endpoint (object.uk-lon-1.dodil.io), signed with SigV4 (OBJECT_ENDPOINT, BUCKET, OBJECT_ACCESS_KEY, OBJECT_SECRET_KEY).
Kept once: an existing key is not overwritten."""
def __init__(self):
self.endpoint = os.environ.get("OBJECT_ENDPOINT", "https://object.uk-lon-1.dodil.io").rstrip("/")
self.bucket = os.environ.get("BUCKET", "")
self.access, self.secret = os.environ.get("OBJECT_ACCESS_KEY", ""), os.environ.get("OBJECT_SECRET_KEY", "")
self.region = os.environ.get("OBJECT_REGION", "us-east-1")
@property
def configured(self) -> bool:
return bool(self.bucket and self.access and self.secret)
def _signed(self, method: str, key: str, body: bytes, content_type: str = "") -> tuple:
import datetime as dt
import hashlib
import hmac
from urllib.parse import quote, urlparse
host = urlparse(self.endpoint).netloc
path = "/" + quote(f"{self.bucket}/{key}")
now = dt.datetime.now(dt.timezone.utc)
amz, day = now.strftime("%Y%m%dT%H%M%SZ"), now.strftime("%Y%m%d")
payload = hashlib.sha256(body).hexdigest()
headers = {"host": host, "x-amz-content-sha256": payload, "x-amz-date": amz}
if content_type:
headers["content-type"] = content_type
signed = ";".join(sorted(headers))
canon = "\n".join([method, path, "", "".join(f"{k}:{headers[k]}\n" for k in sorted(headers)), signed, payload])
scope = f"{day}/{self.region}/s3/aws4_request"
sts = "\n".join(["AWS4-HMAC-SHA256", amz, scope, hashlib.sha256(canon.encode()).hexdigest()])
def h(k, m):
return hmac.new(k, m.encode(), hashlib.sha256).digest()
k = h(h(h(h(("AWS4" + self.secret).encode(), day), self.region), "s3"), "aws4_request")
headers["authorization"] = (f"AWS4-HMAC-SHA256 Credential={self.access}/{scope}, SignedHeaders={signed}, "
f"Signature={hmac.new(k, sts.encode(), hashlib.sha256).hexdigest()}")
return f"{self.endpoint}{path}", headers
def get(self, key):
import httpx
url, headers = self._signed("GET", key, b"")
r = httpx.get(url, headers=headers, timeout=30)
return r.content if r.status_code == 200 else None
def put(self, key, body, content_type):
import httpx
if self.get(key) is not None:
return key
url, headers = self._signed("PUT", key, body, content_type)
r = httpx.put(url, headers=headers, content=body, timeout=30)
if r.status_code >= 300:
raise Missing("object store", f"{r.status_code}: {r.text[:200]}", r.status_code)
return key
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.bp: BpPort = RealBp()
self.product: ProductPort = RealProduct()
self.tax: TaxPort = RealTax()
self.gl: GlPort = RealGl()
self.inventory: InventoryPort = RealInventory()
self.payments: PaymentsPort = RealPayments()
self.ar: ArPort = RealAr()
self.fa: FixedAssetsPort = RealFixedAssets()
self.objects: Optional[ObjectsPort] = None # None: the object store from the environment, else the table
self.odoo: Optional[OdooPort] = None
self.maps: Optional[MapsPort] = None
def installed(self, name: str) -> bool:
real = {"bp": (RealBp, BP), "product": (RealProduct, PRODUCT), "tax": (RealTax, TAX), "gl": (RealGl, GL),
"inventory": (RealInventory, INV), "payments": (RealPayments, PAY), "ar": (RealAr, AR), "fa": (RealFixedAssets, FA)}[name]
return not isinstance(getattr(self, name), real[0]) or real[1].installed
ports = Ports()