# services/production/app/services/quality_template_service.py """ Quality Check Template Service - Business Logic Layer Handles quality template operations with business rules and validation """ from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from typing import List, Optional, Tuple from uuid import UUID, uuid4 from datetime import datetime, timezone import structlog from app.models.production import QualityCheckTemplate, ProcessStage from app.schemas.quality_templates import QualityCheckTemplateCreate, QualityCheckTemplateUpdate from app.repositories.quality_template_repository import QualityTemplateRepository logger = structlog.get_logger() class QualityTemplateService: """Service for managing quality check templates with business logic""" def __init__(self, db: AsyncSession): self.db = db self.repository = QualityTemplateRepository(db) async def create_template( self, tenant_id: str, template_data: QualityCheckTemplateCreate ) -> QualityCheckTemplate: """ Create a new quality check template Business Rules: - Template code must be unique within tenant - Validates template configuration """ try: # Auto-generate template code if not provided if not template_data.template_code: template_data.template_code = await self._generate_template_code( tenant_id, template_data.check_type, template_data.name ) logger.info("Auto-generated template code", template_code=template_data.template_code, check_type=template_data.check_type) # Business Rule: Validate template code uniqueness if template_data.template_code: exists = await self.repository.check_template_code_exists( tenant_id, template_data.template_code ) if exists: raise ValueError(f"Template code '{template_data.template_code}' already exists") # Business Rule: Validate template configuration is_valid, errors = self._validate_template_configuration(template_data.dict()) if not is_valid: raise ValueError(f"Invalid template configuration: {', '.join(errors)}") # Create template via repository template_dict = template_data.dict() template_dict['id'] = uuid4() template_dict['tenant_id'] = UUID(tenant_id) template = await self.repository.create(template_dict) logger.info("Quality template created", template_id=str(template.id), tenant_id=tenant_id, template_code=template.template_code) return template except ValueError as e: logger.warning("Template creation validation failed", tenant_id=tenant_id, error=str(e)) raise except Exception as e: logger.error("Failed to create quality template", tenant_id=tenant_id, error=str(e)) raise async def get_templates( self, tenant_id: str, stage: Optional[ProcessStage] = None, check_type: Optional[str] = None, is_active: Optional[bool] = True, skip: int = 0, limit: int = 100 ) -> Tuple[List[QualityCheckTemplate], int]: """ Get quality check templates with filtering and pagination Business Rules: - Default to active templates only - Limit maximum results per page """ try: # Business Rule: Enforce maximum limit if limit > 1000: limit = 1000 logger.warning("Template list limit capped at 1000", tenant_id=tenant_id, requested_limit=limit) templates, total = await self.repository.get_templates_by_tenant( tenant_id=tenant_id, stage=stage, check_type=check_type, is_active=is_active, skip=skip, limit=limit ) logger.debug("Retrieved quality templates", tenant_id=tenant_id, total=total, returned=len(templates)) return templates, total except Exception as e: logger.error("Failed to get quality templates", tenant_id=tenant_id, error=str(e)) raise async def get_template( self, tenant_id: str, template_id: UUID ) -> Optional[QualityCheckTemplate]: """ Get a specific quality check template Business Rules: - Template must belong to tenant """ try: template = await self.repository.get_by_tenant_and_id(tenant_id, template_id) if template: logger.debug("Retrieved quality template", template_id=str(template_id), tenant_id=tenant_id) else: logger.warning("Quality template not found", template_id=str(template_id), tenant_id=tenant_id) return template except Exception as e: logger.error("Failed to get quality template", template_id=str(template_id), tenant_id=tenant_id, error=str(e)) raise async def update_template( self, tenant_id: str, template_id: UUID, template_data: QualityCheckTemplateUpdate ) -> Optional[QualityCheckTemplate]: """ Update a quality check template Business Rules: - Template must exist and belong to tenant - Template code must remain unique if changed - Validates updated configuration """ try: # Business Rule: Template must exist template = await self.repository.get_by_tenant_and_id(tenant_id, template_id) if not template: logger.warning("Cannot update non-existent template", template_id=str(template_id), tenant_id=tenant_id) return None # Business Rule: Validate template code uniqueness if being updated if template_data.template_code and template_data.template_code != template.template_code: exists = await self.repository.check_template_code_exists( tenant_id, template_data.template_code, exclude_id=template_id ) if exists: raise ValueError(f"Template code '{template_data.template_code}' already exists") # Business Rule: Validate updated configuration update_dict = template_data.dict(exclude_unset=True) if update_dict: # Merge with existing data for validation full_data = template.__dict__.copy() full_data.update(update_dict) is_valid, errors = self._validate_template_configuration(full_data) if not is_valid: raise ValueError(f"Invalid template configuration: {', '.join(errors)}") # Update via repository update_dict['updated_at'] = datetime.now(timezone.utc) updated_template = await self.repository.update(template_id, update_dict) logger.info("Quality template updated", template_id=str(template_id), tenant_id=tenant_id) return updated_template except ValueError as e: logger.warning("Template update validation failed", template_id=str(template_id), tenant_id=tenant_id, error=str(e)) raise except Exception as e: logger.error("Failed to update quality template", template_id=str(template_id), tenant_id=tenant_id, error=str(e)) raise async def delete_template( self, tenant_id: str, template_id: UUID ) -> bool: """ Delete a quality check template Business Rules: - Template must exist and belong to tenant - Consider soft delete for audit trail (future enhancement) """ try: # Business Rule: Template must exist template = await self.repository.get_by_tenant_and_id(tenant_id, template_id) if not template: logger.warning("Cannot delete non-existent template", template_id=str(template_id), tenant_id=tenant_id) return False # Business Rule: Check if template is in use before deletion # Check for quality checks using this template from app.models.production import QualityCheck from sqlalchemy import select, func usage_query = select(func.count(QualityCheck.id)).where( QualityCheck.template_id == template_id ) usage_result = await self.repository.session.execute(usage_query) usage_count = usage_result.scalar() or 0 if usage_count > 0: logger.warning("Cannot delete template in use", template_id=str(template_id), tenant_id=tenant_id, usage_count=usage_count) # Instead of deleting, soft delete by setting is_active = False template.is_active = False await self.repository.session.commit() logger.info("Quality template soft deleted (set to inactive)", template_id=str(template_id), tenant_id=tenant_id, usage_count=usage_count) return True # Template is not in use, safe to delete success = await self.repository.delete(template_id) if success: logger.info("Quality template deleted", template_id=str(template_id), tenant_id=tenant_id) else: logger.warning("Failed to delete quality template", template_id=str(template_id), tenant_id=tenant_id) return success except Exception as e: logger.error("Failed to delete quality template", template_id=str(template_id), tenant_id=tenant_id, error=str(e)) raise async def get_templates_for_stage( self, tenant_id: str, stage: ProcessStage, is_active: Optional[bool] = True ) -> List[QualityCheckTemplate]: """ Get all quality check templates applicable to a specific process stage Business Rules: - Returns templates ordered by criticality - Required templates come first """ try: templates = await self.repository.get_templates_for_stage( tenant_id=tenant_id, stage=stage, is_active=is_active ) logger.debug("Retrieved templates for stage", tenant_id=tenant_id, stage=stage.value, count=len(templates)) return templates except Exception as e: logger.error("Failed to get templates for stage", tenant_id=tenant_id, stage=stage.value if stage else None, error=str(e)) raise async def duplicate_template( self, tenant_id: str, template_id: UUID ) -> Optional[QualityCheckTemplate]: """ Duplicate an existing quality check template Business Rules: - Original template must exist - Duplicate gets modified name and code - All other attributes copied """ try: # Business Rule: Original must exist original = await self.repository.get_by_tenant_and_id(tenant_id, template_id) if not original: logger.warning("Cannot duplicate non-existent template", template_id=str(template_id), tenant_id=tenant_id) return None # Business Rule: Create duplicate with modified identifiers duplicate_data = { 'name': f"{original.name} (Copy)", 'template_code': f"{original.template_code}_copy" if original.template_code else None, 'check_type': original.check_type, 'category': original.category, 'description': original.description, 'instructions': original.instructions, 'parameters': original.parameters, 'thresholds': original.thresholds, 'scoring_criteria': original.scoring_criteria, 'is_active': original.is_active, 'is_required': original.is_required, 'is_critical': original.is_critical, 'weight': original.weight, 'min_value': original.min_value, 'max_value': original.max_value, 'target_value': original.target_value, 'unit': original.unit, 'tolerance_percentage': original.tolerance_percentage, 'applicable_stages': original.applicable_stages, 'created_by': original.created_by } create_data = QualityCheckTemplateCreate(**duplicate_data) duplicate = await self.create_template(tenant_id, create_data) logger.info("Quality template duplicated", original_id=str(template_id), duplicate_id=str(duplicate.id), tenant_id=tenant_id) return duplicate except Exception as e: logger.error("Failed to duplicate quality template", template_id=str(template_id), tenant_id=tenant_id, error=str(e)) raise async def get_templates_by_recipe_config( self, tenant_id: str, stage: ProcessStage, recipe_quality_config: dict ) -> List[QualityCheckTemplate]: """ Get quality check templates based on recipe configuration Business Rules: - Returns only active templates - Filters by template IDs specified in recipe config - Ordered by criticality """ try: # Business Rule: Extract template IDs from recipe config stage_config = recipe_quality_config.get('stages', {}).get(stage.value) if not stage_config: logger.debug("No quality config for stage", tenant_id=tenant_id, stage=stage.value) return [] template_ids = stage_config.get('template_ids', []) if not template_ids: logger.debug("No template IDs in config", tenant_id=tenant_id, stage=stage.value) return [] # Get templates by IDs via repository template_ids_uuid = [UUID(tid) for tid in template_ids] templates = await self.repository.get_templates_by_ids(tenant_id, template_ids_uuid) logger.debug("Retrieved templates by recipe config", tenant_id=tenant_id, stage=stage.value, count=len(templates)) return templates except Exception as e: logger.error("Failed to get templates by recipe config", tenant_id=tenant_id, stage=stage.value if stage else None, error=str(e)) raise async def _generate_template_code( self, tenant_id: str, check_type: str, template_name: str ) -> str: """ Generate unique template code for quality check template Format: TPL-{TYPE}-{SEQUENCE} Examples: - Product Quality → TPL-PQ-0001 - Process Hygiene → TPL-PH-0001 - Equipment → TPL-EQ-0001 - Safety → TPL-SA-0001 - Temperature Control → TPL-TC-0001 Following the same pattern as inventory SKU and order number generation """ try: # Map check_type to 2-letter prefix type_map = { 'product_quality': 'PQ', 'process_hygiene': 'PH', 'equipment': 'EQ', 'safety': 'SA', 'cleaning': 'CL', 'temperature': 'TC', 'documentation': 'DC' } # Get prefix from check_type, fallback to first 2 chars of name type_prefix = type_map.get(check_type.lower()) if not type_prefix: # Fallback: use first 2 chars of template name or check_type name_for_prefix = template_name or check_type type_prefix = name_for_prefix[:2].upper() if len(name_for_prefix) >= 2 else "TP" tenant_uuid = UUID(tenant_id) # Count existing templates with this prefix for this tenant stmt = select(func.count(QualityCheckTemplate.id)).where( QualityCheckTemplate.tenant_id == tenant_uuid, QualityCheckTemplate.template_code.like(f"TPL-{type_prefix}-%") ) result = await self.db.execute(stmt) count = result.scalar() or 0 # Generate sequential number sequence = count + 1 template_code = f"TPL-{type_prefix}-{sequence:04d}" logger.info("Generated template code", template_code=template_code, type_prefix=type_prefix, sequence=sequence, tenant_id=tenant_id) return template_code except Exception as e: logger.error("Error generating template code, using fallback", error=str(e), check_type=check_type) # Fallback to UUID-based code fallback_code = f"TPL-{uuid4().hex[:8].upper()}" logger.warning("Using fallback template code", template_code=fallback_code) return fallback_code def _validate_template_configuration( self, template_data: dict ) -> Tuple[bool, List[str]]: """ Validate quality check template configuration (business rules) Business Rules: - Measurement checks require unit - Min value must be less than max value - Visual checks require scoring criteria - Process stages must be valid """ errors = [] # Business Rule: Type-specific validation check_type = template_data.get('check_type') if check_type in ['measurement', 'temperature', 'weight']: if not template_data.get('unit'): errors.append(f"Unit is required for {check_type} checks") min_val = template_data.get('min_value') max_val = template_data.get('max_value') if min_val is not None and max_val is not None and min_val >= max_val: errors.append("Minimum value must be less than maximum value") # Business Rule: Visual checks need scoring criteria scoring = template_data.get('scoring_criteria', {}) if check_type == 'visual' and not scoring: errors.append("Visual checks require scoring criteria") # Business Rule: Validate process stages stages = template_data.get('applicable_stages', []) if stages: valid_stages = [stage.value for stage in ProcessStage] invalid_stages = [s for s in stages if s not in valid_stages] if invalid_stages: errors.append(f"Invalid process stages: {invalid_stages}") is_valid = len(errors) == 0 if not is_valid: logger.warning("Template configuration validation failed", check_type=check_type, errors=errors) return is_valid, errors