Files
bakery-ia/services/recipes/app/models/recipes.py
2025-12-14 11:58:14 +01:00

531 lines
26 KiB
Python

# 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"
MEDIUM = "medium"
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)
# 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_configuration = Column(JSONB, nullable=True) # Stage-based quality check config
# 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,
'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_configuration': self.quality_check_configuration,
'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.MEDIUM)
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,
}