"""SQLAlchemy models for the INVENTORY domain (warehouses, inventory, transactions). Merged from the software team's shipped inventory-dashboard fork with the v2.6 fixes: - unique stock index uses COALESCE(brand_id, sentinel) so NULL brand is ONE concrete value (a plain unique index would allow duplicate unbranded rows). - schema is created by sql/migration.sql (fresh) / init_db.ensure_v26 (existing volumes) — NOT through SQLAlchemy metadata. """ import uuid from datetime import datetime from decimal import Decimal from sqlalchemy import ( Boolean, DateTime, Enum, ForeignKey, Numeric, Text, UniqueConstraint, Uuid, func, text, ) from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db.base import Base from app.models.catalog import _timestamptz, _uuid_pk inventory_transaction_reason_enum = Enum( "PURCHASE", "WITHDRAWAL", "RETURN", "TRANSFER_IN", "TRANSFER_OUT", "DAMAGE", "WASTE", "LOSS", "FOUND", "STOCK_COUNT", name="inventory_transaction_reason", native_enum=True, ) # The 10 supported transaction types. No OTHER / generic adjustment. INVENTORY_TRANSACTION_TYPES = ( "PURCHASE", "WITHDRAWAL", "RETURN", "TRANSFER_IN", "TRANSFER_OUT", "DAMAGE", "WASTE", "LOSS", "FOUND", "STOCK_COUNT", ) # Types that INCREASE inventory quantity. INCREASING_TYPES = frozenset({"PURCHASE", "RETURN", "FOUND", "TRANSFER_IN"}) # Types that DECREASE inventory quantity. DECREASING_TYPES = frozenset({"WITHDRAWAL", "DAMAGE", "WASTE", "LOSS", "TRANSFER_OUT"}) class Warehouse(Base): """warehouses — storage location bound to exactly one project.""" __tablename__ = "warehouses" id: Mapped[uuid.UUID] = _uuid_pk() project_id: Mapped[uuid.UUID] = mapped_column( ForeignKey("projects.id", ondelete="CASCADE"), nullable=False ) name: Mapped[str] = mapped_column(Text, nullable=False) description: Mapped[str | None] = mapped_column(Text) created_at: Mapped[datetime] = _timestamptz() project: Mapped["Project"] = relationship(lazy="raise") class InventoryItem(Base): """inventory — current stock snapshot, unique per (warehouse, material, brand, specs).""" __tablename__ = "inventory" # The DB unique index (uq_inventory_stock) uses COALESCE on brand_id — # this Table-level constraint is only a model mirror for tooling; the # real constraint lives in migration.sql / init_db.ensure_v26. __mapper_args__ = {"eager_defaults": True} # fetch onupdate=func.now() via RETURNING id: Mapped[uuid.UUID] = _uuid_pk() warehouse_id: Mapped[uuid.UUID] = mapped_column( ForeignKey("warehouses.id", ondelete="CASCADE"), nullable=False ) material_id: Mapped[uuid.UUID] = mapped_column( ForeignKey("materials.id", ondelete="RESTRICT"), nullable=False ) brand_id: Mapped[uuid.UUID | None] = mapped_column( ForeignKey("brands.id", ondelete="SET NULL") ) quantity: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False, server_default=text("0")) unit_id: Mapped[int] = mapped_column( ForeignKey("units.id", ondelete="RESTRICT"), nullable=False ) # Canonical technical specifications (sorted JSON). specs_hash is the # SHA256 of the sorted canonical JSON (or 'default' when empty). specs: Mapped[dict] = mapped_column(JSONB, nullable=False, server_default=text("'{}'::jsonb")) specs_hash: Mapped[str] = mapped_column(Text, nullable=False, server_default=text("'default'")) updated_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now(), onupdate=func.now() ) warehouse: Mapped["Warehouse"] = relationship(lazy="raise") material: Mapped["Material"] = relationship(lazy="raise") brand: Mapped["Brand | None"] = relationship(lazy="raise") unit: Mapped["Unit"] = relationship(lazy="raise") transactions: Mapped[list["InventoryTransaction"]] = relationship( back_populates="inventory", lazy="raise", passive_deletes=True, order_by="InventoryTransaction.transaction_date", ) class InventoryTransaction(Base): """inventory_transactions — full chronological ledger of stock changes. quantity is SIGNED: positive = increase (PURCHASE/RETURN/FOUND/TRANSFER_IN), negative = decrease (WITHDRAWAL/DAMAGE/WASTE/LOSS/TRANSFER_OUT). A transfer is recorded as two linked rows (TRANSFER_OUT in source, TRANSFER_IN in destination) sharing the same transfer_pair_id. receipt_allocation_id links an auto-posted PURCHASE back to its item_receipt_allocation — UNIQUE, so the same delivery never posts twice. """ __tablename__ = "inventory_transactions" __mapper_args__ = {"eager_defaults": True} # fetch server_default now() via RETURNING id: Mapped[uuid.UUID] = _uuid_pk() inventory_id: Mapped[uuid.UUID] = mapped_column( ForeignKey("inventory.id", ondelete="CASCADE"), nullable=False ) warehouse_id: Mapped[uuid.UUID] = mapped_column( ForeignKey("warehouses.id", ondelete="RESTRICT"), nullable=False ) receipt_allocation_id: Mapped[uuid.UUID | None] = mapped_column( ForeignKey("item_receipt_allocations.id", ondelete="SET NULL"), unique=True ) transaction_type: Mapped[str] = mapped_column( inventory_transaction_reason_enum, nullable=False ) quantity: Mapped[Decimal] = mapped_column(Numeric(18, 6), nullable=False) # STOCK_COUNT only: quantity before the count (recorded) and the counted # value (physical). NULL for every other type. recorded_qty: Mapped[Decimal | None] = mapped_column(Numeric(18, 6)) physical_qty: Mapped[Decimal | None] = mapped_column(Numeric(18, 6)) # NULL for a standalone entry; links TRANSFER_OUT to its TRANSFER_IN row. transfer_pair_id: Mapped[uuid.UUID | None] = mapped_column(Uuid(as_uuid=True)) performed_by: Mapped[str | None] = mapped_column(Text) # user display name (dev placeholder) delivered_to: Mapped[str | None] = mapped_column(Text) transaction_date: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() ) note: Mapped[str | None] = mapped_column(Text) created_at: Mapped[datetime] = _timestamptz() inventory: Mapped["InventoryItem"] = relationship(back_populates="transactions", lazy="raise") warehouse: Mapped["Warehouse"] = relationship(lazy="raise") # Re-export for convenience (mirrors the ERD enum name) __all__ = [ "Warehouse", "InventoryItem", "InventoryTransaction", "inventory_transaction_reason_enum", "INVENTORY_TRANSACTION_TYPES", "INCREASING_TYPES", "DECREASING_TYPES", ]