dodil/erp-brick-inventory
DODIL ERP brick: inventory
src/brick_inventory/application/warehouses.py
"""Warehouses, zones and bins; which products are stocked and tracked; the settings (INV-01, INV-02, INV-03, INV-17, INV-36)."""
from typing import Optional
from erp_platform.bricks import Refused
from brick_inventory.application import unit as U
from brick_inventory.application.brick import INV
from brick_inventory.application.common import capacity_value as _check_capacity_value
from brick_inventory.application.common import need, need_read, qstr
from brick_inventory.application.inputs import Code, IsoDate, Short, day
from brick_inventory.domain import location as L
from brick_inventory.domain import settings as S
from brick_inventory.domain.errors import BadInput, NotFound, RuleBroken
from brick_inventory.domain.values import COMPANY_VIRTUALS, STOCK_TYPES, ZERO
def _virtuals(unit, company: str) -> None:
kinds = {"SUPPLIER": "supplier", "CUSTOMER": "customer", "LOSS": "loss", "TRANSIT": "transit", "OPENING": "opening",
"CONSUMPTION": "consumption", "PRODUCTION": "production"}
for name in COMPANY_VIRTUALS:
lid = L.virtual_id(company, name)
if unit.repo.get("inv_location", location_id=lid) is None:
unit.put("inv_location", {"location_id": lid, "warehouse_id": "", "company_code": company, "parent_location_id": "",
"kind": kinds[name], "name": name.title(), "stock_status": "unrestricted",
"allowed_stock_types": ",".join(STOCK_TYPES), "status": "active"}, "location_id")
def special_location(unit, company: str, kind: str, bp_id: int) -> str:
"""The customer's consignment location or the subcontractor's, made on first use (stock of ours, held there)."""
lid = L.consignment_id(company, bp_id) if kind == "consignment_customer" else L.subcontract_id(company, bp_id)
if unit.repo.get("inv_location", location_id=lid) is None:
unit.put("inv_location", {"location_id": lid, "warehouse_id": "", "company_code": company, "parent_location_id": "",
"kind": kind, "name": lid, "stock_status": "unrestricted", "bp_id": bp_id,
"allowed_stock_types": ",".join(STOCK_TYPES), "status": "active"}, "location_id")
unit._locations[lid] = {"location_id": lid, "warehouse_id": "", "company_code": company, "kind": kind,
"status": "active", "allowed_stock_types": ",".join(STOCK_TYPES), "stock_status": "unrestricted"}
return lid
def location_view(row: dict) -> dict:
return {"location_id": row["location_id"], "warehouse_id": row["warehouse_id"] or None, "company_code": row["company_code"],
"parent_location_id": row["parent_location_id"] or None, "kind": row["kind"], "name": row["name"],
"stock_status": row["stock_status"], "allowed_stock_types": (row["allowed_stock_types"] or "").split(","),
"pick_sequence": row["pick_sequence"], "max_quantity_base": qstr(row["max_quantity_base"]),
"max_weight": qstr(row["max_weight"]), "max_volume": qstr(row["max_volume"]), "barcode": row["barcode"],
"bp_id": row["bp_id"], "status": row["status"]}
@INV.use_case(kind="command", route="POST /warehouses", permission="inv:warehouse:write", idempotent=True)
def add_warehouse(caller: dict, warehouse_id: Code, company_code: Code, name: Optional[Short] = None,
address: Optional[Short] = None) -> dict:
"""Add a site where stock is kept, belonging to one company. Its stock root (kind view, the same code) and the company's
virtual places (SUPPLIER, CUSTOMER, LOSS, TRANSIT, OPENING, CONSUMPTION: not stock) are made with it. The code is fixed
once set; adding the same warehouse again changes nothing."""
@U.decide
def body(unit):
wid, company = L.check_code(warehouse_id, "a warehouse code"), L.check_company(company_code)
if "/" in wid or ":" in wid:
raise BadInput("a warehouse code has no / or : in it (WH1)")
found = unit.repo.get("inv_warehouse", warehouse_id=wid)
if found:
if found["company_code"] != company:
raise RuleBroken(f"warehouse {wid} belongs to company {found['company_code']}: its company is fixed once set")
return {"warehouse_id": wid, "company_code": company, "created": False}
if unit.repo.get("inv_location", location_id=wid):
raise RuleBroken(f"{wid} already exists")
unit.put("inv_warehouse", {"warehouse_id": wid, "name": name or wid, "company_code": company, "address": address or "",
"status": "active"}, "warehouse_id")
unit.put("inv_location", {"location_id": wid, "warehouse_id": wid, "company_code": company, "parent_location_id": "",
"kind": "view", "name": name or wid, "stock_status": "unrestricted",
"allowed_stock_types": ",".join(STOCK_TYPES), "status": "active"}, "location_id")
_virtuals(unit, company)
return {"warehouse_id": wid, "company_code": company, "created": True}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /locations", permission="inv:warehouse:write")
def add_location(caller: dict, location_id: Code, kind: Code = "internal", name: Optional[Short] = None,
stock_status: Optional[Code] = None, allowed_stock_types: Optional[Short] = None,
pick_sequence: Optional[int] = None, max_quantity_base: Optional[str] = None, max_weight: Optional[str] = None,
max_volume: Optional[str] = None, barcode: Optional[Code] = None) -> dict:
"""Add a zone (kind view) or a bin (kind internal) under a warehouse or a zone: WH1/A is a zone of WH1, WH1/A/01 a bin of
WH1/A. The code is the full path and is fixed once a move names it. A bin's stock_status (unrestricted, quality, blocked)
is the default type of what is put there and its allowed_stock_types the limit (a QC area takes quality only). Capacity
(quantity in the base unit, weight, volume) is optional; empty means unlimited. A second WH1/A/01 is refused."""
@U.decide
def body(unit):
lid = L.check_code(location_id)
if unit.repo.get("inv_location", location_id=lid):
raise RuleBroken(f"{lid} already exists")
parent_id = lid.rsplit("/", 1)[0] if "/" in lid else None
parent = unit.repo.get("inv_location", location_id=parent_id) if parent_id else None
warehouse = unit.repo.get("inv_warehouse", warehouse_id=lid.split("/", 1)[0])
if warehouse is None:
raise RuleBroken(f"there is no warehouse {lid.split('/', 1)[0]}: add it first")
L.new_location(lid, warehouse, parent, kind)
status, allowed = L.check_status(stock_status, allowed_stock_types) if kind == "internal" else ("unrestricted", ",".join(STOCK_TYPES))
cap = {k: _check_capacity_value(v, k) for k, v in (("max_quantity_base", max_quantity_base), ("max_weight", max_weight),
("max_volume", max_volume))}
row = {"location_id": lid, "warehouse_id": warehouse["warehouse_id"], "company_code": warehouse["company_code"],
"parent_location_id": parent["location_id"], "kind": kind, "name": name or lid, "stock_status": status,
"allowed_stock_types": allowed, "pick_sequence": pick_sequence, "barcode": barcode, "status": "active", **cap}
unit.put("inv_location", row, "location_id")
return location_view({**{k: None for k in ("bp_id",)}, **row})
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /locations/settings", permission="inv:warehouse:write", idempotent=True)
def set_location(caller: dict, location_id: Code, name: Optional[Short] = None, stock_status: Optional[Code] = None,
allowed_stock_types: Optional[Short] = None, pick_sequence: Optional[int] = None,
max_quantity_base: Optional[str] = None, max_weight: Optional[str] = None, max_volume: Optional[str] = None,
barcode: Optional[Code] = None) -> dict:
"""Change a bin's name, barcode, pick sequence (the walking order), capacity or stock status. A location's code never
changes. A bin's status (unrestricted, quality, blocked) changes only while it holds no stock: move the stock first."""
@U.decide
def body(unit):
loc = unit.repo.get("inv_location", location_id=location_id)
if loc is None:
raise NotFound(f"no location {location_id}")
if loc["kind"] not in ("internal", "view"):
raise RuleBroken(f"{location_id} is a virtual place: it cannot be renamed, archived or counted")
new = {"location_id": location_id}
if stock_status is not None or allowed_stock_types is not None:
status, allowed = L.check_status(stock_status or loc["stock_status"],
allowed_stock_types if allowed_stock_types is not None else
(None if stock_status else loc["allowed_stock_types"]))
if (status, allowed) != (loc["stock_status"], loc["allowed_stock_types"]):
held = _held(unit, loc)
if held:
raise RuleBroken(f"{location_id} holds {held}: move the stock first, a location's status changes only while it is empty")
new.update(stock_status=status, allowed_stock_types=allowed)
for k, v in (("name", name), ("pick_sequence", pick_sequence), ("barcode", barcode)):
if v is not None:
new[k] = v
for k, v in (("max_quantity_base", max_quantity_base), ("max_weight", max_weight), ("max_volume", max_volume)):
if v is not None:
new[k] = _check_capacity_value(v, k)
unit.put("inv_location", new, "location_id")
return location_view({**loc, **new})
return U.run(caller, body)
def _held(unit, loc: dict) -> str:
"""What a location (and everything under it) holds now, in words ('3 LTR of COLA-075'), or ''."""
ids = {r["location_id"] for r in unit.repo.locations_in(loc["warehouse_id"]) if L.in_subtree(r["location_id"], loc["location_id"])}
ids.add(loc["location_id"])
rows = [(k, v) for k, v in unit.repo.stock(locations=ids).items() if v != ZERO]
if not rows:
return ""
(p, _l, lot, _t, _o), qty = rows[0]
return f"{qstr(qty)} {unit.base_unit(p)} of {p}".replace(" ", " ")
@INV.use_case(kind="command", route="POST /locations/archive", permission="inv:warehouse:write", idempotent=True)
def archive_location(caller: dict, location_id: Code) -> dict:
"""Archive a bin or zone nobody uses any more: no new move may name it, and every past move still reads the same. A bin that
still holds stock is refused (move it first); a virtual place (supplier, customer, loss, transit, opening) cannot be archived."""
@U.decide
def body(unit):
loc = unit.repo.get("inv_location", location_id=location_id)
if loc is None:
raise NotFound(f"no location {location_id}")
if loc["kind"] not in ("internal", "view") or loc["location_id"] == loc["warehouse_id"]:
raise RuleBroken(f"{location_id} cannot be renamed, archived or counted: it is a virtual place or a warehouse root")
held = _held(unit, loc)
if held:
raise RuleBroken(f"{location_id} holds {held}: move it first")
if loc["kind"] == "view":
kids = [r["location_id"] for r in unit.repo.locations_in(loc["warehouse_id"])
if r["parent_location_id"] == location_id and r["status"] == "active"]
if kids:
raise RuleBroken(f"{location_id} still has {kids[0]}{' and others' if len(kids) > 1 else ''}: archive those first")
unit.put("inv_location", {"location_id": location_id, "status": "archived"}, "location_id")
return {"location_id": location_id, "status": "archived"}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /warehouses")
def list_warehouses(caller: dict) -> dict:
"""The warehouses (everyone with stock access sees every site: floor staff move between them)."""
need_read(caller)
with U.read() as unit:
return {"warehouses": [{"warehouse_id": w["warehouse_id"], "name": w["name"], "company_code": w["company_code"],
"status": w["status"]} for w in unit.repo.find("inv_warehouse", order=("warehouse_id",))]}
@INV.use_case(kind="query", route="GET /warehouses/{warehouse_id}")
def get_warehouse(caller: dict, warehouse_id: Code) -> dict:
"""One warehouse: {warehouse_id, company_code, status} (product validates a site row against it), with its locations."""
need_read(caller)
with U.read() as unit:
w = unit.repo.get("inv_warehouse", warehouse_id=warehouse_id)
if w is None:
raise Refused(404, f"no warehouse {warehouse_id}")
locs = [location_view(r) for r in unit.repo.locations_in(warehouse_id)]
return {"warehouse_id": w["warehouse_id"], "company_code": w["company_code"], "status": w["status"],
"name": w["name"], "address": w["address"], "locations": locs}
@INV.use_case(kind="query", route="GET /locations")
def list_locations(caller: dict, usage: Optional[Code] = None, warehouse_id: Optional[Code] = None) -> dict:
"""The location picker: usage `internal` (the bins stock is kept in) or `receiving` (where goods-in puts a delivery), of one
warehouse or all, as {location_id, warehouse_id, name, usage, default}. Archived places and virtual ones are never offered."""
need_read(caller)
with U.read() as unit:
if usage not in (None, "internal", "receiving"):
raise Refused(422, "usage is internal or receiving")
rows = [r for r in unit.repo.find("inv_location", order=("pick_sequence", "location_id"))
if r["kind"] == "internal" and r["status"] == "active" and (not warehouse_id or r["warehouse_id"] == warehouse_id)]
two_step = unit.settings.value("receipt_steps", unit.today) == "two"
out, seen = [], set()
for r in rows:
wid = r["warehouse_id"]
is_in = r["location_id"] == f"{wid}/IN"
if usage == "receiving" and two_step and not is_in and any(x["location_id"] == f"{wid}/IN" for x in rows):
continue
default = (is_in if two_step and any(x["location_id"] == f"{wid}/IN" for x in rows) else wid not in seen)
seen.add(wid)
out.append({"location_id": r["location_id"], "warehouse_id": wid, "name": r["name"], "usage": "internal",
"default": bool(default)})
return {"locations": out}
@INV.use_case(kind="command", route="POST /stocked-products", permission="inv:product:write", idempotent=True)
def set_tracking(caller: dict, product_id: Code, tracking: Code, removal_strategy: Optional[Code] = None,
shelf_life_days: Optional[int] = None) -> dict:
"""Say how a product is tracked from today: none, lot (batch) or serial, and how lots are picked (fifo, fefo, manual) and
how long a lot lasts. A service is not stocked. Lots and serials must be turned on by the owner first. A product with stock
on hand cannot change its tracking: the answer names the quantity and where it is."""
@U.decide
def body(unit):
from brick_inventory.application import products
if tracking not in ("none", "lot", "serial"):
raise BadInput("tracking is none, lot or serial")
strategy = removal_strategy or "fifo"
if strategy not in ("fifo", "fefo", "manual"):
raise BadInput("removal_strategy is fifo, fefo or manual")
if shelf_life_days is not None and not 1 <= shelf_life_days <= 36500:
raise BadInput("shelf_life_days is between 1 and 36,500")
products.stocked(unit, product_id)
if tracking != "none" and not unit.settings.flag("lots_enabled", unit.today):
raise RuleBroken("lots and serials are off: the owner turns them on first (setting lots_enabled)")
current = unit.tracking(product_id)
if tracking != current["tracking"]:
on_hand = {k: v for k, v in unit.repo.stock(product_id=product_id).items()
if v > 0 and unit.location(k[1])["kind"] in ("internal", "transit", "consignment_customer", "subcontractor")}
if on_hand:
(_, loc, _, _, _), qty = next(iter(on_hand.items()))
raise RuleBroken(f"{product_id} has {qstr(qty)} {unit.base_unit(product_id)} on hand at {loc}: its tracking changes "
"only when none is left")
unit.put("inv_stocked_product", {"product_id": product_id, "valid_from": unit.today, "tracking": tracking,
"removal_strategy": strategy, "shelf_life_days": shelf_life_days,
"set_by": U.who(caller)}, ("product_id", "valid_from"))
return {"product_id": product_id, "tracking": tracking, "removal_strategy": strategy,
"shelf_life_days": shelf_life_days, "valid_from": unit.today.isoformat()}
return U.run(caller, body)
@INV.use_case(kind="command", route="POST /settings", idempotent=True)
def set_inventory_setting(caller: dict, key: Code, value: Short, valid_from: Optional[IsoDate] = None) -> dict:
"""Set how this business keeps stock, from a date (never back-dated). The controls on warehouse staff (negative stock, the
approval thresholds, lots and serials, the ledger mode, the valuation method, whether replenishment sends on its own, expired
picks, the AI step's mode) need inv:settings:controls; day-to-day settings need inv:settings:write. A scoped setting is
named `valuation_method:UK1`, `default_warehouse:UK1`, `pick_pack_days:WH1`. A change of valuation method or ledger mode
takes effect from the first day of a LATER month."""
@U.decide
def body(unit):
base = S.base_key(key)
need(caller, "inv:settings:controls" if base in S.CONTROLS else "inv:settings:write")
stored = S.check_value(key, value)
start = day(valid_from) or unit.today
if start < unit.today:
raise RuleBroken(f"settings start today or later, never back-dated: {start.isoformat()} is before {unit.today.isoformat()}")
if base in ("valuation_method", "ledger_mode") and (start.day != 1 or start <= unit.today):
raise RuleBroken(f"{base} changes from the first day of a later month: {start.isoformat()} is not one")
if base == "allow_negative_stock" and stored != "none":
for w in stored.split(","):
if unit.repo.get("inv_warehouse", warehouse_id=w) is None:
raise RuleBroken(f"there is no warehouse {w}")
if base in ("default_warehouse",) and stored and unit.repo.get("inv_warehouse", warehouse_id=stored) is None:
raise RuleBroken(f"there is no warehouse {stored}")
if base == "lots_enabled" and stored == "false":
on = [k for k in unit.repo.find("inv_stocked_product") if k["tracking"] != "none"]
if on:
raise RuleBroken(f"{on[0]['product_id']} is tracked by {on[0]['tracking']}: set it to none before turning lots off")
unit.put("inv_settings", {"key": key, "valid_from": start, "value": stored, "set_by": U.who(caller)}, ("key", "valid_from"))
return {"key": key, "value": stored, "valid_from": start.isoformat(), "set_by": U.who(caller)}
return U.run(caller, body)
@INV.use_case(kind="query", route="GET /settings")
def get_inventory_settings(caller: dict) -> dict:
"""The settings in force today, and the rows that will take effect later."""
need_read(caller)
with U.read() as unit:
s = unit.settings
keys = sorted({r["key"] for r in unit.repo.setting_rows()} | set(S.DEFAULTS))
now = {k: s.value(k, unit.today) for k in keys if ":" not in k}
scoped = {k: s.value(k, unit.today) for k in keys if ":" in k}
later = [{"key": r["key"], "value": r["value"], "valid_from": r["valid_from"].isoformat()}
for r in unit.repo.setting_rows() if r["valid_from"] > unit.today]
return {"today": unit.today.isoformat(), "settings": now, "scoped": scoped, "later": later}