484 lines
18 KiB
Python
484 lines
18 KiB
Python
# 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 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:
|
|
# 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
|
|
|
|
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
|