# services/recipes/app/models/recipes.py """ Recipe and Production Management models for Recipe Service Comprehensive recipe management, production tracking, and inventory consumption """ 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, List from shared.database.base import Base class RecipeStatus(enum.Enum): """Recipe lifecycle status""" DRAFT = "draft" ACTIVE = "active" TESTING = "testing" ARCHIVED = "archived" DISCONTINUED = "discontinued" class ProductionStatus(enum.Enum): """Production batch status""" PLANNED = "planned" IN_PROGRESS = "in_progress" COMPLETED = "completed" FAILED = "failed" CANCELLED = "cancelled" class MeasurementUnit(enum.Enum): """Units for recipe measurements""" GRAMS = "g" KILOGRAMS = "kg" MILLILITERS = "ml" LITERS = "l" CUPS = "cups" TABLESPOONS = "tbsp" TEASPOONS = "tsp" UNITS = "units" PIECES = "pieces" PERCENTAGE = "%" class ProductionPriority(enum.Enum): """Production batch priority levels""" LOW = "low" NORMAL = "normal" HIGH = "high" URGENT = "urgent" class Recipe(Base): """Master recipe definitions""" __tablename__ = "recipes" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) # Recipe identification name = Column(String(255), nullable=False, index=True) recipe_code = Column(String(100), nullable=True, index=True) version = Column(String(20), nullable=False, default="1.0") # Product association finished_product_id = Column(UUID(as_uuid=True), nullable=False, index=True) # Links to inventory ingredient with product_type=finished_product # Recipe details description = Column(Text, nullable=True) category = Column(String(100), nullable=True, index=True) # bread, pastries, cakes, etc. cuisine_type = Column(String(100), nullable=True) difficulty_level = Column(Integer, nullable=False, default=1) # 1-5 scale # Production metrics yield_quantity = Column(Float, nullable=False) # How many units this recipe produces yield_unit = Column(SQLEnum(MeasurementUnit), nullable=False) prep_time_minutes = Column(Integer, nullable=True) cook_time_minutes = Column(Integer, nullable=True) total_time_minutes = Column(Integer, nullable=True) rest_time_minutes = Column(Integer, nullable=True) # Rising time, cooling time, etc. # Cost and pricing estimated_cost_per_unit = Column(Numeric(10, 2), nullable=True) last_calculated_cost = Column(Numeric(10, 2), nullable=True) cost_calculation_date = Column(DateTime(timezone=True), nullable=True) target_margin_percentage = Column(Float, nullable=True) suggested_selling_price = Column(Numeric(10, 2), nullable=True) # Instructions and notes instructions = Column(JSONB, nullable=True) # Structured step-by-step instructions preparation_notes = Column(Text, nullable=True) storage_instructions = Column(Text, nullable=True) quality_standards = Column(Text, nullable=True) # Recipe metadata serves_count = Column(Integer, nullable=True) # How many people/portions nutritional_info = Column(JSONB, nullable=True) # Calories, protein, etc. allergen_info = Column(JSONB, nullable=True) # List of allergens dietary_tags = Column(JSONB, nullable=True) # vegan, gluten-free, etc. # Production settings batch_size_multiplier = Column(Float, nullable=False, default=1.0) # Standard batch multiplier minimum_batch_size = Column(Float, nullable=True) maximum_batch_size = Column(Float, nullable=True) optimal_production_temperature = Column(Float, nullable=True) # Celsius optimal_humidity = Column(Float, nullable=True) # Percentage # Quality control quality_check_points = Column(JSONB, nullable=True) # Key checkpoints during production quality_check_configuration = Column(JSONB, nullable=True) # Stage-based quality check config common_issues = Column(JSONB, nullable=True) # Known issues and solutions # Status and lifecycle status = Column(SQLEnum(RecipeStatus), nullable=False, default=RecipeStatus.DRAFT, index=True) is_seasonal = Column(Boolean, default=False) season_start_month = Column(Integer, nullable=True) # 1-12 season_end_month = Column(Integer, nullable=True) # 1-12 is_signature_item = Column(Boolean, default=False) # 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) updated_by = Column(UUID(as_uuid=True), nullable=True) # Relationships ingredients = relationship("RecipeIngredient", back_populates="recipe", cascade="all, delete-orphan") production_batches = relationship("ProductionBatch", back_populates="recipe", cascade="all, delete-orphan") __table_args__ = ( Index('idx_recipes_tenant_name', 'tenant_id', 'name'), Index('idx_recipes_tenant_product', 'tenant_id', 'finished_product_id'), Index('idx_recipes_status', 'tenant_id', 'status'), Index('idx_recipes_category', 'tenant_id', 'category', 'status'), Index('idx_recipes_seasonal', 'tenant_id', 'is_seasonal', 'season_start_month', 'season_end_month'), Index('idx_recipes_signature', 'tenant_id', 'is_signature_item', 'status'), ) 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), 'name': self.name, 'recipe_code': self.recipe_code, 'version': self.version, 'finished_product_id': str(self.finished_product_id), 'description': self.description, 'category': self.category, 'cuisine_type': self.cuisine_type, 'difficulty_level': self.difficulty_level, 'yield_quantity': self.yield_quantity, 'yield_unit': self.yield_unit.value if self.yield_unit else None, 'prep_time_minutes': self.prep_time_minutes, 'cook_time_minutes': self.cook_time_minutes, 'total_time_minutes': self.total_time_minutes, 'rest_time_minutes': self.rest_time_minutes, 'estimated_cost_per_unit': float(self.estimated_cost_per_unit) if self.estimated_cost_per_unit else None, 'last_calculated_cost': float(self.last_calculated_cost) if self.last_calculated_cost else None, 'cost_calculation_date': self.cost_calculation_date.isoformat() if self.cost_calculation_date else None, 'target_margin_percentage': self.target_margin_percentage, 'suggested_selling_price': float(self.suggested_selling_price) if self.suggested_selling_price else None, 'instructions': self.instructions, 'preparation_notes': self.preparation_notes, 'storage_instructions': self.storage_instructions, 'quality_standards': self.quality_standards, 'serves_count': self.serves_count, 'nutritional_info': self.nutritional_info, 'allergen_info': self.allergen_info, 'dietary_tags': self.dietary_tags, 'batch_size_multiplier': self.batch_size_multiplier, 'minimum_batch_size': self.minimum_batch_size, 'maximum_batch_size': self.maximum_batch_size, 'optimal_production_temperature': self.optimal_production_temperature, 'optimal_humidity': self.optimal_humidity, 'quality_check_points': self.quality_check_points, 'quality_check_configuration': self.quality_check_configuration, 'common_issues': self.common_issues, 'status': self.status.value if self.status else None, 'is_seasonal': self.is_seasonal, 'season_start_month': self.season_start_month, 'season_end_month': self.season_end_month, 'is_signature_item': self.is_signature_item, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, 'created_by': str(self.created_by) if self.created_by else None, 'updated_by': str(self.updated_by) if self.updated_by else None, } class RecipeIngredient(Base): """Ingredients required for each recipe""" __tablename__ = "recipe_ingredients" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) recipe_id = Column(UUID(as_uuid=True), ForeignKey('recipes.id'), nullable=False, index=True) ingredient_id = Column(UUID(as_uuid=True), nullable=False, index=True) # Links to inventory ingredients # Quantity specifications quantity = Column(Float, nullable=False) unit = Column(SQLEnum(MeasurementUnit), nullable=False) quantity_in_base_unit = Column(Float, nullable=True) # Converted to ingredient's base unit # Alternative measurements alternative_quantity = Column(Float, nullable=True) # e.g., "2 cups" vs "240ml" alternative_unit = Column(SQLEnum(MeasurementUnit), nullable=True) # Ingredient specifications preparation_method = Column(String(255), nullable=True) # "sifted", "room temperature", "chopped" ingredient_notes = Column(Text, nullable=True) # Special instructions for this ingredient is_optional = Column(Boolean, default=False) # Recipe organization ingredient_order = Column(Integer, nullable=False, default=1) # Order in recipe ingredient_group = Column(String(100), nullable=True) # "wet ingredients", "dry ingredients", etc. # Substitutions substitution_options = Column(JSONB, nullable=True) # Alternative ingredients substitution_ratio = Column(Float, nullable=True) # 1:1, 1:2, etc. # Cost tracking unit_cost = Column(Numeric(10, 2), nullable=True) total_cost = Column(Numeric(10, 2), nullable=True) cost_updated_at = Column(DateTime(timezone=True), nullable=True) # Relationships recipe = relationship("Recipe", back_populates="ingredients") __table_args__ = ( Index('idx_recipe_ingredients_recipe', 'recipe_id', 'ingredient_order'), Index('idx_recipe_ingredients_ingredient', 'ingredient_id'), Index('idx_recipe_ingredients_tenant', 'tenant_id', 'recipe_id'), Index('idx_recipe_ingredients_group', 'recipe_id', 'ingredient_group', 'ingredient_order'), ) 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), 'recipe_id': str(self.recipe_id), 'ingredient_id': str(self.ingredient_id), 'quantity': self.quantity, 'unit': self.unit.value if self.unit else None, 'quantity_in_base_unit': self.quantity_in_base_unit, 'alternative_quantity': self.alternative_quantity, 'alternative_unit': self.alternative_unit.value if self.alternative_unit else None, 'preparation_method': self.preparation_method, 'ingredient_notes': self.ingredient_notes, 'is_optional': self.is_optional, 'ingredient_order': self.ingredient_order, 'ingredient_group': self.ingredient_group, 'substitution_options': self.substitution_options, 'substitution_ratio': self.substitution_ratio, 'unit_cost': float(self.unit_cost) if self.unit_cost else None, 'total_cost': float(self.total_cost) if self.total_cost else None, 'cost_updated_at': self.cost_updated_at.isoformat() if self.cost_updated_at else None, } class ProductionBatch(Base): """Track production batches and inventory consumption""" __tablename__ = "production_batches" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) recipe_id = Column(UUID(as_uuid=True), ForeignKey('recipes.id'), nullable=False, index=True) # Batch identification batch_number = Column(String(100), nullable=False, index=True) production_date = Column(DateTime(timezone=True), nullable=False, index=True) planned_start_time = Column(DateTime(timezone=True), nullable=True) actual_start_time = Column(DateTime(timezone=True), nullable=True) planned_end_time = Column(DateTime(timezone=True), nullable=True) actual_end_time = Column(DateTime(timezone=True), nullable=True) # Production planning planned_quantity = Column(Float, nullable=False) actual_quantity = Column(Float, nullable=True) yield_percentage = Column(Float, nullable=True) # actual/planned * 100 batch_size_multiplier = Column(Float, nullable=False, default=1.0) # Production details status = Column(SQLEnum(ProductionStatus), nullable=False, default=ProductionStatus.PLANNED, index=True) priority = Column(SQLEnum(ProductionPriority), nullable=False, default=ProductionPriority.NORMAL) assigned_staff = Column(JSONB, nullable=True) # List of staff assigned to this batch production_notes = Column(Text, nullable=True) # Quality metrics quality_score = Column(Float, nullable=True) # 1-10 scale quality_notes = Column(Text, nullable=True) defect_rate = Column(Float, nullable=True) # Percentage of defective products rework_required = Column(Boolean, default=False) # Cost tracking planned_material_cost = Column(Numeric(10, 2), nullable=True) actual_material_cost = Column(Numeric(10, 2), nullable=True) labor_cost = Column(Numeric(10, 2), nullable=True) overhead_cost = Column(Numeric(10, 2), nullable=True) total_production_cost = Column(Numeric(10, 2), nullable=True) cost_per_unit = Column(Numeric(10, 2), nullable=True) # Environmental conditions production_temperature = Column(Float, nullable=True) production_humidity = Column(Float, nullable=True) oven_temperature = Column(Float, nullable=True) baking_time_minutes = Column(Integer, nullable=True) # Waste and efficiency waste_quantity = Column(Float, nullable=False, default=0.0) waste_reason = Column(String(255), nullable=True) efficiency_percentage = Column(Float, nullable=True) # Based on time vs planned # Sales integration customer_order_reference = Column(String(100), nullable=True) # If made to order pre_order_quantity = Column(Float, nullable=True) # Pre-sold quantity shelf_quantity = Column(Float, nullable=True) # For shelf/display # 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) completed_by = Column(UUID(as_uuid=True), nullable=True) # Relationships recipe = relationship("Recipe", back_populates="production_batches") ingredient_consumptions = relationship("ProductionIngredientConsumption", back_populates="production_batch", cascade="all, delete-orphan") __table_args__ = ( Index('idx_production_batches_tenant_date', 'tenant_id', 'production_date'), Index('idx_production_batches_recipe', 'recipe_id', 'production_date'), Index('idx_production_batches_status', 'tenant_id', 'status', 'production_date'), Index('idx_production_batches_batch_number', 'tenant_id', 'batch_number'), Index('idx_production_batches_priority', 'tenant_id', 'priority', 'planned_start_time'), ) 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), 'recipe_id': str(self.recipe_id), 'batch_number': self.batch_number, 'production_date': self.production_date.isoformat() if self.production_date else None, 'planned_start_time': self.planned_start_time.isoformat() if self.planned_start_time else None, 'actual_start_time': self.actual_start_time.isoformat() if self.actual_start_time else None, 'planned_end_time': self.planned_end_time.isoformat() if self.planned_end_time else None, 'actual_end_time': self.actual_end_time.isoformat() if self.actual_end_time else None, 'planned_quantity': self.planned_quantity, 'actual_quantity': self.actual_quantity, 'yield_percentage': self.yield_percentage, 'batch_size_multiplier': self.batch_size_multiplier, 'status': self.status.value if self.status else None, 'priority': self.priority.value if self.priority else None, 'assigned_staff': self.assigned_staff, 'production_notes': self.production_notes, 'quality_score': self.quality_score, 'quality_notes': self.quality_notes, 'defect_rate': self.defect_rate, 'rework_required': self.rework_required, 'planned_material_cost': float(self.planned_material_cost) if self.planned_material_cost else None, 'actual_material_cost': float(self.actual_material_cost) if self.actual_material_cost else None, 'labor_cost': float(self.labor_cost) if self.labor_cost else None, 'overhead_cost': float(self.overhead_cost) if self.overhead_cost else None, 'total_production_cost': float(self.total_production_cost) if self.total_production_cost else None, 'cost_per_unit': float(self.cost_per_unit) if self.cost_per_unit else None, 'production_temperature': self.production_temperature, 'production_humidity': self.production_humidity, 'oven_temperature': self.oven_temperature, 'baking_time_minutes': self.baking_time_minutes, 'waste_quantity': self.waste_quantity, 'waste_reason': self.waste_reason, 'efficiency_percentage': self.efficiency_percentage, 'customer_order_reference': self.customer_order_reference, 'pre_order_quantity': self.pre_order_quantity, 'shelf_quantity': self.shelf_quantity, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, 'created_by': str(self.created_by) if self.created_by else None, 'completed_by': str(self.completed_by) if self.completed_by else None, } class ProductionIngredientConsumption(Base): """Track actual ingredient consumption during production""" __tablename__ = "production_ingredient_consumption" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) production_batch_id = Column(UUID(as_uuid=True), ForeignKey('production_batches.id'), nullable=False, index=True) recipe_ingredient_id = Column(UUID(as_uuid=True), ForeignKey('recipe_ingredients.id'), nullable=False, index=True) ingredient_id = Column(UUID(as_uuid=True), nullable=False, index=True) # Links to inventory ingredients stock_id = Column(UUID(as_uuid=True), nullable=True, index=True) # Specific stock batch used # Consumption details planned_quantity = Column(Float, nullable=False) actual_quantity = Column(Float, nullable=False) unit = Column(SQLEnum(MeasurementUnit), nullable=False) variance_quantity = Column(Float, nullable=True) # actual - planned variance_percentage = Column(Float, nullable=True) # (actual - planned) / planned * 100 # Cost tracking unit_cost = Column(Numeric(10, 2), nullable=True) total_cost = Column(Numeric(10, 2), nullable=True) # Consumption details consumption_time = Column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc)) consumption_notes = Column(Text, nullable=True) staff_member = Column(UUID(as_uuid=True), nullable=True) # Quality and condition ingredient_condition = Column(String(50), nullable=True) # fresh, near_expiry, etc. quality_impact = Column(String(255), nullable=True) # Impact on final product quality substitution_used = Column(Boolean, default=False) substitution_details = Column(Text, nullable=True) # Relationships production_batch = relationship("ProductionBatch", back_populates="ingredient_consumptions") __table_args__ = ( Index('idx_consumption_batch', 'production_batch_id'), Index('idx_consumption_ingredient', 'ingredient_id', 'consumption_time'), Index('idx_consumption_tenant', 'tenant_id', 'consumption_time'), Index('idx_consumption_recipe_ingredient', 'recipe_ingredient_id'), Index('idx_consumption_stock', 'stock_id'), ) 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), 'production_batch_id': str(self.production_batch_id), 'recipe_ingredient_id': str(self.recipe_ingredient_id), 'ingredient_id': str(self.ingredient_id), 'stock_id': str(self.stock_id) if self.stock_id else None, 'planned_quantity': self.planned_quantity, 'actual_quantity': self.actual_quantity, 'unit': self.unit.value if self.unit else None, 'variance_quantity': self.variance_quantity, 'variance_percentage': self.variance_percentage, 'unit_cost': float(self.unit_cost) if self.unit_cost else None, 'total_cost': float(self.total_cost) if self.total_cost else None, 'consumption_time': self.consumption_time.isoformat() if self.consumption_time else None, 'consumption_notes': self.consumption_notes, 'staff_member': str(self.staff_member) if self.staff_member else None, 'ingredient_condition': self.ingredient_condition, 'quality_impact': self.quality_impact, 'substitution_used': self.substitution_used, 'substitution_details': self.substitution_details, } class ProductionSchedule(Base): """Production planning and scheduling""" __tablename__ = "production_schedules" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) tenant_id = Column(UUID(as_uuid=True), nullable=False, index=True) # Schedule details schedule_date = Column(DateTime(timezone=True), nullable=False, index=True) schedule_name = Column(String(255), nullable=True) # Production planning total_planned_batches = Column(Integer, nullable=False, default=0) total_planned_items = Column(Float, nullable=False, default=0.0) estimated_production_hours = Column(Float, nullable=True) estimated_material_cost = Column(Numeric(10, 2), nullable=True) # Schedule status is_published = Column(Boolean, default=False) is_completed = Column(Boolean, default=False) completion_percentage = Column(Float, nullable=True) # Planning constraints available_staff_hours = Column(Float, nullable=True) oven_capacity_hours = Column(Float, nullable=True) production_capacity_limit = Column(Float, nullable=True) # Notes and instructions schedule_notes = Column(Text, nullable=True) preparation_instructions = Column(Text, nullable=True) special_requirements = Column(JSONB, 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)) created_by = Column(UUID(as_uuid=True), nullable=True) published_by = Column(UUID(as_uuid=True), nullable=True) published_at = Column(DateTime(timezone=True), nullable=True) __table_args__ = ( Index('idx_production_schedules_tenant_date', 'tenant_id', 'schedule_date'), Index('idx_production_schedules_published', 'tenant_id', 'is_published', 'schedule_date'), Index('idx_production_schedules_completed', 'tenant_id', 'is_completed', 'schedule_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), 'schedule_date': self.schedule_date.isoformat() if self.schedule_date else None, 'schedule_name': self.schedule_name, 'total_planned_batches': self.total_planned_batches, 'total_planned_items': self.total_planned_items, 'estimated_production_hours': self.estimated_production_hours, 'estimated_material_cost': float(self.estimated_material_cost) if self.estimated_material_cost else None, 'is_published': self.is_published, 'is_completed': self.is_completed, 'completion_percentage': self.completion_percentage, 'available_staff_hours': self.available_staff_hours, 'oven_capacity_hours': self.oven_capacity_hours, 'production_capacity_limit': self.production_capacity_limit, 'schedule_notes': self.schedule_notes, 'preparation_instructions': self.preparation_instructions, 'special_requirements': self.special_requirements, 'created_at': self.created_at.isoformat() if self.created_at else None, 'updated_at': self.updated_at.isoformat() if self.updated_at else None, 'created_by': str(self.created_by) if self.created_by else None, 'published_by': str(self.published_by) if self.published_by else None, 'published_at': self.published_at.isoformat() if self.published_at else None, }