dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
src/brick_purchasing/domain/request.py
"""A purchase request: somebody asked to spend money, and who agreed (PUR-01 to PUR-03, PUR-38, PUR-65).
The aggregate owns its invariants: complete before it is submitted, a second person approves, every release level is a different
person, a refused request raises no order. The use case supplies what the aggregate cannot know (the approver's limit in force
today, the release strategy in force on the submit date)."""
import json
from dataclasses import dataclass, field
from datetime import date
from decimal import Decimal
from typing import Optional
from brick_purchasing.domain import approval
from brick_purchasing.domain.approval import Release, Strategy, identities
from brick_purchasing.domain.errors import RuleBroken
from brick_purchasing.domain.readiness import Gate, no, ok, require
from brick_purchasing.domain.values import ZERO
@dataclass
class RequestLine:
purchase_request_id: str
line_no: int
product_id: str = ""
description: str = ""
quantity: Decimal = ZERO
unit_code: str = ""
unit_price: Decimal = ZERO
net_amount: Decimal = ZERO
bp_id: Optional[int] = None # the supplier the requester has in mind (optional)
needed_by: Optional[date] = None
@dataclass
class PurchaseRequest:
purchase_request_id: str
requested_by_email: str = ""
requested_by_sub: str = ""
cost_centre: str = ""
category: str = "general"
reason: str = ""
amount: Decimal = ZERO
currency: str = "GBP"
status: str = "draft"
needed_by: Optional[date] = None
submitted_on: Optional[date] = None
approved_by_email: str = ""
approved_by_sub: str = ""
approved_on: Optional[date] = None
approval_note: str = ""
refused_reason: str = ""
closed_reason: str = ""
release_levels: int = 1
release_thresholds: str = "[]"
strategy_valid_from: Optional[date] = None
source_kind: str = "person" # person | sales_order | replenishment
request_ref: str = ""
ship_to_bp_id: Optional[int] = None
ship_to_address: str = ""
sales_order_id: str = ""
so_line_no: Optional[int] = None
warehouse_id: str = ""
location_id: str = ""
order_id: str = "" # the order raised from it
ship_to_address_lines: str = "" # v8: the address object (shared.yaml conventions.address): lines as a JSON array, then the parts
ship_to_address_city: str = ""
ship_to_address_postcode: str = ""
ship_to_address_region: str = ""
ship_to_address_country: str = ""
cancelled_by_email: str = "" # v8 (PUR-67): who, when and why
cancelled_by_sub: str = ""
cancelled_on: Optional[date] = None
cancel_reason: str = ""
lines: list = field(default_factory=list, repr=False)
releases: list = field(default_factory=list, repr=False)
events: list = field(default_factory=list, repr=False)
@property
def owner_ids(self) -> frozenset:
return identities(self.requested_by_email, self.requested_by_sub)
# ── readiness ──────────────────────────────────────────────────────────────────────────
def complete_gate(self) -> Gate:
missing = [w for w, bad in (("a line", not self.lines), ("an amount", self.amount <= 0),
("a cost centre", not self.cost_centre), ("a reason", not self.reason.strip())) if bad]
return ok("complete") if not missing else no("complete", f"the request has no {', '.join(missing)}", "fill in what is missing")
def readiness(self) -> list:
return [self.complete_gate()]
# ── transitions ─────────────────────────────────────────────────────────────────────────
def set_lines(self, lines: list) -> None:
if self.status != "draft":
raise RuleBroken("only a draft request can be changed")
self.lines = lines
self.amount = sum((x.net_amount for x in lines), ZERO)
def submit(self, today: date, strategy: Strategy) -> None:
if self.status != "draft":
raise RuleBroken(f"a {self.status} request cannot be submitted")
require([self.complete_gate()])
self.status, self.submitted_on = "awaiting_approval", today
self.release_levels, self.strategy_valid_from = strategy.levels, strategy.valid_from
self.release_thresholds = json.dumps([str(t) for t in strategy.thresholds])
self.events.append(("request_submitted", self.purchase_request_id))
def approve_level(self, *, email, sub, limit, today) -> Release:
self._awaiting()
return approval.step(self, "request", self.purchase_request_id, 1, owner_ids=self.owner_ids, email=email, sub=sub, limit=limit,
needed=self.amount, today=today, noun="request", final=False)
def approve(self, *, email, sub, limit, today, note: str = "") -> None:
self._awaiting()
approval.step(self, "request", self.purchase_request_id, 1, owner_ids=self.owner_ids, email=email, sub=sub, limit=limit,
needed=self.amount, today=today, noun="request", final=True)
self.status, self.approved_on, self.approval_note = "approved", today, note or ""
self.approved_by_email, self.approved_by_sub = (email or "").lower(), (sub or "").lower()
self.events.append(("request_approved", self.purchase_request_id))
def refuse(self, reason: str) -> None:
self._awaiting()
if not reason.strip():
raise RuleBroken("a refusal needs a reason")
self.status, self.refused_reason = "refused", reason.strip()
def close(self, reason: str) -> None:
if self.status != "approved":
raise RuleBroken(f"only an approved request can be closed without ordering (this one is {self.status})")
if not reason.strip():
raise RuleBroken("closing a request needs a reason")
self.status, self.closed_reason = "closed", reason.strip()
def cancel(self, reason: str, *, email: str = "", sub: str = "", today: Optional[date] = None, with_order: bool = False,
from_sales_order: bool = False) -> None:
"""v8 (PUR-67): cancel_request (draft, awaiting approval or approved), cancel_request_from_sales_order (awaiting approval or approved) and
cancel_with_order (ordered, only because its order was cancelled). Who, when and why are kept; the request leaves the approval queue."""
allowed = ("ordered",) if with_order else (("awaiting_approval", "approved") if from_sales_order else ("draft", "awaiting_approval", "approved"))
if self.status not in allowed:
raise RuleBroken(f"only a request that is {' or '.join(allowed).replace('_', ' ')} can be cancelled this way (this one is {self.status})")
if not (reason or "").strip():
raise RuleBroken("cancelling a request needs a reason")
self.status, self.cancel_reason, self.closed_reason = "cancelled", reason.strip(), reason.strip()
self.cancelled_by_email, self.cancelled_by_sub, self.cancelled_on = (email or "").lower(), (sub or "").lower(), today
self.events.append(("request_cancelled", self.purchase_request_id))
def mark_ordered(self, order_id: str) -> None:
if self.status not in ("approved", "ordered"):
raise RuleBroken("an order can only be raised from an approved request: this one is " + self.status)
self.status, self.order_id = "ordered", self.order_id or order_id
def _awaiting(self) -> None:
if self.status != "awaiting_approval":
raise RuleBroken(f"the request is {self.status}, not awaiting approval")
def waiting(self) -> str:
return approval.stage_note(self.release_levels, len(self.releases)) if self.status == "awaiting_approval" else ""