dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/consignment.py
"""Special stocks: supplier consignment (INV-46), customer consignment (INV-45), subcontracting (INV-47).
Supplier consignment: the stock sits in our bin, owned by the supplier, valued at nothing. Taking it into our ownership (a
delivery picking it, or the warehouse manager withdrawing it) is a move between owners at the same place, valued ONCE at the
price purchasing answers (POST /consignment-withdrawals, idempotent on move_ref), with its own journal."""
import uuid
from decimal import Decimal
from typing import Optional
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 qstr
from brick_inventory.application.inputs import Code, ProvidedIn, Ref, day
from brick_inventory.application.unit import Unit
from brick_inventory.application.warehouses import special_location
from brick_inventory.domain.errors import BadInput, GateBroken, RuleBroken, ShortOf
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
def _consigned(unit: Unit, product_id: str, location_id: str, lot: str, stock_type: str) -> list:
"""The (owner, quantity) rows of stock in this bin that a supplier still owns."""
out = []
for r in ledger.stock_rows(unit, product_id):
if r.location_id == location_id and r.lot_code == lot and r.stock_type == stock_type and r.owner != OURS and r.quantity_base > 0:
out.append((r.owner, r.quantity_base))
return sorted(out)
def take_into_ownership(unit: Unit, document_id: str, line_no: int, product_id: str, location_id: str, lot: str, stock_type: str,
quantity_base: Decimal, unit_code: str, base_code: str, company: str, memo: str) -> list:
"""Move `quantity_base` of a bin's consigned stock into our ownership: asks purchasing for the price (a refusal refuses the
caller and nothing moves), then appends the owner-to-us move valued at its answer. Returns the posted moves."""
posted = []
left = quantity_base
for i, (owner, have) in enumerate(_consigned(unit, product_id, location_id, lot, stock_type), start=1):
if left <= 0:
break
take = min(left, have)
ref = f"inv:{document_id}:{line_no}" + (f":{owner}" if i > 1 else "")
try:
answer = seams.consignment_withdrawal({"move_ref": ref, "bp_id": int(owner), "product_id": product_id, "quantity": qstr(take),
"unit_code": base_code, "withdrawn_on": unit.today.isoformat()})
except seams.Missing as m:
reason = m.reason if isinstance(m.reason, dict) else {}
raise GateBroken(reason.get("gate") or "consignment_price", reason.get("reason") or m.text,
reason.get("fix") or "agree a consignment price with the supplier (purchasing)", 409) from None
amount = money(Decimal(str(answer["amount"])))
spec = MoveSpec(document_id, line_no, product_id, take, location_id, location_id, unit.today, base_code, take, 1, 1, lot or NO_LOT,
stock_type, stock_type, owner, OURS, n=100 + i, move_ref=ref, reason=memo, value_in=amount)
pm = ledger.post_move(unit, spec)
posted.append((owner, take, amount, pm))
left -= take
vclass = unit.product(product_id).get("valuation_class") or "*"
postings.stage(unit, f"{document_id}:cw:{line_no}:{owner}", "consignment_withdrawal", company, unit.today, memo,
[postings.entry("stock_valuation", amount, valuation_class=vclass, memo=memo),
postings.entry("consignment_liability", amount, natural=postings.CREDIT, memo=memo)])
return posted
def withdraw_if_needed(unit: Unit, d: dict, pick: dict, spec: MoveSpec):
"""A pick from a bin whose stock belongs to a consigning supplier takes it into our ownership first (INV-46)."""
lot = pick["lot_code"] or ""
stype = pick["stock_type"] or UNRESTRICTED
own = unit.stock(ledger.key_of(pick["product_id"], pick["from_location_id"], lot, stype, OURS))
short = pick["quantity_base"] - max(own, ZERO)
if short <= 0:
return None
return take_into_ownership(unit, d["document_id"], pick["line_no"], pick["product_id"], pick["from_location_id"], lot, stype, short,
pick["unit_code"], unit.base_unit(pick["product_id"]), unit.warehouse(d["warehouse_id"])["company_code"],
f"consignment withdrawal for {d['document_id']}")
@INV.use_case(kind="command", route="POST /consignment-stock/withdrawals", permission="inv:transfer:write", idempotent=True)
def withdraw_consignment(caller: dict, product_id: Code, location_id: Code, quantity: str, unit_code: Code, lot_code: Optional[Code] = None,
request_id: Optional[Code] = None) -> dict:
"""Take stock a supplier still owns into our ownership where it stands, with no delivery: valued at purchasing's consignment
price (purchasing is told once and settles with the supplier), journal stock_valuation / consignment_liability."""
@U.decide
def body(unit):
loc = unit.location(location_id)
if loc["kind"] != "internal":
raise RuleBroken(f"{location_id} is not a bin")
conv = products.convert(unit, product_id, quantity, unit_code, document_ref="withdrawal", what="the quantity")
doc_id = f"cw:{loc['warehouse_id']}:{request_id or uuid.uuid4().hex[:12]}"
if unit.repo.get("inv_document", document_id=doc_id):
return {"document_id": doc_id, "replay": True}
have = sum((q for _o, q in _consigned(unit, product_id, location_id, lot_code or "", UNRESTRICTED)), ZERO)
if have < conv.base:
raise RuleBroken(f"{location_id} holds {qstr(have)} {unit.base_unit(product_id)} of {product_id} owned by a supplier, {qstr(conv.base)} asked")
docs.stage_document(unit, doc_id, "withdrawal", loc["warehouse_id"], loc["company_code"], lifecycle="done", done_on=unit.today,
done_by=U.who(caller), source_kind="consignment", source_ref=location_id,
number=docs.claim_number(unit, loc["warehouse_id"], "transfer", doc_id))
got = take_into_ownership(unit, doc_id, 1, product_id, location_id, lot_code or "", UNRESTRICTED, conv.base, unit_code,
unit.base_unit(product_id), loc["company_code"], f"consignment withdrawal {doc_id}")
unit.flush_races()
return {"document_id": doc_id, "replay": False, "withdrawn": [{"owner_bp_id": o, "quantity_base": qstr(q), "amount": f"{a:.2f}",
"move_id": pm.row["move_id"]} for o, q, a, pm in got]}
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 /subcontract-provisions", permission="inv:receipt:intake", idempotent=True, tool=False)
def provide_to_subcontractor(caller: dict, move_ref: Ref, purchase_order_id: Ref, line_no: int, bp_id: str, from_location_id: Code,
components: list[ProvidedIn], provided_on: str) -> dict:
"""Purchasing's subcontract order names the components to send: they leave our bin for stock held at the subcontractor - still
ours, valued as before, never available to anyone. Keyed by move_ref: a repeat moves nothing. A short component is 409
{gate: on_hand}. The subcontract receipt later consumes them (POST /receipts with consume_provided)."""
@U.decide
def body(unit):
doc_id = f"prov:{move_ref}"
existing = unit.repo.moves_of(document_id=doc_id)
if existing:
return {"move_id": existing[0]["move_id"], "replay": True,
"components": [{"product_id": m["product_id"], "quantity_base": qstr(m["quantity_base"])} for m in existing]}
on = day(provided_on)
if on > unit.today:
raise BadInput(f"provided_on {on.isoformat()} is in the future")
loc = unit.location(from_location_id)
if loc["kind"] != "internal":
raise RuleBroken(f"{from_location_id} is not a bin")
sub = special_location(unit, loc["company_code"], "subcontractor", int(bp_id))
docs.stage_document(unit, doc_id, "transfer", loc["warehouse_id"], loc["company_code"], lifecycle="done", done_on=unit.today,
done_by=U.who(caller), source_kind="subcontract_provision", source_ref=purchase_order_id, bp_id=int(bp_id),
number=docs.claim_number(unit, loc["warehouse_id"], "transfer", doc_id))
out = []
for i, c in enumerate(components, start=1):
conv = products.convert(unit, c.product_id, c.quantity, c.unit_code, document_ref=doc_id, what="a component quantity")
spec = MoveSpec(doc_id, line_no, c.product_id, conv.base, from_location_id, sub, on, c.unit_code, conv.quantity, conv.numerator,
conv.denominator, n=i, move_ref=move_ref)
try:
pm = ledger.post_move(unit, spec)
except ShortOf as s:
raise GateBroken("on_hand", str(s), "receive or move the missing component into that bin first", 409) from None
out.append(pm)
unit.flush_races()
return {"move_id": out[0].row["move_id"], "replay": False,
"components": [{"product_id": p.row["product_id"], "quantity_base": qstr(p.row["quantity_base"])} for p in out]}
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 /consignment-stock/pickups", permission="inv:transfer:write", idempotent=True)
def pick_up_consignment_stock(caller: dict, bp_id: int, product_id: Code, quantity: str, unit_code: Code, to_location_id: Code,
request_id: Optional[Code] = None) -> dict:
"""Bring stock we keep at a customer's site back to one of our bins (the CONS_PICKUP order): a move from UK1:CONSIGNMENT:<bp>
to the bin with no change of value, no cost of sales and no delivery on the order. More than the customer holds is a 409."""
@U.decide
def body(unit):
dst = unit.location(to_location_id)
if dst["kind"] != "internal":
raise RuleBroken(f"{to_location_id} is not a bin")
src = special_location(unit, dst["company_code"], "consignment_customer", int(bp_id))
doc_id = f"cp:{dst['warehouse_id']}:{request_id or uuid.uuid4().hex[:12]}"
old = unit.repo.moves_of(document_id=doc_id)
if old:
return {"document_id": doc_id, "move_id": old[0]["move_id"], "replay": True}
conv = products.convert(unit, product_id, quantity, unit_code, document_ref=doc_id, what="the picked-up quantity")
docs.stage_document(unit, doc_id, "transfer", dst["warehouse_id"], dst["company_code"], lifecycle="done", done_on=unit.today,
done_by=U.who(caller), source_kind="consignment_pickup", source_ref=src, bp_id=int(bp_id),
number=docs.claim_number(unit, dst["warehouse_id"], "transfer", doc_id))
try:
pm = ledger.post_move(unit, MoveSpec(doc_id, 1, product_id, conv.base, src, to_location_id, unit.today, unit_code, conv.quantity,
conv.numerator, conv.denominator, reason="consignment pick-up"))
except ShortOf as s:
raise GateBroken("consignment_stock", f"customer {bp_id} holds less than that of {product_id}: {s}",
"ask sales-order for the customer's consignment stock first", 409) from None
unit.flush_races()
return {"document_id": doc_id, "move_id": pm.row["move_id"], "from_location_id": src, "to_location_id": to_location_id,
"quantity_base": qstr(conv.base), "replay": False}
return U.run(caller, body, lost_message=lambda repo, row=None: "another mover took the stock at the same moment")