dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/returns.py
"""Customer returns against an RMA (INV-14): sales-order authorises the return (POST /expected-returns), goods-in receives it
into quality stock at the cost the goods left at, the inspector decides every unit (restock, scrap, back to the customer), and
sales-order hears of the receipt and of the inspection, once each."""
import json
import uuid
from decimal import Decimal
from typing import Optional
from erp_platform.bricks import Refused
from brick_inventory.application import docs, ledger, postings, products
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import need_read, qstr
from brick_inventory.application.deliveries import _in_unit, _merge
from brick_inventory.application.inputs import Code, ExpectedReturnLine, InspectionIn, Ref, ReturnLineIn
from brick_inventory.domain import location as L
from brick_inventory.domain.errors import BadInput, GateBroken, RuleBroken
from brick_inventory.domain.moves import MoveSpec
from brick_inventory.domain.values import NO_LOT, OURS, UNRESTRICTED, ZERO, money
from brick_inventory.infrastructure import seams
QUALITY = "quality"
DECISIONS = ("restock", "scrap", "back_to_customer")
def _rma_rows(unit, rma_ref: str) -> list:
return unit.repo.find("inv_expected_return", order=("line_no",), rma_ref=rma_ref)
@INV.use_case(kind="command", route="POST /expected-returns", permission="inv:return:expect", idempotent=True, tool=False)
def expect_return(caller: dict, rma_ref: Ref, bp_id: int, warehouse_id: Code, lines: list[ExpectedReturnLine]) -> dict:
"""Sales-order authorises a return (an RMA): what the customer may send back, per line. Idempotent on rma_ref; a customer
return is received only against an open RMA."""
@U.decide
def body(unit):
unit.warehouse(warehouse_id)
existing = _rma_rows(unit, rma_ref)
if existing:
return {"rma_ref": rma_ref, "expected": [{"line_no": r["line_no"], "quantity_base": qstr(r["quantity_base"])} for r in existing],
"replay": True}
out = []
for ln in lines:
conv = products.convert(unit, ln.product_id, ln.quantity, ln.unit_code, document_ref=rma_ref, what="the expected return")
unit.put("inv_expected_return", {
"rma_ref": rma_ref, "line_no": ln.line_no, "sales_order_id": rma_ref.removeprefix("so:"), "bp_id": bp_id,
"warehouse_id": warehouse_id, "product_id": ln.product_id, "unit_code": ln.unit_code, "quantity_base": conv.base,
"original_delivery_ref": ln.original_delivery_ref, "serials": json.dumps(ln.serials or []),
"inspect": True if ln.inspect is None else ln.inspect, "received_base": ZERO, "status": "open"},
("rma_ref", "line_no"))
out.append({"line_no": ln.line_no, "quantity_base": qstr(conv.base)})
return {"rma_ref": rma_ref, "expected": out, "replay": False}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /expected-returns/{rma_ref}/cancel", permission="inv:return:expect", idempotent=True, tool=False)
def cancel_expected_return(caller: dict, rma_ref: Ref) -> dict:
"""Withdraw an RMA while nothing of it is received (INV-14, sales-order's seam): every line is cancelled, a repeat answers the same,
a partly received one is 422 gate rma_received, and a cancelled RMA lets no return be received against it (gate rma_open)."""
@U.decide
def body(unit):
rows = _rma_rows(unit, rma_ref)
if not rows:
raise BadInput(f"no RMA {rma_ref}")
if any(r["received_base"] > 0 for r in rows):
raise GateBroken("rma_received", f"{rma_ref} is partly received: it can no longer be cancelled",
"receive the rest of the return, or let it stand", status=422)
for r in rows:
unit.put("inv_expected_return", {**r, "status": "cancelled"}, ("rma_ref", "line_no"))
return {"rma_ref": rma_ref, "status": "cancelled"}
return U.run(caller, body)
def _original_cost(unit, ref: Optional[str], product_id: str) -> Optional[Decimal]:
"""What one base unit cost when it left on the original delivery (its own moves), or None when the delivery is not found."""
if not ref:
return None
d = unit.repo.get("inv_document", document_id=ref) or next(iter(unit.repo.find("inv_document", number=ref)), None)
if d is None:
return None
moves = unit.repo.moves_of(document_id=d["document_id"], product_id=product_id)
qty = sum((m["quantity_base"] for m in moves), ZERO)
if not qty:
return None
return -sum((m["value_amount"] for m in moves), ZERO) / qty
def _delivered_serial(unit, product_id: str, serial: str, bp_id: int) -> bool:
for m in unit.repo.moves_of(product_id=product_id, lot_code=serial):
d = unit.repo.get("inv_document", document_id=m["document_id"])
if d and d["kind"] == "delivery" and d["bp_id"] == bp_id:
return True
return False
def _pieces(line: dict) -> list:
"""The (lot or serial, base quantity) pieces a return line was received in."""
note = line.get("note") or ""
serials = json.loads(note) if note.startswith("[") else []
if serials:
return [(s, Decimal(1)) for s in serials]
return [(line["lot_code"] or NO_LOT, line["quantity_base"])]
@INV.use_case(kind="command", route="POST /returns", permission="inv:receipt:validate", idempotent=True)
def receive_customer_return(caller: dict, bp_id: int, warehouse_id: Code, lines: list[ReturnLineIn], rma_ref: Optional[Ref] = None,
location_id: Optional[Code] = None, request_id: Optional[Ref] = None) -> dict:
"""Take back what a customer returns, against an open RMA of that customer (`lines[].line_no` is the RMA line). The goods
go CUSTOMER -> the return location (default WH/QC) in stock type quality, valued at the cost they left at on the original
delivery - not today's average - and the returns order records the receipt (record_delivery, source inventory). More than the
RMA expects, a parcel with no open RMA, or a serial never delivered to that customer is refused and nothing is written."""
@U.decide
def body(unit):
wh = unit.warehouse(warehouse_id)
doc_id = f"ret:{warehouse_id}:{request_id or uuid.uuid4().hex[:12]}"
old = unit.repo.get("inv_document", document_id=doc_id)
if old:
return {"document_id": doc_id, "number": old["number"], "replay": True}
company = wh["company_code"]
target = location_id or f"{warehouse_id}/QC"
unit.location(target)
stype = QUALITY if (unit.location(target)["stock_status"] == QUALITY) else UNRESTRICTED
customer = L.virtual_id(company, "CUSTOMER")
number = docs.claim_number(unit, warehouse_id, "return", doc_id)
delivery_ref = f"inv:{number}"
posted, got = [], []
used_rma: set = set()
asked = rma_ref
for i, ln in enumerate(lines, start=1):
row = None
if asked:
row = {r["line_no"]: r for r in _rma_rows(unit, asked)}.get(ln.line_no)
elif ln.product_id:
row = next((r for r in unit.repo.find("inv_expected_return", order=("rma_ref", "line_no"), bp_id=bp_id, product_id=ln.product_id,
status="open") if r["warehouse_id"] == warehouse_id and r["received_base"] < r["quantity_base"]), None)
if row is None:
raise GateBroken("rma_open", f"no open RMA of {bp_id} names {ln.product_id or ('line ' + str(ln.line_no) + ' of ' + str(asked))}",
"ask sales-order for an open RMA of this customer")
used_rma.add(row["rma_ref"])
if row["status"] != "open" or row["bp_id"] != bp_id or row["warehouse_id"] != warehouse_id:
raise GateBroken("rma_open", f"no open RMA of {bp_id} names {row['product_id']}",
"ask sales-order for an open RMA of this customer")
conv = products.convert(unit, row["product_id"], ln.quantity, ln.unit_code, document_ref=doc_id, what="the returned quantity")
if row["received_base"] + conv.base > row["quantity_base"]:
raise RuleBroken(f"{row['rma_ref']} line {ln.line_no} expects {qstr(row['quantity_base'] - row['received_base'])} "
f"{unit.base_unit(row['product_id'])} more, {qstr(conv.base)} came: receiving more than the RMA expects is refused")
serials = ln.serials or []
known = set(json.loads(row["serials"] or "[]"))
for s in serials:
if s not in known and not _delivered_serial(unit, row["product_id"], s, bp_id):
raise GateBroken("serial_was_delivered", f"serial {s} of {row['product_id']} was never delivered to customer {bp_id}",
"receive only serials this customer was delivered (GET /serials/{product_id}/{serial})")
cost = _original_cost(unit, row["original_delivery_ref"], row["product_id"])
pieces = [(s, Decimal(1)) for s in serials] if serials else [(NO_LOT, conv.base)]
if serials and Decimal(len(serials)) != conv.base:
raise RuleBroken(f"{len(serials)} serial numbers for {qstr(conv.base)} {unit.base_unit(row['product_id'])}")
value_total = money(conv.base * cost) if cost is not None else None
value_left = value_total
for j, (lot, q) in enumerate(pieces, start=1):
value = None if value_total is None else (value_left if j == len(pieces) else money(q * cost))
value_left = None if value_total is None else value_left - value
spec = MoveSpec(doc_id, ln.line_no, row["product_id"], q, customer, target, unit.today, row["unit_code"], None,
1, 1, lot, UNRESTRICTED, stype, OURS, OURS, n=j, value_in=value, reason=f"return against {row['rma_ref']}")
posted.append(ledger.post_move(unit, spec))
docs.stage_line(unit, doc_id, i, line_kind="return", parent_line_no=ln.line_no, product_id=row["product_id"],
unit_code=ln.unit_code, unit_numerator=conv.numerator, unit_denominator=conv.denominator,
quantity_expected=conv.quantity, quantity_done=conv.quantity, quantity_base=conv.base,
to_location_id=target, source_line_ref=row["rma_ref"], stock_type=stype, note=json.dumps(serials) if serials else None,
value_amount=sum((p.row["value_amount"] for p in posted[-len(pieces):]), ZERO),
**products.measures(unit.product(row["product_id"]), ln.unit_code))
unit.put("inv_expected_return", {**row, "received_base": row["received_base"] + conv.base}, ("rma_ref", "line_no"))
got.append((row, conv))
docs.stage_document(unit, doc_id, "return", warehouse_id, company, number=number, lifecycle="done", done_on=unit.today,
done_by=U.who(caller), source_kind="rma", source_ref=",".join(sorted(used_rma)), bp_id=bp_id, rma_ref=sorted(used_rma)[0])
by_class: dict = {}
for pm in posted:
vc = unit.product(pm.row["product_id"]).get("valuation_class") or "*"
by_class[vc] = by_class.get(vc, ZERO) + pm.row["value_amount"]
jl = []
for vc, v in by_class.items():
jl += postings.movement_lines(v, "cost_of_goods_sold", vc, memo=f"return {number}")
postings.stage(unit, doc_id, "return", company, unit.today, f"return {number}", _merge(jl))
unit.flush_races()
recorded = []
so_id = got[0][0]["sales_order_id"] if got else None
if seams.sales_installed():
for row, conv in got:
qty = _in_unit(conv.base, conv.numerator, conv.denominator) if row["unit_code"] == conv.unit_code else conv.quantity
try:
seams.record_delivery(row["sales_order_id"], {"line_no": row["line_no"], "quantity": qstr(qty), "source": "inventory",
"delivery_ref": delivery_ref, "delivered_on": unit.today.isoformat()})
except seams.Missing as m:
raise GateBroken("order_accepts", f"the returns order refused the receipt: {m.text}", "receive no more than the RMA expects", 409) from None
recorded.append((row["line_no"], qty))
from brick_inventory.application.deliveries import _undo_order
unit.compensations.append(lambda rec=list(recorded), so=so_id, ref=delivery_ref, today=unit.today: _undo_order(so, ref, rec, today))
return {"document_id": doc_id, "number": number, "delivery_ref": delivery_ref, "lifecycle": "done", "replay": False,
"lines": [{"line_no": r["line_no"], "quantity_base": qstr(c.base), "location_id": target, "stock_type": stype} for r, c in got],
"moves": [{"move_id": p.row["move_id"], "from_location_id": p.row["from_location_id"], "to_location_id": p.row["to_location_id"],
"quantity_base": qstr(p.row["quantity_base"]), "value_amount": f"{p.row['value_amount']:.2f}"} for p in posted]}
return U.run(caller, body, lost_message=lambda repo, row=None: "another writer took the same number at the same moment")
@INV.use_case(kind="command", route="POST /returns/{document_id}/inspect", permission="inv:receipt:validate", idempotent=True)
def inspect_return(caller: dict, document_id: Ref, decisions: list[InspectionIn]) -> dict:
"""Decide every unit of a returned line: restock (quality -> unrestricted bin, no value change; `to_location_id` is the bin
named in the reason-less default WH/A/01 unless the line says otherwise), scrap (a loss, purpose scrap) or back to the
customer (at the value it came in). Every unit must be decided; accepted (restock + scrap) and rejected (back to customer) are
reported to sales-order once per line."""
@U.decide
def body(unit):
d = unit.repo.get("inv_document", document_id=document_id)
if d is None or d["kind"] != "return":
raise Refused(404, f"no return {document_id}")
done = unit.repo.find("inv_inspection", document_id=document_id)
lines = {l["line_no"]: l for l in unit.repo.find("inv_document_line", order=("line_no",), document_id=document_id) if l["line_kind"] == "return"}
if done:
return {"document_id": document_id, "replay": True, "lines": sorted({r["line_no"] for r in done})}
wh, company = d["warehouse_id"], d["company_code"]
by_line: dict = {}
for dec in decisions:
if dec.decision not in DECISIONS:
raise BadInput(f"a decision is one of {', '.join(DECISIONS)}")
if dec.line_no not in lines:
raise RuleBroken(f"{document_id} has no line {dec.line_no}")
by_line.setdefault(dec.line_no, []).append(dec)
posted, reports, n = [], [], 10
for ln_no, ln in lines.items():
ds = by_line.get(ln_no, [])
pid = ln["product_id"]
bases = [products.convert(unit, pid, x.quantity, x.unit_code, document_ref=document_id, what="an inspected quantity").base for x in ds]
if sum(bases, ZERO) != ln["quantity_base"]:
raise GateBroken("every_unit_decided", f"{qstr(sum(bases, ZERO))} of the {qstr(ln['quantity_base'])} {unit.base_unit(pid)} are decided",
"decide every returned unit: restock, scrap or back to the customer", 409)
where = ln["to_location_id"]
if unit.location(where)["stock_status"] != QUALITY:
raise RuleBroken(f"{where} is not a quality location: this return was not held for inspection")
pieces = _pieces(ln)
unit_cost_in = (ln["value_amount"] / ln["quantity_base"]) if ln["value_amount"] else ZERO
wunit = unit.product(pid)
vclass = wunit.get("valuation_class") or "*"
tot = {"accepted": ZERO, "rejected": ZERO}
for no, (dec, base) in enumerate(zip(ds, bases), start=1):
left = base
while left > 0:
lot, have = pieces[0]
take = min(left, have)
pieces[0] = (lot, have - take)
if pieces[0][1] == 0:
pieces.pop(0)
left -= take
n += 1
if dec.decision == "restock":
to_loc, to_type = f"{wh}/A/01", UNRESTRICTED
# the restock bin: the warehouse's first unrestricted bin when WH/A/01 does not exist
if unit.repo.get("inv_location", location_id=to_loc) is None:
to_loc = next(x["location_id"] for x in unit.repo.locations_in(wh)
if x["kind"] == "internal" and x["stock_status"] == UNRESTRICTED)
spec = MoveSpec(document_id, ln_no, pid, take, where, to_loc, unit.today, ln["unit_code"], None, 1, 1, lot, QUALITY, to_type,
OURS, OURS, n=n, reason=dec.reason or "restocked after inspection")
tot["accepted"] += take
elif dec.decision == "scrap":
spec = MoveSpec(document_id, ln_no, pid, take, where, L.virtual_id(company, "LOSS"), unit.today, ln["unit_code"], None, 1, 1,
lot, QUALITY, QUALITY, OURS, OURS, n=n, value_out=money(take * unit_cost_in),
reason=dec.reason or "scrapped after inspection")
tot["accepted"] += take
else:
spec = MoveSpec(document_id, ln_no, pid, take, where, L.virtual_id(company, "CUSTOMER"), unit.today, ln["unit_code"], None,
1, 1, lot, QUALITY, QUALITY, OURS, OURS, n=n, value_out=money(take * unit_cost_in),
reason=dec.reason or "sent back to the customer")
tot["rejected"] += take
pm = ledger.post_move(unit, spec)
posted.append((dec.decision, vclass, pm))
unit.put("inv_inspection", {"document_id": document_id, "line_no": ln_no, "decision_no": no, "decision": dec.decision,
"quantity_base": base, "reason": dec.reason, "inspected_by_email": (caller.get("email") or "").lower(),
"inspected_by_sub": (caller.get("sub") or "").lower(), "reported_to_order": True},
("document_id", "line_no", "decision_no"))
reports.append((ln_no, ln, tot, ds))
number = d["number"]
jl = []
for kind, vclass, pm in posted:
v = pm.row["value_amount"]
if kind == "scrap":
jl += postings.movement_lines(v, "inventory_scrap", vclass, memo=f"return {number} scrapped")
elif kind == "back_to_customer":
jl += postings.movement_lines(v, "cost_of_goods_sold", vclass, memo=f"return {number} sent back")
postings.stage(unit, document_id, "return_inspection", company, unit.today, f"return inspection {number}", _merge(jl))
unit.flush_races()
told = []
if seams.sales_installed():
rma = {r["line_no"]: r for r in _rma_rows(unit, d["rma_ref"])}
for ln_no, ln, tot, ds in reports:
row = rma[ln["parent_line_no"]]
conv1 = products.convert(unit, ln["product_id"], Decimal(1), row["unit_code"])
def inu(b):
return qstr(_in_unit(b, conv1.numerator, conv1.denominator))
decision = "restock" if any(x.decision == "restock" for x in ds) else "scrap"
body_ = {"line_no": row["line_no"], "inspection_ref": f"inv:{number}:{ln_no}", "received_quantity": inu(ln["quantity_base"]),
"accepted_quantity": inu(tot["accepted"]), "rejected_quantity": inu(tot["rejected"]), "decision": decision,
"reason": "; ".join(x.reason for x in ds if x.reason) or decision, "inspected_on": unit.today.isoformat()}
try:
told.append(seams.return_inspection(row["sales_order_id"], body_))
except seams.Missing as m:
raise GateBroken("order_accepts", f"the returns order refused the inspection: {m.text}", "try again when sales-order answers", 409) from None
return {"document_id": document_id, "replay": False,
"lines": [{"line_no": ln_no, "accepted": qstr(t["accepted"]), "rejected": qstr(t["rejected"])} for ln_no, _l, t, _d in reports],
"moves": [{"move_id": pm.row["move_id"], "decision": k, "to_location_id": pm.row["to_location_id"],
"quantity_base": qstr(pm.row["quantity_base"]), "value_amount": f"{pm.row['value_amount']:.2f}"} for k, _v, pm in posted]}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /returns/{document_id}")
def get_return(caller: dict, document_id: Ref) -> dict:
"""A customer return and what the inspector decided for each of its units."""
need_read(caller)
with U.read() as unit:
d = unit.repo.get("inv_document", document_id=document_id)
if d is None or d["kind"] != "return":
raise Refused(404, f"no return {document_id}")
ins = unit.repo.find("inv_inspection", order=("line_no", "decision_no"), document_id=document_id)
return {"document_id": document_id, "number": d["number"], "rma_ref": d["rma_ref"], "bp_id": d["bp_id"], "lifecycle": d["lifecycle"],
"inspections": [{"line_no": r["line_no"], "decision": r["decision"], "quantity_base": qstr(r["quantity_base"]),
"reported_to_order": bool(r["reported_to_order"])} for r in ins]}