491 lines
16 KiB
Python
491 lines
16 KiB
Python
# services/production/app/api/quality_templates.py
|
|
"""
|
|
Quality Check Templates API - CRUD operations on quality check templates
|
|
"""
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Path, Query, status
|
|
from typing import Optional
|
|
from uuid import UUID
|
|
import structlog
|
|
|
|
from shared.auth.decorators import get_current_user_dep
|
|
from shared.auth.access_control import require_user_role
|
|
from shared.routing import RouteBuilder, RouteCategory
|
|
from app.core.database import get_db
|
|
from app.repositories.quality_template_repository import QualityTemplateRepository
|
|
from app.models.production import ProcessStage, QualityCheckTemplate
|
|
from app.schemas.quality_templates import (
|
|
QualityCheckTemplateCreate,
|
|
QualityCheckTemplateUpdate,
|
|
QualityCheckTemplateResponse,
|
|
QualityCheckTemplateList,
|
|
QualityCheckType
|
|
)
|
|
|
|
logger = structlog.get_logger()
|
|
route_builder = RouteBuilder('production')
|
|
router = APIRouter(tags=["quality-templates"])
|
|
|
|
|
|
# ===== Quality Template CRUD Endpoints =====
|
|
|
|
@router.get(
|
|
route_builder.build_base_route("quality-templates"),
|
|
response_model=QualityCheckTemplateList
|
|
)
|
|
async def list_quality_templates(
|
|
tenant_id: UUID = Path(...),
|
|
stage: Optional[ProcessStage] = Query(None, description="Filter by process stage"),
|
|
check_type: Optional[QualityCheckType] = Query(None, description="Filter by check type"),
|
|
is_active: Optional[bool] = Query(True, description="Filter by active status"),
|
|
skip: int = Query(0, ge=0, description="Number of templates to skip"),
|
|
limit: int = Query(100, ge=1, le=1000, description="Number of templates to return"),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db = Depends(get_db)
|
|
):
|
|
"""
|
|
List quality check templates with filtering and pagination
|
|
|
|
Filters:
|
|
- stage: Filter by applicable process stage
|
|
- check_type: Filter by type of quality check
|
|
- is_active: Filter by active status (default: True)
|
|
"""
|
|
try:
|
|
repo = QualityTemplateRepository(db)
|
|
|
|
templates, total = await repo.get_templates_by_tenant(
|
|
tenant_id=str(tenant_id),
|
|
stage=stage,
|
|
check_type=check_type.value if check_type else None,
|
|
is_active=is_active,
|
|
skip=skip,
|
|
limit=limit
|
|
)
|
|
|
|
logger.info("Retrieved quality templates",
|
|
tenant_id=str(tenant_id),
|
|
total=total,
|
|
filters={"stage": stage, "check_type": check_type, "is_active": is_active})
|
|
|
|
return QualityCheckTemplateList(
|
|
templates=[QualityCheckTemplateResponse.from_orm(t) for t in templates],
|
|
total=total,
|
|
skip=skip,
|
|
limit=limit
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("Error listing quality templates",
|
|
error=str(e), tenant_id=str(tenant_id))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to retrieve quality templates"
|
|
)
|
|
|
|
|
|
@router.post(
|
|
route_builder.build_base_route("quality-templates"),
|
|
response_model=QualityCheckTemplateResponse,
|
|
status_code=status.HTTP_201_CREATED
|
|
)
|
|
@require_user_role(['admin', 'owner', 'member'])
|
|
async def create_quality_template(
|
|
template_data: QualityCheckTemplateCreate,
|
|
tenant_id: UUID = Path(...),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db = Depends(get_db)
|
|
):
|
|
"""Create a new quality check template"""
|
|
try:
|
|
repo = QualityTemplateRepository(db)
|
|
|
|
# Check if template code already exists (if provided)
|
|
if template_data.template_code:
|
|
code_exists = await repo.check_template_code_exists(
|
|
tenant_id=str(tenant_id),
|
|
template_code=template_data.template_code
|
|
)
|
|
if code_exists:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Template code '{template_data.template_code}' already exists"
|
|
)
|
|
|
|
# Create template
|
|
template_dict = template_data.dict()
|
|
template_dict['tenant_id'] = str(tenant_id)
|
|
template_dict['created_by'] = UUID(current_user["sub"])
|
|
|
|
template = QualityCheckTemplate(**template_dict)
|
|
db.add(template)
|
|
await db.commit()
|
|
await db.refresh(template)
|
|
|
|
logger.info("Created quality template",
|
|
template_id=str(template.id),
|
|
template_name=template.name,
|
|
tenant_id=str(tenant_id))
|
|
|
|
return QualityCheckTemplateResponse.from_orm(template)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
await db.rollback()
|
|
logger.error("Error creating quality template",
|
|
error=str(e), tenant_id=str(tenant_id))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to create quality template"
|
|
)
|
|
|
|
|
|
@router.get(
|
|
route_builder.build_resource_detail_route("quality-templates", "template_id"),
|
|
response_model=QualityCheckTemplateResponse
|
|
)
|
|
async def get_quality_template(
|
|
tenant_id: UUID = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db = Depends(get_db)
|
|
):
|
|
"""Get a specific quality check template"""
|
|
try:
|
|
repo = QualityTemplateRepository(db)
|
|
|
|
template = await repo.get_by_tenant_and_id(
|
|
tenant_id=str(tenant_id),
|
|
template_id=template_id
|
|
)
|
|
|
|
if not template:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Quality template not found"
|
|
)
|
|
|
|
return QualityCheckTemplateResponse.from_orm(template)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
logger.error("Error getting quality template",
|
|
error=str(e),
|
|
template_id=str(template_id),
|
|
tenant_id=str(tenant_id))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to retrieve quality template"
|
|
)
|
|
|
|
|
|
@router.put(
|
|
route_builder.build_resource_detail_route("quality-templates", "template_id"),
|
|
response_model=QualityCheckTemplateResponse
|
|
)
|
|
@require_user_role(['admin', 'owner', 'member'])
|
|
async def update_quality_template(
|
|
template_data: QualityCheckTemplateUpdate,
|
|
tenant_id: UUID = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db = Depends(get_db)
|
|
):
|
|
"""Update a quality check template"""
|
|
try:
|
|
repo = QualityTemplateRepository(db)
|
|
|
|
# Get existing template
|
|
template = await repo.get_by_tenant_and_id(
|
|
tenant_id=str(tenant_id),
|
|
template_id=template_id
|
|
)
|
|
|
|
if not template:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Quality template not found"
|
|
)
|
|
|
|
# Check if template code already exists (if being updated)
|
|
if template_data.template_code and template_data.template_code != template.template_code:
|
|
code_exists = await repo.check_template_code_exists(
|
|
tenant_id=str(tenant_id),
|
|
template_code=template_data.template_code,
|
|
exclude_id=template_id
|
|
)
|
|
if code_exists:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Template code '{template_data.template_code}' already exists"
|
|
)
|
|
|
|
# Update template fields
|
|
update_data = template_data.dict(exclude_unset=True)
|
|
for field, value in update_data.items():
|
|
setattr(template, field, value)
|
|
|
|
await db.commit()
|
|
await db.refresh(template)
|
|
|
|
logger.info("Updated quality template",
|
|
template_id=str(template_id),
|
|
tenant_id=str(tenant_id))
|
|
|
|
return QualityCheckTemplateResponse.from_orm(template)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
await db.rollback()
|
|
logger.error("Error updating quality template",
|
|
error=str(e),
|
|
template_id=str(template_id),
|
|
tenant_id=str(tenant_id))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to update quality template"
|
|
)
|
|
|
|
|
|
@router.delete(
|
|
route_builder.build_resource_detail_route("quality-templates", "template_id"),
|
|
status_code=status.HTTP_204_NO_CONTENT
|
|
)
|
|
@require_user_role(['admin', 'owner'])
|
|
async def delete_quality_template(
|
|
tenant_id: UUID = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db = Depends(get_db)
|
|
):
|
|
"""
|
|
Delete a quality check template (soft delete by setting is_active to False)
|
|
|
|
Note: For safety, this performs a soft delete. Hard deletes would require
|
|
checking for dependencies in recipes and production batches.
|
|
"""
|
|
try:
|
|
repo = QualityTemplateRepository(db)
|
|
|
|
# Get existing template
|
|
template = await repo.get_by_tenant_and_id(
|
|
tenant_id=str(tenant_id),
|
|
template_id=template_id
|
|
)
|
|
|
|
if not template:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Quality template not found"
|
|
)
|
|
|
|
# Soft delete by marking as inactive
|
|
template.is_active = False
|
|
await db.commit()
|
|
|
|
logger.info("Deleted quality template (soft delete)",
|
|
template_id=str(template_id),
|
|
tenant_id=str(tenant_id))
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
await db.rollback()
|
|
logger.error("Error deleting quality template",
|
|
error=str(e),
|
|
template_id=str(template_id),
|
|
tenant_id=str(tenant_id))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to delete quality template"
|
|
)
|
|
|
|
|
|
# ===== Additional Quality Template Operations =====
|
|
|
|
@router.get(
|
|
route_builder.build_custom_route(
|
|
RouteCategory.BASE,
|
|
["quality-templates", "stages", "{stage}"]
|
|
),
|
|
response_model=QualityCheckTemplateList
|
|
)
|
|
async def get_templates_for_stage(
|
|
tenant_id: UUID = Path(...),
|
|
stage: ProcessStage = Path(...),
|
|
is_active: bool = Query(True, description="Filter by active status"),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db = Depends(get_db)
|
|
):
|
|
"""Get all quality templates applicable to a specific process stage"""
|
|
try:
|
|
repo = QualityTemplateRepository(db)
|
|
|
|
templates = await repo.get_templates_for_stage(
|
|
tenant_id=str(tenant_id),
|
|
stage=stage,
|
|
is_active=is_active
|
|
)
|
|
|
|
logger.info("Retrieved templates for stage",
|
|
tenant_id=str(tenant_id),
|
|
stage=stage,
|
|
count=len(templates))
|
|
|
|
return QualityCheckTemplateList(
|
|
templates=[QualityCheckTemplateResponse.from_orm(t) for t in templates],
|
|
total=len(templates),
|
|
skip=0,
|
|
limit=len(templates)
|
|
)
|
|
|
|
except Exception as e:
|
|
logger.error("Error getting templates for stage",
|
|
error=str(e),
|
|
stage=stage,
|
|
tenant_id=str(tenant_id))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to retrieve templates for stage"
|
|
)
|
|
|
|
|
|
@router.post(
|
|
route_builder.build_resource_action_route("quality-templates", "template_id", "duplicate"),
|
|
response_model=QualityCheckTemplateResponse,
|
|
status_code=status.HTTP_201_CREATED
|
|
)
|
|
@require_user_role(['admin', 'owner', 'member'])
|
|
async def duplicate_quality_template(
|
|
tenant_id: UUID = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
db = Depends(get_db)
|
|
):
|
|
"""Duplicate an existing quality check template"""
|
|
try:
|
|
repo = QualityTemplateRepository(db)
|
|
|
|
# Get existing template
|
|
original = await repo.get_by_tenant_and_id(
|
|
tenant_id=str(tenant_id),
|
|
template_id=template_id
|
|
)
|
|
|
|
if not original:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Quality template not found"
|
|
)
|
|
|
|
# Create duplicate
|
|
duplicate_data = {
|
|
'tenant_id': original.tenant_id,
|
|
'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': UUID(current_user["sub"])
|
|
}
|
|
|
|
duplicate = QualityCheckTemplate(**duplicate_data)
|
|
db.add(duplicate)
|
|
await db.commit()
|
|
await db.refresh(duplicate)
|
|
|
|
logger.info("Duplicated quality template",
|
|
original_id=str(template_id),
|
|
duplicate_id=str(duplicate.id),
|
|
tenant_id=str(tenant_id))
|
|
|
|
return QualityCheckTemplateResponse.from_orm(duplicate)
|
|
|
|
except HTTPException:
|
|
raise
|
|
except Exception as e:
|
|
await db.rollback()
|
|
logger.error("Error duplicating quality template",
|
|
error=str(e),
|
|
template_id=str(template_id),
|
|
tenant_id=str(tenant_id))
|
|
raise HTTPException(
|
|
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
|
|
detail="Failed to duplicate quality template"
|
|
)
|
|
|
|
|
|
@router.post(
|
|
route_builder.build_operations_route("quality-templates/validate"),
|
|
response_model=dict
|
|
)
|
|
@require_user_role(['admin', 'owner', 'member'])
|
|
async def validate_quality_template(
|
|
template_data: dict,
|
|
tenant_id: UUID = Path(...),
|
|
current_user: dict = Depends(get_current_user_dep),
|
|
):
|
|
"""
|
|
Validate quality template configuration without creating it
|
|
|
|
Returns validation result with any errors found
|
|
"""
|
|
try:
|
|
errors = []
|
|
|
|
# Basic validation
|
|
if not template_data.get('name'):
|
|
errors.append("Template name is required")
|
|
|
|
if not template_data.get('check_type'):
|
|
errors.append("Check type is required")
|
|
|
|
# Validate measurement fields
|
|
check_type = template_data.get('check_type')
|
|
if check_type in ['measurement', 'temperature', 'weight']:
|
|
if template_data.get('min_value') is not None and template_data.get('max_value') is not None:
|
|
if template_data['min_value'] >= template_data['max_value']:
|
|
errors.append("Minimum value must be less than maximum value")
|
|
|
|
# Validate weight
|
|
weight = template_data.get('weight', 1.0)
|
|
if weight < 0 or weight > 10:
|
|
errors.append("Weight must be between 0 and 10")
|
|
|
|
is_valid = len(errors) == 0
|
|
|
|
logger.info("Validated quality template",
|
|
tenant_id=str(tenant_id),
|
|
valid=is_valid,
|
|
error_count=len(errors))
|
|
|
|
return {
|
|
"valid": is_valid,
|
|
"errors": errors
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.error("Error validating quality template",
|
|
error=str(e), tenant_id=str(tenant_id))
|
|
return {
|
|
"valid": False,
|
|
"errors": [f"Validation error: {str(e)}"]
|
|
}
|