dodil/erp-brick-inventory

DODIL ERP brick: inventory

src/brick_inventory/application/docs.py

"""Documents (the paper a move is recorded on) and their numbers.

A number is claimed only when the document is DONE, with a plain INSERT on (series_id, number): never reused, no gap from a
cancelled document (it never had one). The number's shape is the setting number_format ({warehouse}/{kind}/{seq:5}).
"""
import json
import re

from brick_inventory.application.unit import Unit

KIND_CODES = {"receipt": "IN", "delivery": "OUT", "return": "RET", "transfer": "INT", "putaway": "INT", "type_change": "INT",
              "adjustment": "C", "issue": "GI", "pick_batch": "PB", "opening": "OPN", "revaluation": "REV"}
_SEQ = re.compile(r"\{seq(?::(\d+))?\}")


def render_number(fmt: str, warehouse_id: str, code: str, seq: int) -> str:
    out = fmt.replace("{warehouse}", warehouse_id).replace("{kind}", code)
    return _SEQ.sub(lambda m: str(seq).rjust(int(m.group(1) or 0), "0"), out)


def series_id(warehouse_id: str, code: str) -> str:
    return f"{warehouse_id}/{code}"


def claim_number(unit: Unit, warehouse_id: str, kind: str, document_id: str) -> str:
    """The next number of a series: last + 1, a plain INSERT that only one writer can win (23505 for the other)."""
    code = KIND_CODES[kind]
    sid = series_id(warehouse_id, code)
    n = unit.next_number(sid)
    unit.race("inv_number_series", {"series_id": sid, "number": n, "document_id": document_id,
                                    "claimed_at": unit.now_utc.replace(tzinfo=None)})
    return render_number(unit.settings.value("number_format", unit.today), warehouse_id, code, n)


def stage_document(unit: Unit, document_id: str, kind: str, warehouse_id: str, company_code: str, *, stored=None, **fields) -> dict:
    row = {"document_id": document_id, "kind": kind, "warehouse_id": warehouse_id, "company_code": company_code,
           "lifecycle": fields.pop("lifecycle", "done"), "reversed": False,
           "requested_by_email": (unit.caller.get("email") or "").strip().lower() or None,
           "requested_by_sub": (unit.caller.get("sub") or "").strip().lower() or None, **fields}
    unit.put("inv_document", row, "document_id", stored=stored)
    return row


def stage_line(unit: Unit, document_id: str, line_no: int, *, stored=None, **fields) -> dict:
    row = {"document_id": document_id, "line_no": line_no, **fields}
    unit.put("inv_document_line", row, ("document_id", "line_no"), stored=stored)
    return row


def warehouse_of(unit: Unit, location_id: str) -> str:
    return unit.location(location_id)["warehouse_id"]


def extra_of(row: dict) -> dict:
    return json.loads(row["extra"]) if row.get("extra") else {}


def dump(obj) -> str:
    return json.dumps(obj, default=str, separators=(",", ":"))