dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/base.py
"""What every use case module shares: the Brick, the permissions, who sees what, today, numbers, claims, neighbours.
Each use case is declared ONCE and served as an API route and as an MCP tool. Two things are decided in the use cases and nowhere else:
WHO MAY (a PERMISSION on the declaration, never a role's name) and WHICH ROWS (`_sees_*`, from the caller's email AND sub: a row the caller
may not see is a 404, never a 403). Each command is one unit of work: load, let the domain decide, save whole at the version it was loaded
(`If-Match` for a screen). A neighbour is reached only through infrastructure/seams.py."""
import dataclasses
import json
from contextlib import contextmanager
from datetime import date, datetime, timezone
from decimal import Decimal
from typing import Optional
from erp_platform import auth, retry, work
from erp_platform.bricks import Brick, Refused
from erp_platform.repository import Conflict, retry_on_conflict
from sqlalchemy.exc import IntegrityError
from brick_purchasing.domain.approval import identities
from brick_purchasing.domain.errors import Invalid, RuleBroken
from brick_purchasing.domain.values import SETTING_DEFS, render_number
from brick_purchasing.infrastructure import business_time
from brick_purchasing.infrastructure.seams import Missing
# Every decision is a PERMISSION (module.yaml permissions.catalog); no role's name is ever compared.
P = "purchasing:"
REQ_WRITE, REQ_READ, REQ_READ_OWN, REQ_APPROVE, REQ_INTAKE = (P + "request:" + x for x in ("write", "read", "read_own", "approve", "intake"))
ORD_WRITE, ORD_READ, ORD_EXPECTED, ORD_APPROVE, ORD_CONFIRM, ORD_SEND = (
P + "order:" + x for x in ("write", "read", "read_expected", "approve", "confirm", "send"))
RCP_WRITE, RCP_READ = P + "receipt:write", P + "receipt:read"
BILL_WRITE, BILL_READ, BILL_APPROVE = P + "bill:write", P + "bill:read", P + "bill:approve"
PAY_WRITE, PAY_READ, PAY_APPROVE = P + "payment:write", P + "payment:read", P + "payment:approve"
SUP_WRITE, SUP_READ, PRICE_WRITE = P + "supplier:write", P + "supplier:read", P + "supplier_price:write"
BANK_WRITE, BANK_APPROVE = P + "supplier_bank:write", P + "supplier_bank:approve"
SETTINGS_WRITE, CONTROLS = P + "settings:write", P + "settings:controls"
MIG_READ, MIG_SIGN, MIG_LOAD, MIG_RECONCILE = P + "migration:read", P + "migration:sign", P + "migration:load", P + "migration:reconcile"
RFQ_WRITE, AGR_WRITE, AGR_APPROVE, SRC_WRITE, EVAL_WRITE = (P + x for x in ("rfq:write", "agreement:write", "agreement:approve", "source:write", "evaluation:write"))
DP_REQUEST, DP_APPROVE, DP_CLEAR = P + "down_payment:request", P + "down_payment:approve", P + "down_payment:clear"
SB_WRITE, SB_APPROVE, ERS_RUN, SUBCONTRACT, CONSIGN_SETTLE = (P + x for x in ("self_billing:write", "self_billing:approve", "ers:run",
"subcontract:provide", "consignment:settle"))
# machine permissions: held by a service client by an EXACT grant, never reached by a wildcard
NUMBER_CLAIM, RECEIPT_HAND, BILL_POST, PAY_POST, PAY_EXPORT = (P + x for x in ("number_series:claim", "receipt:hand_over", "bill:post", "payment:post",
"payment:export"))
AI_READ, AI_CHECK, CONSIGN_INTAKE, SRC_PROPOSE, ASSET_HAND, CIS_RECORD, PAY_CONFIRM = (
P + x for x in ("bill:ai_read", "bill:ai_check", "consignment:intake", "source:propose", "asset:hand_over", "cis:record", "payment:confirm"))
MACHINE = (NUMBER_CLAIM, RECEIPT_HAND, BILL_POST, PAY_POST, PAY_EXPORT, AI_READ, AI_CHECK, MIG_LOAD, MIG_RECONCILE, CONSIGN_INTAKE, SRC_PROPOSE,
ASSET_HAND, CIS_RECORD, REQ_INTAKE, PAY_CONFIRM)
PURCHASING = Brick("purchasing", version="0.1.0", ids_as_strings=True, machine_permissions=MACHINE, instructions=(
"Purchasing & suppliers: agree what you will spend before you spend it (requests, approved by a second person within a limit, in up to three "
"release levels), order in the supplier's own pack and price (an order line keeps its exact conversion and its order price unit), record what "
"actually arrived (receive_goods_receipt is the one way a delivery is recorded; stock moves are inventory's), check what a supplier bills you "
"against the order and the receipts (the three-way match, derived, never stored), and pay it once in an approved run (a file for the bank; "
"'paid' is the bank's word). Every gate answers {gate, met, reason, fix}. Nothing is ever edited: a correction is a credit, a return, a "
"reversing allocation or a new revision."))
PURCHASING.declare_process("process.yaml")
@contextmanager
def _repository_default():
from erp_platform import db # imported late: it needs BUCKET
with _on_engine(db.engine) as repo:
yield repo
def _lost_a_unique_race(exc: IntegrityError) -> bool:
text_ = f"{getattr(exc, 'orig', '')} {exc}"
return "23505" in text_ or "duplicate key" in text_ or "UNIQUE constraint" in text_
@contextmanager
def _on_engine(engine):
"""One transaction. On DataK3 a write conflict on a claim row can surface at COMMIT (the loser's COMMIT raises 23505), after the use case
has returned - so it is caught here, around the commit, and answered as the same retry-safe 409 a conflict seen mid-use-case gets, never a 500."""
from brick_purchasing.infrastructure.tables import Repository
try:
with engine.begin() as conn:
yield Repository(conn)
except IntegrityError as exc:
if not _lost_a_unique_race(exc):
raise
raise Refused(409, "Someone else wrote the same record a moment ago - load it again and retry") from None
def repository():
"""The unit of work's connection: use_cases.repository, looked up per call. erp-platform's in-process composition
(Catalogue, `erp-compose init`'s composition test, `serve` on tier 1 and `--local`) swaps
`<package>.application.use_cases.repository`; every module here reads base.repository, so without this forward the
swap was ignored and a composed purchasing still reached for BUCKET (found after the small-ERP dry run,
2026-10-25, the same defect as erp-brick-inventory 7766e67). Tests may still monkeypatch this name."""
from brick_purchasing.application import use_cases
return use_cases.repository()
def repo_():
"""The repository the module currently points at (tests replace use_cases.repository; this reads it live)."""
import brick_purchasing.application.base as me
return me.repository()
@contextmanager
def _decide():
"""A broken business rule is a 409 with the rule's own words (a gate: its reason and fix); an input outside its bounds is a 422."""
try:
yield
except Invalid as bad:
raise Refused(422, str(bad)) from None
except RuleBroken as broken:
raise Refused(409, str(broken)) from None
def _dup(exc: BaseException) -> bool:
text = " ".join(str(e) for e in _chain(exc)).lower()
return retry.sqlstate(exc) == "23505" or "unique constraint failed" in text or "duplicate key" in text
def _chain(exc):
seen = set()
while exc is not None and id(exc) not in seen:
seen.add(id(exc))
yield exc
exc = getattr(exc, "orig", None) or exc.__cause__ or exc.__context__
@contextmanager
def _guard():
"""A claim or version lost to a racer (a duplicate key, at the statement or at COMMIT) is a Conflict: the unit re-runs on the latest."""
try:
yield
except Refused:
raise
except Conflict:
raise
except Exception as exc: # noqa: BLE001 — re-raised unless a lost claim
if _dup(exc):
raise Conflict() from None
raise
def unit(version, fn):
"""Run one unit of work (load, decide, save, COMMIT). A caller who sent NO version asked for "this change on whatever is there":
a race lost to another writer is re-run on the latest. One who sent a version is told it changed (409) and reloads."""
def run():
with _guard():
return fn()
return run() if version is not None else retry_on_conflict(run)
def _same_version(what: str, current: int, asked: Optional[int]) -> None:
if asked is not None and int(asked) != current:
raise Refused(409, f"{what} changed since you opened it (you have version {asked}, it is at {current}) — load it again")
# ── who ───────────────────────────────────────────────────────────────────────────────────────
def _email(caller: dict) -> str:
return (caller.get("email") or "").strip().lower()
def _sub(caller: dict) -> str:
return (caller.get("sub") or "").strip().lower()
def _who(caller: dict) -> str:
return _email(caller) or _sub(caller)
def _ids(caller: dict) -> frozenset:
return identities(caller.get("email"), caller.get("sub"))
def _stamp() -> datetime:
"""Audit stamps are real time (UTC, as the columns hold it); business dates come from _today."""
return datetime.now(timezone.utc).replace(tzinfo=None)
def _bnow() -> datetime:
return work.now().astimezone(timezone.utc).replace(tzinfo=None)
def _today(repo) -> date:
return business_time.today(repo.settings)
def _day(v, what: str) -> Optional[date]:
try:
return date.fromisoformat(v) if isinstance(v, str) and v else (v or None)
except ValueError:
raise Refused(422, f"{what} is a date like 2026-11-01") from None
def _require_any(caller: dict, *perms: str) -> str:
for p in perms:
if auth.has_permission(caller, p):
return p
raise Refused(403, f"missing permission: {' or '.join(perms)}")
def _currency(code: str) -> str:
if len(code) != 3 or not code.isalpha():
raise Refused(422, f"{code!r} is not an ISO 4217 currency code such as GBP")
return code.upper()
# ── settings ──────────────────────────────────────────────────────────────────────────────────
class Settings:
"""The settings as they stand on a date, typed (a decimal is a Decimal)."""
def __init__(self, repo, on: date):
self.repo, self.on = repo, on
def __getitem__(self, key: str):
v = self.repo.settings.value_on(key, self.on)
d = SETTING_DEFS[key]
return Decimal(str(v)) if d.kind == "decimal" else v
def mode(self, step: str) -> str:
return self.repo.settings.value_on(f"ai.{step}.mode", self.on)
# ── numbers: never reused, never filling a gap below the high-water mark ───────────────────────────
def stamp_number(repo, kind: str, fmt_key: str, on: date, document_id: str) -> str:
"""The next number of a series: the high-water mark moves UP (saved at the version it was loaded, so two claims racing collide), and the
RENDERED number is claimed by a plain INSERT, so a number is held by one document only — a migrated number below the mark is never
handed out again, and a cancelled confirmed order keeps its number."""
fmt = repo.settings.value_on(fmt_key, on)
series_id = f"{kind}|{fmt}"
s, ver = repo.numbers.get(series_id, on.year)
s.format = fmt
while True:
number = render_number(fmt, on.year, s.claim())
if repo.numbers.claim_number(number, series_id, document_id):
break
repo.numbers.save(s, ver)
return number
# ── neighbours ─────────────────────────────────────────────────────────────────────────────────
def _refuse(m: Missing, what: str) -> Refused:
if not m.installed:
return Refused(409, f"{what}: that module is not part of this installation")
gates = "; ".join(f"{g.get('gate')}: {g.get('reason')} (fix: {g.get('fix')})" for g in m.gates)
if m.status in (400, 404, 409, 422):
return Refused(422 if m.status != 409 else 409, f"{what}: {gates or m.reason}")
return Refused(502, f"{what}: {m.reason} — try again")
def _unit_conv(caller: dict, product: dict, unit_code: str) -> Optional[tuple]:
"""(numerator, denominator) of a unit to the product's base unit from product's answer, or None."""
if unit_code == product.get("base_unit_code"):
return 1, 1
for u in product.get("units") or []:
if u.get("unit_code") == unit_code:
return int(u["numerator"]), int(u["denominator"])
return None
def purchase_role_unit(product: dict) -> str:
"""The product's default purchase unit on the date: roles.purchase (a code or {unit_code}), else purchase_unit_code, else the base."""
role = (product.get("roles") or {}).get("purchase")
code = role.get("unit_code") if isinstance(role, dict) else role
return code or product.get("purchase_unit_code") or product.get("base_unit_code") or ""
# ── views ──────────────────────────────────────────────────────────────────────────────────────
def view(obj, *, drop: tuple = ()) -> dict:
"""A dataclass as a dict, nested parts included, without the in-memory event list."""
out = {}
for f in dataclasses.fields(obj):
if f.name == "events" or f.name in drop:
continue
v = getattr(obj, f.name)
if isinstance(v, list):
out[f.name] = [view(x) if dataclasses.is_dataclass(x) else x for x in v]
else:
out[f.name] = v
return out
def j(value) -> str:
return json.dumps(value, default=str)
def money(d) -> str:
return str(Decimal(str(d)).quantize(Decimal("0.01")))