dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/issues.py
"""Goods issue to a cost centre and scrapping (INV-37, INV-38): stock leaves for our own use or as a loss, in the product's issue
unit, valued at the cost in force, with a second person above a threshold; reversed at the value it was posted at, once."""
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, shaped
from brick_inventory.application.counts import unit_cost
from brick_inventory.application.inputs import Code, IssueLineIn, Reason, Ref
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
PURPOSES = ("cost_centre", "scrap", "internal_use")
SCRAP_REASONS = ("DAMAGED", "EXPIRED", "QUALITY")
def _issue(unit, document_id: str) -> tuple:
d = unit.repo.get("inv_document", document_id=document_id)
if d is None or d["kind"] != "issue":
raise Refused(404, f"no goods issue {document_id}")
lines = [l for l in unit.repo.find("inv_document_line", order=("line_no",), document_id=document_id) if l["line_kind"] == "issue"]
return d, lines
def _value_of(unit, d: dict, lines: list) -> Decimal:
total = ZERO
for l in lines:
total += money(l["quantity_base"] * unit_cost(unit, d["company_code"], l["product_id"]))
return total
def _threshold(unit) -> Decimal:
return unit.settings.number("issue_approval_threshold", unit.today)
@INV.use_case(kind="command", route="POST /goods-issues", permission="inv:issue:write", idempotent=True)
def plan_goods_issue(caller: dict, purpose: Code, lines: list[IssueLineIn], cost_centre: Optional[Code] = None, scrap_reason: Optional[Code] = None,
request_id: Optional[Ref] = None) -> dict:
"""Plan taking stock out for our own use. Purpose `cost_centre` charges the cost to a cost centre (checked against the ledger's
open cost centres, gate cost_centre_valid); `scrap` writes damaged or expired stock off with a reason (DAMAGED, EXPIRED,
QUALITY); `internal_use` is consumption without a cost centre. Each line is in any unit of the product - the product's
issue unit is proposed when none is given - from a bin (and lot). Nothing moves until it is posted."""
@U.decide
def body(unit):
if purpose not in PURPOSES:
raise BadInput(f"purpose is one of {', '.join(PURPOSES)}")
if purpose == "cost_centre":
if not cost_centre:
raise GateBroken("cost_centre_valid", "an issue to a cost centre names it", "pick a cost centre (gl)", 422)
if seams.ledger_installed():
try:
cc = seams.cost_centre(cost_centre)
except seams.Missing as m:
raise GateBroken("cost_centre_valid", "the ledger has no open cost centre with that code" if m.status == 404 else f"the ledger could not check the cost centre: {m.text}",
"pick a cost centre (gl)", 409) from None
if cc.get("status") != "open":
raise GateBroken("cost_centre_valid", f"cost centre {cost_centre} is {cc.get('status')}", "pick an open cost centre (gl)", 409)
if purpose == "scrap" and (not scrap_reason or scrap_reason not in SCRAP_REASONS):
raise GateBroken("scrap_reason", f"a scrap names its reason: {', '.join(SCRAP_REASONS)}", "give a reason", 422)
first = unit.location(lines[0].from_location_id)
rid = request_id or uuid.uuid4().hex[:12]
doc_id = f"gi:{first['warehouse_id']}:{rid}"
if unit.repo.get("inv_document", document_id=doc_id):
return {"document_id": doc_id, "replay": True}
docs.stage_document(unit, doc_id, "issue", first["warehouse_id"], first["company_code"], lifecycle="draft", purpose=purpose,
cost_centre=cost_centre, reason=scrap_reason, source_kind="issue", source_ref=purpose)
out = []
for i, ln in enumerate(lines, start=1):
loc = unit.location(ln.from_location_id)
if loc["kind"] != "internal" or loc["warehouse_id"] != first["warehouse_id"]:
raise RuleBroken(f"{ln.from_location_id} is not a bin of warehouse {first['warehouse_id']}")
info = products.stocked(unit, ln.product_id)
code = ln.unit_code or products.role_unit(info, "issue")
conv = products.convert(unit, ln.product_id, ln.quantity, code, document_ref=doc_id, what="the issued quantity")
tr = unit.tracking(ln.product_id)["tracking"]
if tr != "none" and not ln.lot_code:
raise RuleBroken(f"{ln.product_id} is tracked by {tr}: name the {'serial number' if tr == 'serial' else 'lot'}")
docs.stage_line(unit, doc_id, i, line_kind="issue", product_id=ln.product_id, unit_code=code, unit_numerator=conv.numerator,
unit_denominator=conv.denominator, quantity_expected=conv.quantity, quantity_base=conv.base,
from_location_id=ln.from_location_id, lot_code=ln.lot_code or NO_LOT, stock_type=ln.stock_type or loc["stock_status"] or UNRESTRICTED)
out.append({"line_no": i, "product_id": ln.product_id, "quantity": qstr(conv.quantity), "unit_code": code, "quantity_base": qstr(conv.base)})
return {"document_id": doc_id, "lifecycle": "draft", "purpose": purpose, "cost_centre": cost_centre, "lines": out, "replay": False}
return U.run(caller, body)
def _post(unit, d: dict, lines: list, caller: dict) -> dict:
wh = unit.warehouse(d["warehouse_id"])
company = wh["company_code"]
sink = L.virtual_id(company, "LOSS" if d["purpose"] == "scrap" else "CONSUMPTION")
doc_id = d["document_id"]
number = docs.claim_number(unit, d["warehouse_id"], "issue", doc_id)
posted, by_class = [], {}
for l in lines:
spec = MoveSpec(doc_id, l["line_no"], l["product_id"], l["quantity_base"], l["from_location_id"], sink, unit.today, l["unit_code"],
l["quantity_expected"], l["unit_numerator"], l["unit_denominator"], l["lot_code"] or NO_LOT, l["stock_type"] or UNRESTRICTED,
l["stock_type"] or UNRESTRICTED, OURS, OURS, reason=d["reason"] or "")
pm = ledger.post_move(unit, spec)
posted.append(pm)
vc = unit.product(l["product_id"]).get("valuation_class") or "*"
by_class[vc] = by_class.get(vc, ZERO) + pm.row["value_amount"]
docs.stage_line(unit, doc_id, l["line_no"], quantity_done=l["quantity_expected"], value_amount=pm.row["value_amount"])
purpose_account = "inventory_scrap" if d["purpose"] == "scrap" else "consumption"
journal = []
for vc, v in by_class.items():
journal += postings.movement_lines(v, purpose_account, vc, cost_centre=d["cost_centre"] or "", memo=d["reason"] or f"issue {number}")
from brick_inventory.application.deliveries import _merge
postings.stage(unit, doc_id, "issue", company, unit.today, f"goods issue {number}", _merge(journal))
docs.stage_document(unit, doc_id, "issue", d["warehouse_id"], company, lifecycle="posted", number=number, done_on=unit.today, done_by=U.who(caller),
purpose=d["purpose"], cost_centre=d["cost_centre"], reason=d["reason"], source_kind="issue", source_ref=d["source_ref"])
total = sum((p.row["value_amount"] for p in posted), ZERO)
return {"document_id": doc_id, "number": number, "lifecycle": "posted", "purpose": d["purpose"], "cost_centre": d["cost_centre"],
"value_amount": f"{-total:.2f}", "moves": [{"move_id": p.row["move_id"], "to_location_id": p.row["to_location_id"],
"quantity_base": qstr(p.row["quantity_base"]), "unit_code": p.row["unit_code"],
"quantity": qstr(p.row["quantity"])} for p in posted]}
@INV.use_case(kind="command", route="POST /goods-issues/{document_id}/post", permission="inv:issue:write", idempotent=True)
def post_goods_issue(caller: dict, document_id: Ref) -> dict:
"""Post a planned issue: the stock moves from its bin to the company's consumption (or loss, for scrap) at the average cost,
numbered {warehouse}/GI/00001, and a journal is asked of the ledger (consumption with the cost centre on the line, or
inventory_scrap with the reason). An issue worth the approval threshold or more is not posted by its planner: submit it for a
stock controller. Posting twice posts once."""
@U.decide
def body(unit):
d, lines = _issue(unit, document_id)
if d["lifecycle"] == "posted":
return {"document_id": document_id, "number": d["number"], "lifecycle": "posted", "replay": True}
if d["lifecycle"] != "draft":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only a draft issue is posted")
value = _value_of(unit, d, lines)
if value >= _threshold(unit):
raise GateBroken("under_issue_threshold", f"the issue is worth {value:.2f}, at or above the {_threshold(unit):.2f} approval threshold",
"submit it for a stock controller", 409)
return {**_post(unit, d, lines, caller), "replay": False}
return U.run(caller, body, lost_message=lambda repo, row=None: "another mover took the stock at the same moment")
@INV.use_case(kind="command", route="POST /goods-issues/{document_id}/submit", permission="inv:issue:write", idempotent=True)
def submit_goods_issue(caller: dict, document_id: Ref) -> dict:
"""Submit an issue worth the threshold or more: it waits for a stock controller who is neither its planner nor its submitter.
Nothing moves."""
@U.decide
def body(unit):
d, lines = _issue(unit, document_id)
if d["lifecycle"] == "awaiting_approval":
return {"document_id": document_id, "lifecycle": "awaiting_approval", "replay": True}
if d["lifecycle"] != "draft":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only a draft issue is submitted")
value = _value_of(unit, d, lines)
if value < _threshold(unit):
raise RuleBroken(f"the issue is worth {value:.2f}, under the {_threshold(unit):.2f} approval threshold: post it")
docs.stage_document(unit, document_id, "issue", d["warehouse_id"], d["company_code"], lifecycle="awaiting_approval", source_kind="issue",
source_ref=d["source_ref"], purpose=d["purpose"], approved_by_email=None,
requested_by_email=d["requested_by_email"] or (caller.get("email") or "").lower() or None)
unit.put("inv_document", {"document_id": document_id, "extra": docs.dump({"submitted_by": U.who(caller), "submitted_sub": (caller.get("sub") or "").lower(),
"value": f"{value:.2f}"})}, "document_id")
return {"document_id": document_id, "lifecycle": "awaiting_approval", "value_amount": f"{value:.2f}", "replay": False}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /goods-issues/{document_id}/approve", permission="inv:issue:approve", idempotent=True)
def approve_goods_issue(caller: dict, document_id: Ref) -> dict:
"""A second person approves an issue over the threshold and it is posted: neither its planner nor its submitter, by email AND
sign-in subject (a second account of the same person is refused)."""
@U.decide
def body(unit):
d, lines = _issue(unit, document_id)
if d["lifecycle"] == "posted":
return {"document_id": document_id, "number": d["number"], "lifecycle": "posted", "replay": True}
if d["lifecycle"] != "awaiting_approval":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only an issue waiting for approval is approved")
ex = docs.extra_of(d)
planners = {x for x in (d["requested_by_email"], d["requested_by_sub"], ex.get("submitted_by"), ex.get("submitted_sub")) if x}
if set(U.identities(caller)) & {p.lower() for p in planners}:
raise GateBroken("approver_not_requester", "you planned or submitted this issue (same email or sign-in subject)", "ask another stock controller", 409)
return {**_post(unit, d, lines, caller), "approved_by": U.who(caller), "replay": False}
return U.run(caller, body, lost_message=lambda repo, row=None: "another mover took the stock at the same moment")
@INV.use_case(kind="command", route="POST /goods-issues/{document_id}/reject", permission="inv:issue:approve", idempotent=True)
def reject_goods_issue(caller: dict, document_id: Ref, reason: Reason) -> dict:
"""Reject an issue waiting for approval, with a reason: it is cancelled and nothing moved."""
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a reason is required to reject an issue")
d, _ = _issue(unit, document_id)
if d["lifecycle"] == "cancelled":
return {"document_id": document_id, "lifecycle": "cancelled", "replay": True}
if d["lifecycle"] != "awaiting_approval":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only an issue waiting for approval is rejected")
docs.stage_document(unit, document_id, "issue", d["warehouse_id"], d["company_code"], lifecycle="cancelled", reason=reason, source_kind="issue",
source_ref=d["source_ref"], purpose=d["purpose"])
return {"document_id": document_id, "lifecycle": "cancelled", "reason": reason, "replay": False}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /goods-issues/{document_id}/cancel", permission="inv:issue:write", idempotent=True)
def cancel_goods_issue(caller: dict, document_id: Ref) -> dict:
"""Cancel a draft issue: it never had a number, so no number is wasted."""
@U.decide
def body(unit):
d, _ = _issue(unit, document_id)
if d["lifecycle"] == "cancelled":
return {"document_id": document_id, "lifecycle": "cancelled", "replay": True}
if d["lifecycle"] != "draft":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only a draft issue is cancelled")
docs.stage_document(unit, document_id, "issue", d["warehouse_id"], d["company_code"], lifecycle="cancelled", source_kind="issue",
source_ref=d["source_ref"], purpose=d["purpose"])
return {"document_id": document_id, "lifecycle": "cancelled", "replay": False}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /goods-issues/{document_id}/reverse", permission="inv:move:reverse")
def reverse_goods_issue(caller: dict, document_id: Ref, reason: Reason) -> dict:
"""Reverse a posted issue with a reason (SAP 202 / 552): reversing moves return the stock to its bin, valued at what the issue
was posted at (not today's average), and a reversing journal is asked of the ledger. Reversing twice is refused; so is a
reversal when the ledger's period of the issue is closed."""
@U.decide
def body(unit):
if not reason.strip():
raise BadInput("a reason is required to reverse an issue")
d, lines = _issue(unit, document_id)
if d["lifecycle"] != "posted":
raise RuleBroken(f"{document_id} is {d['lifecycle']}: only a posted issue is reversed")
if d["reversed"]:
raise RuleBroken(f"{document_id} was already reversed")
if seams.ledger_installed() and d["done_on"]:
try:
p = seams.period_open(d["done_on"], d["company_code"])
except seams.Missing:
p = {"open": True}
if not p.get("open", True):
raise GateBroken("period_open", f"the ledger's period of {d['done_on'].isoformat()} is closed", p.get("fix") or "reopen the period", 409)
wh = unit.warehouse(d["warehouse_id"])
company = wh["company_code"]
sink = L.virtual_id(company, "LOSS" if d["purpose"] == "scrap" else "CONSUMPTION")
rid = f"{document_id}:rev"
moves = {m["line_no"]: m for m in unit.repo.moves_of(document_id=document_id)}
by_class, posted = {}, []
for l in lines:
orig = moves[l["line_no"]]
spec = MoveSpec(rid, l["line_no"], l["product_id"], l["quantity_base"], sink, l["from_location_id"], unit.today, l["unit_code"],
l["quantity_expected"], l["unit_numerator"], l["unit_denominator"], l["lot_code"] or NO_LOT, l["stock_type"] or UNRESTRICTED,
l["stock_type"] or UNRESTRICTED, OURS, OURS, reason=reason, reverses_move_id=orig["move_id"],
value_in=abs(orig["value_amount"]))
pm = ledger.post_move(unit, spec)
posted.append(pm)
vc = unit.product(l["product_id"]).get("valuation_class") or "*"
by_class[vc] = by_class.get(vc, ZERO) + pm.row["value_amount"]
purpose_account = "inventory_scrap" if d["purpose"] == "scrap" else "consumption"
journal = []
for vc, v in by_class.items():
journal += postings.movement_lines(v, purpose_account, vc, cost_centre=d["cost_centre"] or "", memo=f"reversal: {reason}")
from brick_inventory.application.deliveries import _merge
postings.stage(unit, rid, "reversal", company, unit.today, f"reversal of goods issue {d['number']}", _merge(journal))
docs.stage_document(unit, rid, "transfer", d["warehouse_id"], company, lifecycle="done", reason=reason, done_on=unit.today, done_by=U.who(caller),
source_kind="reversal", source_ref=document_id)
unit.put("inv_document", {"document_id": document_id, "reversed": True}, "document_id")
return {"document_id": document_id, "reversal_document_id": rid, "reason": reason,
"moves": [{"move_id": p.row["move_id"], "reverses_move_id": p.row["reverses_move_id"], "quantity_base": qstr(p.row["quantity_base"])} for p in posted]}
return U.run(caller, body, lost_message=lambda repo, row=None: "the stock has moved on since: it cannot be reversed")
@INV.use_case(kind="query", route="GET /goods-issues/{document_id}")
def get_goods_issue(caller: dict, document_id: Ref) -> dict:
"""A goods issue: its purpose, cost centre or scrap reason, lines, state and (to those allowed) its value."""
need_read(caller)
with U.read() as unit:
d, lines = _issue(unit, document_id)
return shaped(caller, {"document_id": document_id, "number": d["number"], "lifecycle": d["lifecycle"], "purpose": d["purpose"],
"cost_centre": d["cost_centre"], "scrap_reason": d["reason"] if d["purpose"] == "scrap" else None, "reversed": bool(d["reversed"]),
"lines": [{"line_no": l["line_no"], "product_id": l["product_id"], "quantity": qstr(l["quantity_expected"]),
"unit_code": l["unit_code"], "quantity_base": qstr(l["quantity_base"]), "from_location_id": l["from_location_id"],
"lot_code": l["lot_code"] or None, "value_amount": f"{l['value_amount']:.2f}" if l["value_amount"] is not None else None}
for l in lines]})