# services/inventory/app/models/inventory.py """ Inventory management models for Inventory Service Comprehensive inventory tracking, ingredient management, and supplier integration """ from sqlalchemy import Column, String, DateTime, Float, Integer, Text, Index, Boolean, Numeric, ForeignKey, Enum as SQLEnum from sqlalchemy.dialects.postgresql import UUID, JSONB from sqlalchemy.orm import relationship import uuid import enum from datetime import datetime, timezone from typing import Dict, Any, Optional from shared.database.base import Base class UnitOfMeasure(enum.Enum): """Standard units of measure for ingredients""" KILOGRAMS = "kg" GRAMS = "g" LITERS = "l" MILLILITERS = "ml" UNITS = "units" PIECES = "pcs" PACKAGES = "pkg" BAGS = "bags" BOXES = "boxes" class IngredientCategory(enum.Enum): """Bakery ingredient categories""" FLOUR = "flour" YEAST = "yeast" DAIRY = "dairy" EGGS = "eggs" SUGAR = "sugar" FATS = "fats" SALT = "salt" SPICES = "spices" ADDITIVES = "additives" PACKAGING = "packaging" CLEANING = "cleaning" OTHER = "other" class ProductCategory(enum.Enum): """Finished bakery product categories for retail/distribution model""" BREAD = "bread" CROISSANTS = "croissants" PASTRIES = "pastries" CAKES = "cakes" COOKIES = "cookies" MUFFINS = "muffins" SANDWICHES = "sandwiches" SEASONAL = "seasonal" BEVERAGES = "beverages" OTHER_PRODUCTS = "other_products" class ProductType(enum.Enum): """Type of product in inventory""" INGREDIENT = "ingredient" # Raw materials (flour, yeast, etc.) FINISHED_PRODUCT = "finished_product" # Ready-to-sell items (bread, croissants, etc.) class ProductionStage(enum.Enum): """Production stages for bakery products""" RAW_INGREDIENT = "raw_ingredient" # Basic ingredients (flour, yeast) PAR_BAKED = "par_baked" # Pre-baked items needing final baking FULLY_BAKED = "fully_baked" # Completed products ready for sale PREPARED_DOUGH = "prepared_dough" # Prepared but unbaked dough FROZEN_PRODUCT = "frozen_product" # Frozen intermediate products class StockMovementType(enum.Enum): """Types of inventory movements""" PURCHASE = "PURCHASE" PRODUCTION_USE = "PRODUCTION_USE" ADJUSTMENT = "ADJUSTMENT" WASTE = "WASTE" TRANSFER = "TRANSFER" RETURN = "RETURN" INITIAL_STOCK = "INITIAL_STOCK" TRANSFORMATION = "TRANSFORMATION" # Converting between production stages class Ingredient(Base): """Master catalog for ingredients and finished products""" __tablename__ = "ingredients" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) # Product identification name = Column(String(255), nullable=False, index=True) sku = Column(String(100), nullable=True, index=True) barcode = Column(String(50), nullable=True, index=True) # Product type and categories product_type = Column(SQLEnum(ProductType), nullable=False, default=ProductType.INGREDIENT, index=True) ingredient_category = Column(SQLEnum(IngredientCategory), nullable=True, index=True) # For ingredients product_category = Column(SQLEnum(ProductCategory), nullable=True, index=True) # For finished products subcategory = Column(String(100), nullable=True) # Product details description = Column(Text, nullable=True) brand = Column(String(100), nullable=True) # Brand or central baker name unit_of_measure = Column(SQLEnum(UnitOfMeasure), nullable=False) package_size = Column(Float, nullable=True) # Size per package/unit # Pricing and costs average_cost = Column(Numeric(10, 2), nullable=True) last_purchase_price = Column(Numeric(10, 2), nullable=True) standard_cost = Column(Numeric(10, 2), nullable=True) # Stock management low_stock_threshold = Column(Float, nullable=False, default=10.0) reorder_point = Column(Float, nullable=False, default=20.0) reorder_quantity = Column(Float, nullable=False, default=50.0) max_stock_level = Column(Float, nullable=True) # Shelf life (critical for finished products) - default values only shelf_life_days = Column(Integer, nullable=True) # Default shelf life - actual per batch display_life_hours = Column(Integer, nullable=True) # How long can be displayed (for fresh products) best_before_hours = Column(Integer, nullable=True) # Hours until best before (for same-day products) storage_instructions = Column(Text, nullable=True) # Finished product specific fields central_baker_product_code = Column(String(100), nullable=True) # Central baker's product code delivery_days = Column(String(20), nullable=True) # Days of week delivered (e.g., "Mon,Wed,Fri") minimum_order_quantity = Column(Float, nullable=True) # Minimum order from central baker pack_size = Column(Integer, nullable=True) # How many pieces per pack # Status is_active = Column(Boolean, default=True) is_perishable = Column(Boolean, default=False) allergen_info = Column(JSONB, nullable=True) # JSON array of allergens nutritional_info = Column(JSONB, nullable=True) # Nutritional information for finished products # Audit fields created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) updated_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) created_by = Column(UUID(as_uuid=True), nullable=True) # Relationships stock_items = relationship("Stock", back_populates="ingredient", cascade="all, delete-orphan") movement_items = relationship("StockMovement", back_populates="ingredient", cascade="all, delete-orphan") __table_args__ = ( Index('idx_ingredients_tenant_name', 'tenant_id', 'name', unique=True), Index('idx_ingredients_tenant_sku', 'tenant_id', 'sku'), Index('idx_ingredients_barcode', 'barcode'), Index('idx_ingredients_product_type', 'tenant_id', 'product_type', 'is_active'), Index('idx_ingredients_ingredient_category', 'tenant_id', 'ingredient_category', 'is_active'), Index('idx_ingredients_product_category', 'tenant_id', 'product_category', 'is_active'), Index('idx_ingredients_stock_levels', 'tenant_id', 'low_stock_threshold', 'reorder_point'), ) def to_dict(self) -> Dict[str, Any]: """Convert model to dictionary for API responses""" # Map to response schema format - use appropriate category based on product type category = None if self.product_type == ProductType.FINISHED_PRODUCT and self.product_category: # For finished products, use product_category category = self.product_category.value elif self.product_type == ProductType.INGREDIENT and self.ingredient_category: # For ingredients, use ingredient_category category = self.ingredient_category.value elif self.ingredient_category and self.ingredient_category != IngredientCategory.OTHER: # If ingredient_category is set and not 'OTHER', use it category = self.ingredient_category.value elif self.product_category: # Fall back to product_category if available category = self.product_category.value else: # Final fallback category = "other" return { 'id': str(self.id), 'tenant_id': str(self.tenant_id), 'name': self.name, 'sku': self.sku, 'barcode': self.barcode, 'product_type': self.product_type.value if self.product_type else None, 'category': category, # Map to what IngredientResponse expects 'ingredient_category': self.ingredient_category.value if self.ingredient_category else None, 'product_category': self.product_category.value if self.product_category else None, 'subcategory': self.subcategory, 'description': self.description, 'brand': self.brand, 'unit_of_measure': self.unit_of_measure.value if self.unit_of_measure else None, 'package_size': self.package_size, 'average_cost': float(self.average_cost) if self.average_cost else None, 'last_purchase_price': float(self.last_purchase_price) if self.last_purchase_price else None, 'standard_cost': float(self.standard_cost) if self.standard_cost else None, 'low_stock_threshold': self.low_stock_threshold, 'reorder_point': self.reorder_point, 'reorder_quantity': self.reorder_quantity, 'max_stock_level': self.max_stock_level, 'shelf_life_days': self.shelf_life_days, 'display_life_hours': self.display_life_hours, 'best_before_hours': self.best_before_hours, 'storage_instructions': self.storage_instructions, 'central_baker_product_code': self.central_baker_product_code, 'delivery_days': self.delivery_days, 'minimum_order_quantity': self.minimum_order_quantity, 'pack_size': self.pack_size, 'is_active': self.is_active if self.is_active is not None else True, 'is_perishable': self.is_perishable if self.is_perishable is not None else False, 'allergen_info': self.allergen_info, 'nutritional_info': self.nutritional_info, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else datetime.now(timezone.utc).isoformat(), 'created_by': str(self.created_by) if self.created_by else None, } class Stock(Base): """Current stock levels and batch tracking""" __tablename__ = "stock" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) ingredient_id = Column(UUID(as_uuid=True), ForeignKey('ingredients.id'), nullable=False, index=True) # Supplier association supplier_id = Column(UUID(as_uuid=True), nullable=True, index=True) # Stock identification batch_number = Column(String(100), nullable=True, index=True) lot_number = Column(String(100), nullable=True, index=True) supplier_batch_ref = Column(String(100), nullable=True) # Production stage tracking production_stage = Column(SQLEnum('raw_ingredient', 'par_baked', 'fully_baked', 'prepared_dough', 'frozen_product', name='productionstage', create_type=False), nullable=False, default='raw_ingredient', index=True) transformation_reference = Column(String(100), nullable=True, index=True) # Links related transformations # Quantities current_quantity = Column(Float, nullable=False, default=0.0) reserved_quantity = Column(Float, nullable=False, default=0.0) # Reserved for production available_quantity = Column(Float, nullable=False, default=0.0) # current - reserved # Dates received_date = Column(DateTime(timezone=True), nullable=True) expiration_date = Column(DateTime(timezone=True), nullable=True, index=True) best_before_date = Column(DateTime(timezone=True), nullable=True) # Stage-specific expiration tracking original_expiration_date = Column(DateTime(timezone=True), nullable=True) # Original batch expiration (for par-baked) transformation_date = Column(DateTime(timezone=True), nullable=True) # When product was transformed final_expiration_date = Column(DateTime(timezone=True), nullable=True) # Final product expiration after transformation # Cost tracking unit_cost = Column(Numeric(10, 2), nullable=True) total_cost = Column(Numeric(10, 2), nullable=True) # Location storage_location = Column(String(100), nullable=True) warehouse_zone = Column(String(50), nullable=True) shelf_position = Column(String(50), nullable=True) # Batch-specific storage requirements requires_refrigeration = Column(Boolean, default=False) requires_freezing = Column(Boolean, default=False) storage_temperature_min = Column(Float, nullable=True) # Celsius storage_temperature_max = Column(Float, nullable=True) # Celsius storage_humidity_max = Column(Float, nullable=True) # Percentage shelf_life_days = Column(Integer, nullable=True) # Batch-specific shelf life storage_instructions = Column(Text, nullable=True) # Batch-specific instructions # Status is_available = Column(Boolean, default=True) is_expired = Column(Boolean, default=False, index=True) quality_status = Column(String(20), default="good") # good, damaged, expired, quarantined # Audit fields created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) updated_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) # Relationships ingredient = relationship("Ingredient", back_populates="stock_items") __table_args__ = ( Index('idx_stock_tenant_ingredient', 'tenant_id', 'ingredient_id'), Index('idx_stock_expiration', 'tenant_id', 'expiration_date', 'is_available'), Index('idx_stock_batch', 'tenant_id', 'batch_number'), Index('idx_stock_low_levels', 'tenant_id', 'current_quantity', 'is_available'), Index('idx_stock_quality', 'tenant_id', 'quality_status', 'is_available'), Index('idx_stock_production_stage', 'tenant_id', 'production_stage', 'is_available'), Index('idx_stock_transformation', 'tenant_id', 'transformation_reference'), Index('idx_stock_final_expiration', 'tenant_id', 'final_expiration_date', 'is_available'), ) def to_dict(self) -> Dict[str, Any]: """Convert model to dictionary for API responses""" return { 'id': str(self.id), 'tenant_id': str(self.tenant_id), 'ingredient_id': str(self.ingredient_id), 'batch_number': self.batch_number, 'lot_number': self.lot_number, 'supplier_batch_ref': self.supplier_batch_ref, 'production_stage': self.production_stage if self.production_stage else None, 'transformation_reference': self.transformation_reference, 'current_quantity': self.current_quantity, 'reserved_quantity': self.reserved_quantity, 'available_quantity': self.available_quantity, 'received_date': self.received_date.isoformat() if self.received_date else None, 'expiration_date': self.expiration_date.isoformat() if self.expiration_date else None, 'best_before_date': self.best_before_date.isoformat() if self.best_before_date else None, 'original_expiration_date': self.original_expiration_date.isoformat() if self.original_expiration_date else None, 'transformation_date': self.transformation_date.isoformat() if self.transformation_date else None, 'final_expiration_date': self.final_expiration_date.isoformat() if self.final_expiration_date else None, 'unit_cost': float(self.unit_cost) if self.unit_cost else None, 'total_cost': float(self.total_cost) if self.total_cost else None, 'storage_location': self.storage_location, 'warehouse_zone': self.warehouse_zone, 'shelf_position': self.shelf_position, 'requires_refrigeration': self.requires_refrigeration, 'requires_freezing': self.requires_freezing, 'storage_temperature_min': self.storage_temperature_min, 'storage_temperature_max': self.storage_temperature_max, 'storage_humidity_max': self.storage_humidity_max, 'shelf_life_days': self.shelf_life_days, 'storage_instructions': self.storage_instructions, 'is_available': self.is_available, 'is_expired': self.is_expired, 'quality_status': self.quality_status, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, } class StockMovement(Base): """Track all stock movements for audit trail""" __tablename__ = "stock_movements" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) ingredient_id = Column(UUID(as_uuid=True), ForeignKey('ingredients.id'), nullable=False, index=True) stock_id = Column(UUID(as_uuid=True), ForeignKey('stock.id'), nullable=True, index=True) # Movement details movement_type = Column(SQLEnum('PURCHASE', 'PRODUCTION_USE', 'ADJUSTMENT', 'WASTE', 'TRANSFER', 'RETURN', 'INITIAL_STOCK', name='stockmovementtype', create_type=False), nullable=False, index=True) quantity = Column(Float, nullable=False) unit_cost = Column(Numeric(10, 2), nullable=True) total_cost = Column(Numeric(10, 2), nullable=True) # Balance tracking quantity_before = Column(Float, nullable=True) quantity_after = Column(Float, nullable=True) # References reference_number = Column(String(100), nullable=True, index=True) # PO number, production order, etc. supplier_id = Column(UUID(as_uuid=True), nullable=True, index=True) # Additional details notes = Column(Text, nullable=True) reason_code = Column(String(50), nullable=True) # spoilage, damage, theft, etc. # Timestamp movement_date = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), index=True) # Audit fields created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) created_by = Column(UUID(as_uuid=True), nullable=True) # Relationships ingredient = relationship("Ingredient", back_populates="movement_items") __table_args__ = ( Index('idx_movements_tenant_date', 'tenant_id', 'movement_date'), Index('idx_movements_tenant_ingredient', 'tenant_id', 'ingredient_id', 'movement_date'), Index('idx_movements_type', 'tenant_id', 'movement_type', 'movement_date'), Index('idx_movements_reference', 'reference_number'), Index('idx_movements_supplier', 'supplier_id', 'movement_date'), ) def to_dict(self) -> Dict[str, Any]: """Convert model to dictionary for API responses""" return { 'id': str(self.id), 'tenant_id': str(self.tenant_id), 'ingredient_id': str(self.ingredient_id), 'stock_id': str(self.stock_id) if self.stock_id else None, 'movement_type': self.movement_type if self.movement_type else None, 'quantity': self.quantity, 'unit_cost': float(self.unit_cost) if self.unit_cost else None, 'total_cost': float(self.total_cost) if self.total_cost else None, 'quantity_before': self.quantity_before, 'quantity_after': self.quantity_after, 'reference_number': self.reference_number, 'supplier_id': str(self.supplier_id) if self.supplier_id else None, 'notes': self.notes, 'reason_code': self.reason_code, 'movement_date': self.movement_date.isoformat() if self.movement_date else None, 'created_at': self.created_at.isoformat() if self.created_at else None, 'created_by': str(self.created_by) if self.created_by else None, } class ProductTransformation(Base): """Track product transformations (e.g., par-baked to fully baked)""" __tablename__ = "product_transformations" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) # Transformation details transformation_reference = Column(String(100), nullable=False, index=True) source_ingredient_id = Column(UUID(as_uuid=True), ForeignKey('ingredients.id'), nullable=False) target_ingredient_id = Column(UUID(as_uuid=True), ForeignKey('ingredients.id'), nullable=False) # Stage transformation source_stage = Column(SQLEnum('raw_ingredient', 'par_baked', 'fully_baked', 'prepared_dough', 'frozen_product', name='productionstage', create_type=False), nullable=False) target_stage = Column(SQLEnum('raw_ingredient', 'par_baked', 'fully_baked', 'prepared_dough', 'frozen_product', name='productionstage', create_type=False), nullable=False) # Quantities and conversion source_quantity = Column(Float, nullable=False) # Input quantity target_quantity = Column(Float, nullable=False) # Output quantity conversion_ratio = Column(Float, nullable=False, default=1.0) # target/source ratio # Expiration logic expiration_calculation_method = Column(String(50), nullable=False, default="days_from_transformation") # days_from_transformation, preserve_original expiration_days_offset = Column(Integer, nullable=True) # Days from transformation date # Process tracking transformation_date = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) process_notes = Column(Text, nullable=True) performed_by = Column(UUID(as_uuid=True), nullable=True) # Batch tracking source_batch_numbers = Column(Text, nullable=True) # JSON array of source batch numbers target_batch_number = Column(String(100), nullable=True) # Status is_completed = Column(Boolean, default=True) is_reversed = Column(Boolean, default=False) # Audit fields created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) created_by = Column(UUID(as_uuid=True), nullable=True) __table_args__ = ( Index('idx_transformations_tenant_date', 'tenant_id', 'transformation_date'), Index('idx_transformations_reference', 'transformation_reference'), Index('idx_transformations_source', 'tenant_id', 'source_ingredient_id'), Index('idx_transformations_target', 'tenant_id', 'target_ingredient_id'), Index('idx_transformations_stages', 'source_stage', 'target_stage'), ) def to_dict(self) -> Dict[str, Any]: """Convert model to dictionary for API responses""" return { 'id': str(self.id), 'tenant_id': str(self.tenant_id), 'transformation_reference': self.transformation_reference, 'source_ingredient_id': str(self.source_ingredient_id), 'target_ingredient_id': str(self.target_ingredient_id), 'source_stage': self.source_stage if self.source_stage else None, 'target_stage': self.target_stage if self.target_stage else None, 'source_quantity': self.source_quantity, 'target_quantity': self.target_quantity, 'conversion_ratio': self.conversion_ratio, 'expiration_calculation_method': self.expiration_calculation_method, 'expiration_days_offset': self.expiration_days_offset, 'transformation_date': self.transformation_date.isoformat() if self.transformation_date else None, 'process_notes': self.process_notes, 'performed_by': str(self.performed_by) if self.performed_by else None, 'source_batch_numbers': self.source_batch_numbers, 'target_batch_number': self.target_batch_number, 'is_completed': self.is_completed, 'is_reversed': self.is_reversed, 'created_at': self.created_at.isoformat() if self.created_at else None, 'created_by': str(self.created_by) if self.created_by else None, } class StockAlert(Base): """Automated stock alerts for low stock, expiration, etc.""" __tablename__ = "stock_alerts" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) ingredient_id = Column(UUID(as_uuid=True), ForeignKey('ingredients.id'), nullable=False, index=True) stock_id = Column(UUID(as_uuid=True), ForeignKey('stock.id'), nullable=True, index=True) # Alert details alert_type = Column(String(50), nullable=False, index=True) # low_stock, expiring_soon, expired, reorder severity = Column(String(20), nullable=False, default="medium") # low, medium, high, critical title = Column(String(255), nullable=False) message = Column(Text, nullable=False) # Alert data current_quantity = Column(Float, nullable=True) threshold_value = Column(Float, nullable=True) expiration_date = Column(DateTime(timezone=True), nullable=True) # Status is_active = Column(Boolean, default=True) is_acknowledged = Column(Boolean, default=False) acknowledged_by = Column(UUID(as_uuid=True), nullable=True) acknowledged_at = Column(DateTime(timezone=True), nullable=True) # Resolution is_resolved = Column(Boolean, default=False) resolved_by = Column(UUID(as_uuid=True), nullable=True) resolved_at = Column(DateTime(timezone=True), nullable=True) resolution_notes = Column(Text, nullable=True) # Audit fields created_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)) updated_at = Column(DateTime(timezone=True), default=lambda: datetime.now(timezone.utc), onupdate=lambda: datetime.now(timezone.utc)) __table_args__ = ( Index('idx_alerts_tenant_active', 'tenant_id', 'is_active', 'created_at'), Index('idx_alerts_type_severity', 'alert_type', 'severity', 'is_active'), Index('idx_alerts_ingredient', 'ingredient_id', 'is_active'), Index('idx_alerts_unresolved', 'tenant_id', 'is_resolved', 'is_active'), ) def to_dict(self) -> Dict[str, Any]: """Convert model to dictionary for API responses""" return { 'id': str(self.id), 'tenant_id': str(self.tenant_id), 'ingredient_id': str(self.ingredient_id), 'stock_id': str(self.stock_id) if self.stock_id else None, 'alert_type': self.alert_type, 'severity': self.severity, 'title': self.title, 'message': self.message, 'current_quantity': self.current_quantity, 'threshold_value': self.threshold_value, 'expiration_date': self.expiration_date.isoformat() if self.expiration_date else None, 'is_active': self.is_active, 'is_acknowledged': self.is_acknowledged, 'acknowledged_by': str(self.acknowledged_by) if self.acknowledged_by else None, 'acknowledged_at': self.acknowledged_at.isoformat() if self.acknowledged_at else None, 'is_resolved': self.is_resolved, 'resolved_by': str(self.resolved_by) if self.resolved_by else None, 'resolved_at': self.resolved_at.isoformat() if self.resolved_at else None, 'resolution_notes': self.resolution_notes, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, }