dodil/erp-brick-inventory

DODIL ERP brick: inventory

src/brick_inventory/application/products.py

"""What the product module says about a product, and the exact conversion of a document quantity into its base unit.

A document line is written in the unit the person used (8 PK24); the ledger holds it in the product's BASE unit (144 LTR),
converted exactly by product's own route (numerator / denominator kept beside it). A conversion that is not exact within the
base unit's decimals is refused with product's reason - never rounded. A non-base unit that a line uses is told to product once
(it freezes that pack size for good).
"""
from decimal import Decimal
from typing import Optional

from erp_platform.bricks import Refused

from brick_inventory.application.unit import Unit
from brick_inventory.domain.errors import BadInput, RuleBroken
from brick_inventory.domain.values import Converted, check_quantity, converted, to_base_exact
from brick_inventory.infrastructure import seams


def stocked(unit: Unit, product_id: str, *, for_new_reservation: bool = False) -> dict:
    """The product, if it can be stocked: a service never is; an archived product may still be received back or counted,
    never newly reserved."""
    info = unit.product(product_id)
    if info.get("kind") == "service":
        raise RuleBroken(f"{product_id} is a service: services are not stocked")
    if for_new_reservation and info.get("lifecycle") in ("archived", "retired"):
        raise RuleBroken(f"{product_id} is archived: it is not reserved for new orders")
    return info


def unit_row(info: dict, unit_code: str) -> Optional[dict]:
    for u in info.get("units") or []:
        if u.get("unit_code") == unit_code:
            return u
    return None


def decimals_of(info: dict, unit_code: str) -> Optional[int]:
    row = unit_row(info, unit_code)
    return None if row is None or row.get("quantity_decimals") is None else int(row["quantity_decimals"])


def warehouse_unit(info: dict) -> str:
    return (info.get("roles") or {}).get("warehouse") or info.get("base_unit_code") or ""


def role_unit(info: dict, role: str) -> str:
    return (info.get("roles") or {}).get(role) or info.get("base_unit_code") or ""


def convert(unit: Unit, product_id: str, quantity, unit_code: str, *, document_ref: str = "", what: str = "a quantity",
            positive: bool = True) -> Converted:
    """`quantity` of `unit_code` in the product's base unit, exact. Calls product's convert route, then (for a unit other than
    the base) its unit-used route once."""
    info = stocked(unit, product_id)
    q = check_quantity(quantity, what, decimals=decimals_of(info, unit_code), positive=positive)
    if q == 0:
        return Converted(q, unit_code, 1, 1, Decimal(0), info.get("base_unit_code", ""))
    try:
        answer = seams.convert(product_id, q, unit_code, info.get("base_unit_code") or unit_code)
    except seams.Missing as m:
        if m.status in (404, 422):
            raise BadInput(f"{product_id} is not counted in {unit_code}" if "not counted" not in m.text else m.text) from None
        raise Refused(502, f"the product catalogue could not convert {unit_code} of {product_id}: {m.text}") from None
    conv = converted(q, unit_code, answer.get("base_unit_code") or info.get("base_unit_code", ""), answer["exact"])
    if unit_code != conv.base_unit_code and (product_id, unit_code) not in unit.used_units:
        try:
            seams.mark_unit_used(product_id, unit_code, document_ref or "inventory")
        except seams.Missing as m:
            raise RuleBroken(f"{unit_code} of {product_id} cannot be used on a document yet: {m.text}") from None
        unit.used_units.add((product_id, unit_code))
    return conv


def base_from_copied(quantity: Decimal, numerator: int, denominator: int) -> Decimal:
    """A quantity written in a unit whose conversion a document already copied: exact, no second call."""
    return to_base_exact(quantity, numerator, denominator)


def measures(info: dict, unit_code: str) -> dict:
    """The unit's weights and volume, per one of it, copied onto a document line when it is written (null = unknown)."""
    row = unit_row(info, unit_code) or {}
    m = row.get("measures") or {}
    return {"gross_weight": _d(m.get("gross_weight")), "net_weight": _d(m.get("net_weight")),
            "weight_unit_code": m.get("weight_unit_code"), "volume": _d(m.get("volume")),
            "volume_unit_code": m.get("volume_unit_code")}


def _d(v) -> Optional[Decimal]:
    return None if v in (None, "") else Decimal(str(v))