dodil/erp-brick-inventory

DODIL ERP brick: inventory

src/brick_inventory/application/inputs.py

"""The bounded inputs of this brick's use cases (BRICKS.md: an unbounded int or str is a promise the engine cannot keep).

A quantity is an exact decimal with up to four places below 10^12; a cost per unit a RATE up to four places; both arrive as
text. A value outside its bounds is a 422 from the door, never an engine error.
"""
from datetime import date
from decimal import Decimal, InvalidOperation
from typing import Annotated, Any, Optional

from erp_platform.types import IsoDate, IsoTime, Name, Note, text
from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, WithJsonSchema

Code = text(64, min_length=1)
Reason = text(500)
Ref = text(120, min_length=1)
Short = text(200)


def _decimal(v: Any, what: str, places: int) -> Decimal:
    if isinstance(v, bool) or isinstance(v, float):
        raise ValueError(f"{what} is an exact number: give it as a string such as \"0.75\"")
    try:
        d = v if isinstance(v, Decimal) else Decimal(str(v).strip())
    except (InvalidOperation, ValueError):
        raise ValueError(f"{v!r} is not a number") from None
    if not d.is_finite():
        raise ValueError(f"{what} is a finite number")
    if d.as_tuple().exponent < -places:
        raise ValueError(f"{what} has at most {places} decimal places")
    return d


def _qty(v: Any) -> Decimal:
    d = _decimal(v, "a quantity", 4)
    if not (Decimal(0) <= d < Decimal("1e12")):
        raise ValueError("a quantity is zero or more, and below 1,000,000,000,000")
    return d


def _signed_qty(v: Any) -> Decimal:
    d = _decimal(v, "a quantity", 4)
    if not (Decimal("-1e12") < d < Decimal("1e12")):
        raise ValueError("a quantity is below 1,000,000,000,000")
    return d


def _rate(v: Any) -> Decimal:
    d = _decimal(v, "a price", 4)
    if not (Decimal(0) <= d < Decimal("1e14")):
        raise ValueError("a price is zero or more, and below 100,000,000,000,000")
    return d


def _amount(v: Any) -> Decimal:
    d = _decimal(v, "an amount", 2)
    if not (Decimal("-1e15") < d < Decimal("1e15")):
        raise ValueError("an amount is below 1,000,000,000,000,000")
    return d


def _exact(pattern: str, description: str):
    return WithJsonSchema({"type": "string", "pattern": pattern, "description": description})


Quantity = Annotated[Decimal, BeforeValidator(_qty), _exact(r"^[0-9]{1,12}(\.[0-9]{1,4})?$", "a quantity, an exact decimal with up to 4 places (\"8\")")]
SignedQuantity = Annotated[Decimal, BeforeValidator(_signed_qty), _exact(r"^-?[0-9]{1,12}(\.[0-9]{1,4})?$", "a quantity, signed")]
Rate = Annotated[Decimal, BeforeValidator(_rate), _exact(r"^[0-9]{1,14}(\.[0-9]{1,4})?$", "a cost per unit, an exact decimal with up to 4 places")]
Amount = Annotated[Decimal, BeforeValidator(_amount), _exact(r"^-?[0-9]{1,15}(\.[0-9]{1,2})?$", "an amount with two places (\"30.00\")")]
Seq = Annotated[int, Field(ge=1, le=2_000_000_000)]
Count = Annotated[int, Field(ge=0, le=1_000_000)]


def day(v: Optional[str]) -> Optional[date]:
    return None if v in (None, "") else date.fromisoformat(v)


class _In(BaseModel):
    model_config = ConfigDict(extra="forbid")


class ScanLine(_In):
    """One line of a delivery scanned at goods-in: the purchase order's line, how much, in which unit, and the tracking."""
    line_no: Seq
    quantity: Quantity
    unit_code: Code
    lot_code: Optional[Code] = None
    expires_on: Optional[IsoDate] = None
    serials: Optional[list[Code]] = Field(default=None, max_length=10_000)


class ReceiptLineIn(_In):
    """purchasing posts one received line (POST /receipts)."""
    move_ref: text(120, min_length=3)
    source_kind: text(32) = "purchase_receipt"
    source_id: Ref
    purchase_order_id: Optional[Ref] = None
    bp_id: Optional[Annotated[int, Field(ge=1, le=2 ** 63 - 1)]] = None
    product_id: Code
    quantity: Quantity
    unit_code: Code
    location_id: Code
    received_on: IsoDate
    unit_cost: Optional[Rate] = None
    pricing_unit_code: Optional[Code] = None
    pricing_numerator: Optional[Annotated[int, Field(ge=1, le=2 ** 62)]] = None
    pricing_denominator: Optional[Annotated[int, Field(ge=1, le=2 ** 62)]] = None
    pricing_quantity: Optional[Quantity] = None
    currency: Optional[text(3)] = None
    stock_owner_bp_id: Optional[Annotated[int, Field(ge=1, le=2 ** 63 - 1)]] = None
    consume_provided: Optional[list["ProvidedIn"]] = Field(default=None, max_length=100)
    service_fee: Optional[Rate] = None
    delivery_costs: Optional[list["CostIn"]] = Field(default=None, max_length=50)
    lot_code: Optional[Code] = None
    expires_on: Optional[IsoDate] = None
    stock_type: Optional[Code] = None


class SourceLineIn(_In):
    """One line another module asks inventory to take out (POST /goods-issues/from-source): keyed by its own move_ref."""
    move_ref: text(120, min_length=3)
    product_id: Code
    quantity: Quantity
    unit_code: Code
    lot_code: Optional[Code] = None


class InputLotIn(_In):
    """An input lot an output lot was made from (INV-59): the quantity is in the input's base unit."""
    product_id: Code
    lot_code: Code
    quantity: Quantity


class ProvidedIn(_In):
    product_id: Code
    quantity: Quantity
    unit_code: Code


class CostIn(_In):
    kind: Code
    amount: Amount


class PickIn(_In):
    """What a picker took: a bin (and a lot), how much, in which unit."""
    line_no: Seq
    location_id: Code
    quantity: Quantity
    unit_code: Code
    lot_code: Optional[Code] = None
    stock_type: Optional[Code] = None


class CountEntryIn(_In):
    """One thing the counter wrote for a line, in the unit they counted in (5 PK24; 6 C62). A wrong entry is corrected by a
    negative one."""
    location_id: Code
    product_id: Code
    quantity: SignedQuantity
    unit_code: Code
    lot_code: Optional[Code] = None
    stock_type: Optional[Code] = None


class IssueLineIn(_In):
    product_id: Code
    quantity: Quantity
    unit_code: Optional[Code] = None
    from_location_id: Code
    lot_code: Optional[Code] = None
    stock_type: Optional[Code] = None


class ReservationLineIn(_In):
    line_no: Seq
    product_id: Code
    unit_code: Code
    quantity: Quantity
    unit_numerator: Optional[Annotated[int, Field(ge=1, le=2 ** 62)]] = None
    unit_denominator: Optional[Annotated[int, Field(ge=1, le=2 ** 62)]] = None
    warehouse_id: Optional[Code] = None       # sales-order SO-36: the delivering site rides on each promised line with its schedule
    schedule: Optional[list["ScheduleIn"]] = Field(default=None, max_length=100)


class ScheduleIn(_In):
    confirmed_on: IsoDate
    quantity_base: Quantity


class ExpectedReturnLine(_In):
    line_no: Seq
    product_id: Code
    unit_code: Code
    quantity: Quantity
    unit_numerator: Optional[Annotated[int, Field(ge=1, le=2 ** 62)]] = None
    unit_denominator: Optional[Annotated[int, Field(ge=1, le=2 ** 62)]] = None
    original_delivery_ref: Optional[Ref] = None
    serials: Optional[list[Code]] = Field(default=None, max_length=10_000)
    inspect: Optional[bool] = True


class PackIn(_In):
    package_type_code: Code
    contents: list["PackContent"] = Field(max_length=500)


class PackContent(_In):
    product_id: Code
    lot_code: Optional[Code] = None
    quantity: Quantity
    unit_code: Code


class ReturnLineIn(_In):
    line_no: Seq
    product_id: Optional[Code] = None
    quantity: Quantity
    unit_code: Code
    serials: Optional[list[Code]] = Field(default=None, max_length=10_000)


class InspectionIn(_In):
    line_no: Seq
    decision: text(20)                    # restock | scrap | back_to_customer
    quantity: Quantity
    unit_code: Code
    reason: Optional[Reason] = None


ReceiptLineIn.model_rebuild()
ReservationLineIn.model_rebuild()
PackIn.model_rebuild()



class TransferLineIn(_In):
    """One product moved: how much, in which unit, from which lot, and (to hold or release it) the stock types."""
    product_id: Code
    quantity: Quantity
    unit_code: Code
    lot_code: Optional[Code] = None
    from_stock_type: Optional[Code] = None
    to_stock_type: Optional[Code] = None


class ReceiveLineIn(_In):
    line_no: Seq
    quantity: Quantity
    unit_code: Code

__all__ = [n for n in dir() if not n.startswith("_")] + ["Name", "Note", "IsoDate", "IsoTime", "text"]