dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/application/requests.py
"""Requests: asking for what you need, and who agreed (PUR-01, PUR-02, PUR-03, PUR-38, PUR-65).
A request is routed to the approvers whose limit in force TODAY covers it; the approver is never the requester (email AND sub); up to three
release levels apply, the strategy in force on the SUBMIT date. Sales-order and inventory raise requests here, never orders (PUR-65)."""
import json
import uuid
from typing import Optional, Union
from erp_platform import auth, work
from erp_platform.bricks import Refused
from erp_platform.types import BigId, IsoDate, Limit, text
from brick_purchasing.application import base
from brick_purchasing.application.base import (
PURCHASING,
REQ_APPROVE,
REQ_INTAKE,
REQ_READ,
REQ_READ_OWN,
REQ_WRITE,
_decide,
_ids,
_same_version,
_sub,
_today,
_who,
unit,
view,
)
from brick_purchasing.application.inputs import AddressIn, Code, Currency, Offset, OptCode, Qty, Reason, Ref, RequestLineIn, Version
from brick_purchasing.domain import approval
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.request import PurchaseRequest, RequestLine
from brick_purchasing.domain.values import ZERO, money_round
from brick_purchasing.infrastructure.seams import Missing, ports
repository = lambda: base.repository() # noqa: E731
# ── who sees what ─────────────────────────────────────────────────────────────────────────────
def _sees(caller: dict, r: PurchaseRequest) -> bool:
"""request:read sees every request; request:read_own sees the caller's own (email OR sub). A row the caller may not see is a 404."""
if auth.has_permission(caller, REQ_READ):
return True
return auth.has_permission(caller, REQ_READ_OWN) and bool(_ids(caller) & r.owner_ids)
def _load(repo, caller: dict, request_id: str) -> tuple:
found = repo.requests.get(request_id)
if found is None or not _sees(caller, found[0]):
raise Refused(404, f"no purchase request {request_id}")
return found
def request_view(r: PurchaseRequest, version: int, *, readiness: bool = True) -> dict:
out = view(r)
out["version"] = version
out["waiting"] = r.waiting()
out["approved_levels"] = len(r.releases)
if readiness:
from brick_purchasing.domain.readiness import views
out["readiness"] = views(r.readiness())
return out
def _lines(rows: list, request_id: str, currency: str) -> list:
out = []
for n, x in enumerate(rows, 1):
if not (x.product_id or (x.description or "").strip()):
raise Refused(422, f"line {n}: name a product or describe what is wanted")
out.append(RequestLine(request_id, n, x.product_id or "", (x.description or x.product_id or "").strip(), x.quantity, x.unit_code or "",
x.unit_price, money_round(x.quantity * x.unit_price, currency), x.bp_id, base._day(x.needed_by, "needed_by")))
return out
# ── the person's side ─────────────────────────────────────────────────────────────────────────
@PURCHASING.use_case(kind="command", route="POST /purchase-requests", permission=REQ_WRITE)
def draft_request(caller: dict, cost_centre: Code, reason: Reason, lines: list[RequestLineIn], currency: Currency = "GBP", category: OptCode = "general",
needed_by: Optional[IsoDate] = None) -> dict:
"""Ask for what you need: a cost centre, a reason and the lines (a product or a description, how many, an estimated price). It is a draft
until you submit it; the request id is derived from you and the moment you asked."""
if not lines:
raise Refused(422, "a request needs at least one line")
cur = base._currency(currency)
def run():
with repository() as repo, _decide():
rid = f"{_who(caller)}:{work.now().strftime('%Y%m%dT%H%M%S%f')}-{uuid.uuid4().hex[:6]}"
r = PurchaseRequest(rid, _who(caller), _sub(caller), cost_centre.strip(), (category or "general").strip(), reason.strip(), ZERO, cur,
needed_by=base._day(needed_by, "needed_by"))
r.set_lines(_lines(lines, rid, cur))
v = repo.requests.save(r, 0)
return request_view(r, v)
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{purchase_request_id}/revise", permission=REQ_WRITE)
def revise_request(caller: dict, purchase_request_id: Ref, lines: Optional[list[RequestLineIn]] = None, cost_centre: Optional[Code] = None,
reason: Optional[Reason] = None, version: Optional[Version] = None) -> dict:
"""Change a request you are still drafting: its lines, cost centre or reason. A submitted request is not edited."""
def run():
with repository() as repo, _decide():
r, v = _load(repo, caller, purchase_request_id)
_same_version("the request", v, version)
if not (_ids(caller) & r.owner_ids):
raise Refused(404, f"no purchase request {purchase_request_id}")
if lines is not None:
r.set_lines(_lines(lines, r.purchase_request_id, r.currency))
elif r.status != "draft":
raise RuleBroken("only a draft request can be changed")
if cost_centre:
r.cost_centre = cost_centre.strip()
if reason:
r.reason = reason.strip()
return request_view(r, repo.requests.save(r, v), readiness=True)
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{purchase_request_id}/submit", permission=REQ_WRITE, idempotent=True)
def submit_request(caller: dict, purchase_request_id: Ref, version: Optional[Version] = None) -> dict:
"""Submit the request: it needs a line, an amount, a cost centre and a reason (the readiness answer says which is missing). It is routed to the
approvers whose limit in force today covers it, under the release strategy in force today."""
def run():
with repository() as repo, _decide():
r, v = _load(repo, caller, purchase_request_id)
if r.status == "awaiting_approval":
return request_view(r, v)
_same_version("the request", v, version)
if not (_ids(caller) & r.owner_ids):
raise Refused(404, f"no purchase request {purchase_request_id}")
today = _today(repo)
r.submit(today, approval.strategy_for(repo.strategies.all(), r.category, r.amount, r.currency, today))
return request_view(r, repo.requests.save(r, v))
return unit(version, run)
def _limit(repo, caller: dict, category: str, currency: str, today):
return approval.limit_in_force(repo.limits.all(), _ids(caller), category, currency, today)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{purchase_request_id}/approve-level", permission=REQ_APPROVE)
def approve_request_level(caller: dict, purchase_request_id: Ref, version: Optional[Version] = None) -> dict:
"""Approve one NON-FINAL release level of a request (the strategy in force on the submit date asks for more than one). Your limit meets the
level's threshold; you are not the requester and approved no earlier level (email AND sub). The request stays awaiting approval."""
def run():
with repository() as repo, _decide():
r, v = _load(repo, caller, purchase_request_id)
_same_version("the request", v, version)
today = _today(repo)
r.approve_level(email=caller.get("email"), sub=caller.get("sub"), limit=_limit(repo, caller, r.category, r.currency, today),
today=today)
return request_view(r, repo.requests.save(r, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{purchase_request_id}/approve", permission=REQ_APPROVE)
def approve_request(caller: dict, purchase_request_id: Ref, note: Optional[Reason] = None, version: Optional[Version] = None) -> dict:
"""Approve the spend (the FINAL level): the amount is within your limit in force today for this category, and you are not the requester nor an
earlier level's approver (same email OR same sub). Records who, today in the business time zone, and your note; a buyer can then raise an order."""
def run():
with repository() as repo, _decide():
r, v = _load(repo, caller, purchase_request_id)
_same_version("the request", v, version)
today = _today(repo)
r.approve(email=caller.get("email"), sub=caller.get("sub"), limit=_limit(repo, caller, r.category, r.currency, today), today=today,
note=note or "")
return request_view(r, repo.requests.save(r, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{purchase_request_id}/refuse", permission=REQ_APPROVE, idempotent=True)
def refuse_request(caller: dict, purchase_request_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""Refuse a request with a reason the requester can see. No order can be raised from it."""
def run():
with repository() as repo, _decide():
r, v = _load(repo, caller, purchase_request_id)
if r.status == "refused":
return request_view(r, v)
_same_version("the request", v, version)
r.refuse(reason)
return request_view(r, repo.requests.save(r, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{purchase_request_id}/close", permission=base.ORD_WRITE, idempotent=True)
def close_request(caller: dict, purchase_request_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""Close an approved request without ordering, with a reason."""
def run():
with repository() as repo, _decide():
r, v = _load(repo, caller, purchase_request_id)
if r.status == "closed":
return request_view(r, v)
_same_version("the request", v, version)
r.close(reason)
return request_view(r, repo.requests.save(r, v))
return unit(version, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{purchase_request_id}/cancel-request", permission=None, idempotent=True)
def cancel_request(caller: dict, purchase_request_id: Ref, reason: Reason, version: Optional[Version] = None) -> dict:
"""PUR-67: the requester cancels their own request, or a buyer any request, with a reason, before it is ordered. It leaves the approval queue.
An ordered request is never cancelled on its own (422 {gate: not_ordered}): the order is cancelled and the request follows it."""
base._require_any(caller, REQ_WRITE, base.ORD_WRITE)
def run():
with repository() as repo, _decide():
r, v = _load(repo, caller, purchase_request_id)
if r.status == "cancelled":
return request_view(r, v)
buyer = auth.has_permission(caller, base.ORD_WRITE)
if not buyer and not (_ids(caller) & r.owner_ids):
raise Refused(404, f"no purchase request {purchase_request_id}")
if r.status == "ordered":
order = repo.orders.get(r.order_id)
named = f"{order[0].order_no or r.order_id} at {order[0].bp_id}" if order else r.order_id
raise Refused(422, f"not_ordered: the request is ordered as {named}", gate="not_ordered",
fix="cancel the purchase order (PUR-08); the request follows it")
_same_version("the request", v, version)
r.cancel(reason, email=_who(caller), sub=_sub(caller), today=_today(repo))
return request_view(r, repo.requests.save(r, v))
return unit(version, run)
@PURCHASING.use_case(kind="query", route="GET /purchase-requests/{purchase_request_id}", permission=None, etag="version")
def get_purchase_request(caller: dict, purchase_request_id: Ref) -> dict:
"""One request, with its lines, its approvals so far, what is waiting and the readiness gates — if the caller may see it (a requester sees
their own; request:read sees every one)."""
base._require_any(caller, REQ_READ, REQ_READ_OWN)
with repository() as repo:
r, v = _load(repo, caller, purchase_request_id)
return request_view(r, v)
@PURCHASING.use_case(kind="query", route="GET /purchase-requests", permission=None)
def list_purchase_requests(caller: dict, status: Optional[Code] = None, awaiting_my_approval: bool = False, limit: Limit = 50, offset: Offset = 0) -> dict:
"""The requests the caller may see: their own (request:read_own) or every one (request:read). `awaiting_my_approval` lists those waiting for
an approver whose limit in force today covers them, that the caller did not ask for."""
base._require_any(caller, REQ_READ, REQ_READ_OWN)
with repository() as repo:
mine = None if auth.has_permission(caller, REQ_READ) else (_ids(caller) or frozenset({"\x00nobody"}))
rows, total = repo.requests.page(ids=mine, status="awaiting_approval" if awaiting_my_approval else status, limit=limit, offset=offset)
if awaiting_my_approval:
today = _today(repo)
lim = repo.limits.all()
rows = [r for r in rows if not (_ids(caller) & r.owner_ids)
and (approval.limit_in_force(lim, _ids(caller), r.category, r.currency, today) or ZERO) >= r.amount]
return {"total": total, "requests": [{k: getattr(r, k) for k in ("purchase_request_id", "cost_centre", "category", "reason", "amount", "currency",
"status", "requested_by_email", "submitted_on", "approved_on")} for r in rows]}
# ── raised by other modules (PUR-65): requests, never orders ────────────────────────────────────
def _purchasable(caller: dict, product_id: str, unit_code: str) -> None:
"""The product can be bought: it exists (and, where product is composed, is offered). Refused 422 {gate: purchasable, reason, fix}."""
if not ports.installed("product"):
return
try:
p = ports.product.get(caller, product_id)
except Missing as m:
if m.status == 404:
raise Refused(422, f"purchasable: {product_id} is not a product we can buy ({m.reason}); fix: add it to the catalogue") from None
raise base._refuse(m, f"product {product_id}") from None
if base._unit_conv(caller, p, unit_code) is None:
raise Refused(422, f"purchasable: {product_id} is not counted in {unit_code}; fix: Product: add the unit")
def _address(caller: dict, ship_to_bp_id, given) -> dict:
"""The ship-to as the one address shape. v8: a string is accepted until the next major as {lines: [the string], city: '', country: the ship-to
partner's country} (shared.yaml conventions.address, legacy)."""
if given is None or given == "":
return {"lines": [], "city": "", "postcode": "", "region": "", "country": "", "text": ""}
if isinstance(given, str):
country = ""
if ship_to_bp_id and ports.installed("bp"):
try:
country = str(ports.bp.get(caller, ship_to_bp_id).get("country") or "")
except Missing:
country = ""
return {"lines": [given[:70]], "city": "", "postcode": "", "region": "", "country": country, "text": given}
return {"lines": list(given.lines), "city": given.city, "postcode": given.postcode or "", "region": given.region or "", "country": given.country,
"text": ", ".join([*given.lines, given.city, given.postcode or "", given.country]).replace(", ,", ",")}
def _intake(caller: dict, request_ref: str, make) -> dict:
def run():
with repository() as repo, _decide():
got = repo.requests.by_ref(request_ref)
if got: # PUR-65: {request_ref, request_id, status}, the same for a repeat. (The spec's 201 for a new one
return {"request_ref": request_ref, "request_id": got[0].purchase_request_id, "status": got[0].status}
r = make(repo) # is the platform's to send: it answers 200; a private _status field leaked into the body.)
repo.requests.save(r, 0)
return {"request_ref": request_ref, "request_id": r.purchase_request_id, "status": r.status}
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/from-sales-order", permission=REQ_INTAKE, idempotent=True, tool=False)
def intake_request_from_sales_order(caller: dict, request_ref: text(200, min_length=1), product_id: Code, quantity: Qty, unit_code: Code,
sales_order_id: Ref, so_line_no: int, ship_to_bp_id: Optional[BigId] = None, ship_to_address: Optional[Union[AddressIn, text(1000)]] = None,
needed_by: Optional[IsoDate] = None) -> dict:
"""Sales-order raises a purchase REQUEST for a confirmed third-party line (a drop-ship). Keyed on request_ref (so:{sales_order_id}:{line_no}@r{revision});
a repeat answers the same request. The request is awaiting approval and follows approve_request and raise_order like one a person raised."""
_purchasable(caller, product_id, unit_code)
addr = _address(caller, ship_to_bp_id, ship_to_address)
def make(repo):
rid = request_ref
r = PurchaseRequest(rid, "sales-order", "", "", "general", f"drop-ship for {sales_order_id} line {so_line_no}", ZERO, "GBP", "awaiting_approval",
needed_by=base._day(needed_by, "needed_by"), submitted_on=_today(repo), source_kind="sales_order", request_ref=request_ref,
ship_to_bp_id=ship_to_bp_id, ship_to_address=addr["text"], sales_order_id=sales_order_id, so_line_no=so_line_no,
ship_to_address_lines=json.dumps(addr["lines"]), ship_to_address_city=addr["city"], ship_to_address_postcode=addr["postcode"],
ship_to_address_region=addr["region"], ship_to_address_country=addr["country"])
r.lines = [RequestLine(rid, 1, product_id, product_id, quantity, unit_code, ZERO, ZERO, None, r.needed_by)]
today = _today(repo)
s = approval.strategy_for(repo.strategies.all(), r.category, r.amount, r.currency, today)
r.release_levels, r.strategy_valid_from = s.levels, s.valid_from
return r
return _intake(caller, request_ref, make)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/{request_id}/cancel", permission=REQ_INTAKE, idempotent=True, tool=False)
def cancel_request_from_sales_order(caller: dict, request_id: Ref, reason: Reason) -> dict:
"""Sales-order withdraws a request it raised (request_ref so:...) when its own order line is cancelled: the request leaves the
approval queue. A repeat answers the same. 403 for a request sales-order did not raise; 422 {gate: not_cancellable, reason, fix}
once the request has become an order (it is cancelled through that order: the refusal names the order and the supplier)."""
def run():
with repository() as repo, _decide():
found = repo.requests.get(request_id)
if found is None:
raise Refused(404, f"no purchase request {request_id}")
r, v = found
if r.source_kind != "sales_order":
raise Refused(403, f"request {request_id} was not raised by sales-order: it is not yours to cancel")
if r.status == "cancelled":
return {"request_id": request_id, "status": "cancelled"}
if r.status == "ordered":
order = repo.orders.get(r.order_id)
named = f"order {order[0].order_no or r.order_id} of supplier {order[0].bp_id}" if order else f"order {r.order_id}"
raise Refused(422, f"request {request_id} is ordered as {named}: cancel the order", gate="not_cancellable",
fix="the buyer cancels the purchase order; then cancel again")
if r.status not in ("awaiting_approval", "approved"):
raise Refused(422, f"request {request_id} is {r.status}: it can no longer be cancelled", gate="not_cancellable",
fix="nothing to withdraw")
r.cancel(reason, email="sales-order", sub="sales-order", today=_today(repo), from_sales_order=True)
repo.requests.save(r, v)
return {"request_id": request_id, "status": "cancelled"}
return unit(None, run)
@PURCHASING.use_case(kind="command", route="POST /purchase-requests/from-replenishment", permission=REQ_INTAKE, idempotent=True, tool=False)
def intake_request_from_replenishment(caller: dict, request_ref: text(200, min_length=1), product_id: Code, quantity: Qty, unit_code: Code,
warehouse_id: Code, location_id: Optional[Code] = None, needed_by: Optional[IsoDate] = None,
reason: Optional[Reason] = None) -> dict:
"""Inventory's replenishment raises a purchase REQUEST for a warehouse's goods-in location. Keyed on request_ref
(inv:repl:{warehouse_id}:{product_id}:{business_date}); a repeat answers the same request."""
_purchasable(caller, product_id, unit_code)
def make(repo):
rid = request_ref
r = PurchaseRequest(rid, "inventory", "", "", "general", reason or f"replenishment of {product_id} at {warehouse_id}", ZERO, "GBP",
"awaiting_approval", needed_by=base._day(needed_by, "needed_by"), submitted_on=_today(repo), source_kind="replenishment",
request_ref=request_ref, warehouse_id=warehouse_id, location_id=location_id or "")
r.lines = [RequestLine(rid, 1, product_id, product_id, quantity, unit_code, ZERO, ZERO, None, r.needed_by)]
today = _today(repo)
s = approval.strategy_for(repo.strategies.all(), r.category, r.amount, r.currency, today)
r.release_levels, r.strategy_valid_from = s.levels, s.valid_from
return r
return _intake(caller, request_ref, make)