"""Module 2 — equipment inventory, distribution, collection, stock balance. Two shapes of inventory share one table: * **Serialized** types (ONU, router, switch) — one row per physical unit, quantity 1, and the row itself changes holder when issued. * **Bulk** types (cable, connectors) — a row is a quantity of one condition at one location. Issuing decrements it; the units in the field are tracked by the distribution's outstanding quantity, never by a shadow inventory row. So `equipment_items` rows for bulk types always sit at a location, and "who holds what" is one query over open distributions either way. """ from datetime import date, datetime from decimal import Decimal from pydantic import BaseModel, ConfigDict, Field, model_validator from app.models.enums import ( DistributionStatus, EquipmentCondition, EquipmentItemStatus, HolderType, StockMovementType, ) from app.schemas.auth import UserBrief from app.schemas.customer import CustomerBrief from app.schemas.reference import EquipmentTypeOut, LocationOut from app.schemas.types import DhakaDateTime, TrimmedStr # --- inventory --- class EquipmentItemCreate(BaseModel): """Add stock. Serialized types take a serial number; bulk types take a quantity.""" equipment_type_id: int location_id: int model: TrimmedStr | None = Field(None, max_length=120) serial_number: TrimmedStr | None = Field(None, max_length=120) serial_numbers: list[TrimmedStr] | None = Field(None, description="Provide multiple for bulk serialized ingest") condition: EquipmentCondition = EquipmentCondition.NEW quantity: int = Field(1, gt=0) purchase_date: date | None = None unit_cost: Decimal | None = Field(None, ge=0) notes: str | None = None class EquipmentItemUpdate(BaseModel): """Descriptive fields only — quantity moves via adjust/issue/collect.""" model: TrimmedStr | None = Field(None, max_length=120) serial_number: TrimmedStr | None = Field(None, max_length=120) condition: EquipmentCondition | None = None purchase_date: date | None = None unit_cost: Decimal | None = Field(None, ge=0) notes: str | None = None class HolderOut(BaseModel): """Where an item is right now, flattened so a table row can render it directly.""" type: HolderType label: str location: LocationOut | None = None user: UserBrief | None = None customer: CustomerBrief | None = None @classmethod def from_item(cls, item) -> "HolderOut": if item.holder_type == HolderType.TECHNICIAN and item.holder_user: return cls( type=item.holder_type, label=item.holder_user.full_name, user=UserBrief.model_validate(item.holder_user), ) if item.holder_type == HolderType.CUSTOMER and item.holder_customer: return cls( type=item.holder_type, label=f"{item.holder_customer.full_name} ({item.holder_customer.customer_code})", customer=CustomerBrief.model_validate(item.holder_customer), ) return cls( type=item.holder_type, label=item.location.name if item.location else "Unassigned", location=LocationOut.model_validate(item.location) if item.location else None, ) class EquipmentItemListItem(BaseModel): model_config = ConfigDict(from_attributes=True) id: int equipment_type: EquipmentTypeOut model: str | None serial_number: str | None condition: EquipmentCondition quantity: int status: EquipmentItemStatus holder: HolderOut #: Units that could be issued right now (0 unless the row is in stock at a location). available_quantity: int created_at: DhakaDateTime class EquipmentItemDetail(EquipmentItemListItem): purchase_date: date | None unit_cost: Decimal | None notes: str | None updated_at: DhakaDateTime distributions: list["DistributionListItem"] = [] # --- distribution (issue) --- class DistributionCreate(BaseModel): equipment_item_id: int quantity: int = Field(1, gt=0) issued_to_type: HolderType issued_to_user_id: int | None = None issued_to_customer_id: int | None = None connection_application_id: int | None = None connection_id: int | None = None issued_at: datetime | None = None #: Defaults to the equipment type's `default_return_days` for technician issues. expected_return_date: date | None = None notes: str | None = None @model_validator(mode="after") def check_target(self) -> "DistributionCreate": if self.issued_to_type == HolderType.TECHNICIAN and self.issued_to_user_id is None: raise ValueError("issued_to_user_id is required when issuing to a technician") if self.issued_to_type == HolderType.CUSTOMER and self.issued_to_customer_id is None: raise ValueError("issued_to_customer_id is required when issuing to a customer") if self.issued_to_type == HolderType.WAREHOUSE: raise ValueError("issued_to_type must be 'technician' or 'customer'") return self class ItemBrief(BaseModel): model_config = ConfigDict(from_attributes=True) id: int equipment_type: EquipmentTypeOut model: str | None serial_number: str | None class DistributionListItem(BaseModel): model_config = ConfigDict(from_attributes=True) id: int item: ItemBrief quantity: int returned_quantity: int outstanding_quantity: int issued_to_type: HolderType issued_to_user: UserBrief | None issued_to_customer: CustomerBrief | None issued_by: UserBrief | None issued_at: DhakaDateTime expected_return_date: date | None status: DistributionStatus is_overdue: bool days_overdue: int class DistributionDetail(DistributionListItem): from_location: LocationOut | None connection_application_id: int | None connection_id: int | None notes: str | None created_at: DhakaDateTime collections: list["CollectionOut"] = [] # --- collection (return) --- class CollectionCreate(BaseModel): quantity: int = Field(1, gt=0) condition_on_return: EquipmentCondition = EquipmentCondition.USED returned_to_location_id: int collected_at: datetime | None = None notes: str | None = None class CollectionOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int distribution_id: int | None item: ItemBrief quantity: int condition_on_return: EquipmentCondition collected_at: DhakaDateTime collected_by: UserBrief | None returned_to_location: LocationOut notes: str | None # --- adjustments and write-offs --- class StockAdjust(BaseModel): """Manual correction after a physical count. Signed, so it can go either way.""" quantity_delta: int = Field(description="Signed change; must not be zero.") reason: TrimmedStr = Field(min_length=3, max_length=500) @model_validator(mode="after") def check_delta(self) -> "StockAdjust": if self.quantity_delta == 0: raise ValueError("quantity_delta must not be zero") return self class WriteOff(BaseModel): quantity: int = Field(1, gt=0) reason: TrimmedStr = Field(min_length=3, max_length=500) # --- stock balance and ledger --- class StockAtLocation(BaseModel): location: LocationOut quantity: int class StockByType(BaseModel): """One row of the live stock report, per equipment type.""" equipment_type: EquipmentTypeOut in_stock: int = Field(description="Issuable units held at locations.") quarantined: int = Field(description="Returned faulty, awaiting inspection.") issued_out: int = Field(description="Outstanding units with technicians and customers.") with_technicians: int with_customers: int written_off: int total: int = Field(description="in_stock + quarantined + issued_out.") reorder_level: int below_reorder: bool by_location: list[StockAtLocation] class MovementOut(BaseModel): model_config = ConfigDict(from_attributes=True) id: int equipment_type: EquipmentTypeOut equipment_item_id: int | None location: LocationOut | None movement_type: StockMovementType quantity_delta: int distribution_id: int | None collection_id: int | None moved_at: DhakaDateTime notes: str | None class EquipmentSummary(BaseModel): """Headline numbers for the dashboard and the module header.""" total_items: int in_stock: int issued_out: int quarantined: int faulty: int overdue_distributions: int types_below_reorder: int # Detail schemas reference list schemas declared further down the module. EquipmentItemDetail.model_rebuild() DistributionDetail.model_rebuild()