dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/stock.py
"""What is where, how many, what is promised, and what it is worth - all derived from the moves at the moment of asking."""
from datetime import date, datetime
from decimal import Decimal
from typing import Optional
from zoneinfo import ZoneInfo
from erp_platform import auth
from erp_platform.bricks import Refused
from brick_inventory.application import ledger, products
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import AVAILABILITY, VALUE, num, qstr, shaped
from brick_inventory.application.inputs import Code, IsoDate, IsoTime, day
from brick_inventory.domain import measures
from brick_inventory.domain import stock as stock_rules
from brick_inventory.domain import units as dunits
from brick_inventory.domain.values import ZERO
from brick_inventory.infrastructure import business_time
def as_of_moment(unit, as_of: Optional[str]) -> Optional[datetime]:
"""'yesterday 17:00' as a moment: a date or a time without a zone is read in the business timezone."""
if not as_of:
return None
try:
d = datetime.fromisoformat(as_of.replace("Z", "+00:00")) if "T" in as_of or " " in as_of else \
datetime.combine(date.fromisoformat(as_of), datetime.max.time().replace(microsecond=0))
except ValueError:
raise Refused(422, f"{as_of!r} is not a date or a time") from None
if d.tzinfo is None:
d = d.replace(tzinfo=ZoneInfo(business_time.timezone_name()))
return d
def shown_units(info: dict) -> list:
"""The units a quantity is shown in: the warehouse unit first, then the sales unit, then the base unit for the rest."""
base = info.get("base_unit_code")
out = []
for role in ("warehouse", "sales"):
code = (info.get("roles") or {}).get(role)
row = products.unit_row(info, code) if code else None
if row and code != base and code not in [u["unit_code"] for u in out]:
out.append(row)
return out
def display(info: dict, qty: Decimal) -> list:
base = info.get("base_unit_code", "")
return [{"quantity": qstr(q), "unit_code": u} for q, u in dunits.breakdown(qty, base, shown_units(info))]
def in_transit_in(unit, product_id: str, warehouse_id: str) -> Decimal:
"""Stock on its way TO a warehouse: lines of shipped, not yet received transfers."""
total = ZERO
for d in unit.repo.find("inv_document", kind="transfer", lifecycle="in_transit"):
for ln in unit.repo.find("inv_document_line", document_id=d["document_id"]):
if ln["product_id"] == product_id and ln["line_kind"] == "transit" and (ln["to_location_id"] or "").split("/")[0] == warehouse_id:
total += (ln["quantity_base"] or ZERO) - (ln["quantity_done"] or ZERO)
return total
def warehouse_picture(unit, info: dict, product_id: str, wid: str, rows: list, *, detail: bool) -> dict:
internal = [r for r in rows if r.kind == "internal"]
s = stock_rules.summarise(internal, unit.today)
reserved = unit.repo.reserved(wid, product_id)
available = max(s["available_before_reservations"] - reserved, ZERO)
out = {"warehouse_id": wid, "on_hand": qstr(s["on_hand"]), "reserved": qstr(reserved), "available": qstr(available),
"in_transit_in": qstr(in_transit_in(unit, product_id, wid))}
if detail:
out.update(by_type={k: qstr(v) for k, v in s["by_type"].items()}, consignment=qstr(s["consignment"]),
short=qstr(max(reserved - s["available_before_reservations"], ZERO)),
shown=display(info, s["on_hand"]),
locations=[{"location_id": r.location_id, "lot_code": r.lot_code or None, "stock_type": r.stock_type,
"owner": r.owner or None, "quantity_base": qstr(r.quantity_base), "shown": display(info, r.quantity_base),
"expires_on": r.expires_on.isoformat() if r.expires_on else None,
"negative": r.quantity_base < 0} for r in sorted(internal, key=lambda r: (r.location_id, r.lot_code))])
return out
@INV.use_case(kind="query", route="GET /stock/{product_id}")
def get_stock(caller: dict, product_id: Code, warehouse_id: Optional[Code] = None, as_of: Optional[IsoTime] = None,
location_prefix: Optional[Code] = None) -> dict:
"""How many of a product are where: on hand per warehouse and bin (by lot, stock type and owner), reserved, available, with
the quantity also shown in the packs people handle (the warehouse unit first, then the sales unit). Derived from the moves
at the moment it is asked, or as of a past moment; asking twice gives the same answer. Needs inv:stock:read (a sales rep
asks get_availability, which has no bins)."""
from brick_inventory.application.common import need_read
need_read(caller)
with U.read() as unit:
info = unit.product(product_id)
moment = as_of_moment(unit, as_of)
rows = ledger.stock_rows(unit, product_id, warehouse_id=warehouse_id, as_of=moment)
if location_prefix:
rows = [r for r in rows if r.location_id == location_prefix or r.location_id.startswith(location_prefix + "/")]
by_wh: dict = {}
for r in rows:
by_wh.setdefault(r.warehouse_id or "", []).append(r)
warehouses = [warehouse_picture(unit, info, product_id, wid, rs, detail=True) for wid, rs in sorted(by_wh.items()) if wid]
if moment is not None:
for w in warehouses:
w["reserved"] = w["available"] = None
total = sum((r.quantity_base for r in rows if r.kind == "internal"), ZERO)
transit = sum((r.quantity_base for r in rows if r.kind == "transit"), ZERO)
wu = products.warehouse_unit(info)
sales = (info.get("roles") or {}).get("sales")
out = {"product_id": product_id, "base_unit_code": info.get("base_unit_code"), "warehouse_unit": wu,
"warehouse_unit_set": bool((info.get("roles") or {}).get("warehouse")),
"as_of": moment.isoformat() if moment else None, "warehouses": warehouses, "on_hand": qstr(total),
"in_transit": qstr(transit), "shown": display(info, total)}
if sales and sales != info.get("base_unit_code") and (row := products.unit_row(info, sales)):
out["in_sales_unit"] = dunits.express(total, row)
return out
@INV.use_case(kind="query", route="GET /availability/{product_id}")
def get_availability(caller: dict, product_id: Code, warehouse_id: Optional[Code] = None) -> dict:
"""What can be promised: on hand, reserved and available per warehouse, and what is on its way in. No bins, no lots, no value:
a sales rep asks this, never the stock itself. Stock in quality, blocked, expired or consignment-at-a-customer is never
available."""
if not (auth.has_permission(caller, AVAILABILITY) or auth.has_permission(caller, "inv:stock:read")):
raise Refused(404, "not found")
with U.read() as unit:
info = unit.product(product_id)
rows = ledger.stock_rows(unit, product_id, warehouse_id=warehouse_id)
by_wh: dict = {}
for r in rows:
if r.warehouse_id:
by_wh.setdefault(r.warehouse_id, []).append(r)
if warehouse_id and warehouse_id not in by_wh:
by_wh[warehouse_id] = []
warehouses = [warehouse_picture(unit, info, product_id, wid, rs, detail=False) for wid, rs in sorted(by_wh.items())]
return {"product_id": product_id, "base_unit_code": info.get("base_unit_code"), "warehouses": warehouses}
@INV.use_case(kind="query", route="GET /valuation/products/{product_id}", permission=VALUE)
def get_valuation(caller: dict, product_id: Code, company_code: Code, as_of: Optional[IsoDate] = None) -> dict:
"""The stock value of a product in a company on a date: quantity, value and the average (or standard) cost, with the moves
behind it and the state of the journals asked of the ledger. The value is the sum of the moves' values exactly; at zero
quantity it is 0.00. Needs inv:valuation:read."""
with U.read() as unit:
on = day(as_of) or unit.today
rows = [r for r in unit.repo.valuation_rows(company_code, product_id) if r["moved_on"] <= on]
last = rows[-1] if rows else None
moves = [m for m in unit.repo.find("inv_move", order=("moved_at", "move_id"), product_id=product_id, company_code=company_code)
if m["moved_on"] <= on and m["value_amount"]]
by_doc = {}
for m in moves:
by_doc.setdefault(m["document_id"], []).append(m)
journals = []
for doc in by_doc:
for p in unit.repo.find("inv_posting", document_id=doc):
journals.append({"document_id": doc, "event": p["event"], "status": p["status"], "journal_id": str(p["journal_id"]),
"reason": p["reason"] or None})
value = sum((m["value_amount"] for m in moves), ZERO)
return {"product_id": product_id, "company_code": company_code, "quantity_base": qstr(last["quantity_after"] if last else ZERO),
"value_amount": f"{(last['value_after'] if last else ZERO):.2f}", "average_cost": num(last["average_cost_after"]) if last else "0.0000",
"method": (last or {}).get("method") or unit.settings.method(company_code, on), "as_of": on.isoformat(),
"sum_of_move_values": f"{value:.2f}",
"moves": [{"move_id": m["move_id"], "moved_on": m["moved_on"].isoformat(), "quantity_base": qstr(m["quantity_base"]),
"value_amount": f"{m['value_amount']:.2f}"} for m in moves],
"valuation_rows": [{"seq": r["seq"], "kind": r["kind"], "move_id": r["move_id"] or None,
"quantity_after": qstr(r["quantity_after"]), "value_after": f"{r['value_after']:.2f}",
"average_cost_after": num(r["average_cost_after"]), "price_difference": f"{r['price_difference'] or 0:.2f}"}
for r in rows],
"journals": journals,
"ledger_balance": None, "ledger_difference": None,
"ledger_note": "the stock account's balance is read from the ledger when it offers one; none is read today"}
@INV.use_case(kind="query", route="GET /moves")
def list_moves(caller: dict, product_id: Optional[Code] = None, document_id: Optional[Code] = None, lot_code: Optional[Code] = None,
limit: int = 200) -> dict:
"""The ledger itself: moves of a product, lot or document, oldest first. Value only to inv:valuation:read."""
from brick_inventory.application.common import need_read
need_read(caller)
with U.read() as unit:
if not (product_id or document_id):
raise Refused(422, "name a product or a document")
rows = unit.repo.moves_of(document_id=document_id, product_id=product_id, lot_code=lot_code)[: max(1, min(limit, 1000))]
return shaped(caller, {"moves": [move_view(r) for r in rows]})
def move_view(r: dict) -> dict:
return {"move_id": r["move_id"], "document_id": r["document_id"], "line_no": r["line_no"], "product_id": r["product_id"],
"lot_code": r["lot_code"] or None, "stock_type_from": r["stock_type_from"], "stock_type_to": r["stock_type_to"],
"owner_from": r["owner_from"] or None, "owner_to": r["owner_to"] or None, "quantity_base": qstr(r["quantity_base"]),
"unit_code": r["unit_code"], "quantity": qstr(r["quantity"]), "unit_numerator": r["unit_numerator"],
"unit_denominator": r["unit_denominator"], "from_location_id": r["from_location_id"], "to_location_id": r["to_location_id"],
"moved_on": r["moved_on"].isoformat(), "value_amount": f"{r['value_amount']:.2f}" if r["value_amount"] is not None else None,
"reverses_move_id": r["reverses_move_id"] or None, "move_ref": r["move_ref"] or None, "reason": r["reason"] or None}
def line_view(l: dict) -> dict:
return {"line_no": l["line_no"], "line_kind": l["line_kind"], "product_id": l["product_id"], "unit_code": l["unit_code"],
"unit_numerator": l["unit_numerator"], "unit_denominator": l["unit_denominator"],
"quantity_expected": qstr(l["quantity_expected"]), "quantity_done": qstr(l["quantity_done"]),
"quantity_base": qstr(l["quantity_base"]), "from_location_id": l["from_location_id"], "to_location_id": l["to_location_id"],
"lot_code": l["lot_code"] or None, "source_line_ref": l["source_line_ref"], "unit_cost": num(l["unit_cost"]),
"gross_weight": qstr(l["gross_weight"]), "net_weight": qstr(l["net_weight"]), "weight_unit_code": l["weight_unit_code"],
"volume": qstr(l["volume"]), "volume_unit_code": l["volume_unit_code"], "value_amount": num(l["value_amount"])}
@INV.use_case(kind="query", route="GET /documents/{document_id}")
def get_document(caller: dict, document_id: Code) -> dict:
"""A receipt, delivery, return, transfer, adjustment or issue: its number, its state, its lines and the moves it made."""
from brick_inventory.application.common import need_read
need_read(caller)
with U.read() as unit:
d = unit.repo.get("inv_document", document_id=document_id)
if d is None:
raise Refused(404, f"no document {document_id}")
lines = unit.repo.find("inv_document_line", order=("line_no",), document_id=document_id)
moves = unit.repo.moves_of(document_id=document_id)
view = {k: (v.isoformat() if hasattr(v, "isoformat") else v) for k, v in d.items()
if k not in ("extra", "created_at", "updated_at") and v is not None}
view.update(lines=[line_view(l) for l in lines], moves=[move_view(m) for m in moves])
packages = unit.repo.find("inv_package", document_id=document_id)
if packages:
view["packages"] = [{"sscc": p["sscc"], "package_type_code": p["package_type_code"]} for p in packages]
return shaped(caller, view)
@INV.use_case(kind="query", route="GET /documents/{document_id}/shipping")
def get_document_shipping(caller: dict, document_id: Code) -> dict:
"""A delivery's (or a receipt's) gross and net weight and volume for the carrier: each line's per-unit measures were copied
when it was written, so a later change on the product never changes a document. A line whose measure is unknown is named
("volume unknown for 1 of 2 lines"), never counted as zero. Packages, when the delivery was packed, are listed."""
from brick_inventory.application.common import need_read
need_read(caller)
with U.read() as unit:
d = unit.repo.get("inv_document", document_id=document_id)
if d is None:
raise Refused(404, f"no document {document_id}")
lines = [l for l in unit.repo.find("inv_document_line", order=("line_no",), document_id=document_id)
if l["line_kind"] in ("receipt", "pick", "return") and l["quantity_done"]]
try:
ship = measures.shipping([{**l, "label": f"{l['product_id']} ({l['unit_code']})"} for l in lines])
except Exception as e: # noqa: BLE001
raise Refused(409, str(e)) from None
packs = [{"sscc": p["sscc"], "package_type_code": p["package_type_code"],
"gross_weight": qstr(p["gross_weight"]), "weight_unit_code": p["weight_unit_code"],
"contents": __import__("json").loads(p["contents"] or "[]")}
for p in unit.repo.find("inv_package", document_id=document_id)]
return {"document_id": document_id, "number": d["number"], **ship, "packages": packs}