7de011621d
All checks were successful
Deploy to Production / deploy (push) Successful in 1m7s
- Add InspectionTemplate and InspectionTemplateItem models - Add 9 API endpoints for template CRUD and item management - Add Alembic migration for inspection_templates tables - Add 15 backend tests (39/39 total pass) - Add TemplateEditor component with item management UI - Add templates list, create, and edit pages - Add tab bar, template grid, form row CSS classes - Fix CORS middleware ordering in main.py - Move CORS middleware before router registration Co-Authored-By: Claude Opus 4 <noreply@anthropic.com>
534 lines
17 KiB
Python
534 lines
17 KiB
Python
from typing import List, Optional
|
|
from uuid import UUID
|
|
|
|
from fastapi import APIRouter, HTTPException, Depends, Path
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from src.database.config import get_db
|
|
from src.database.models import InspectionTemplate, InspectionTemplateItem, Machine
|
|
from src.auth.models import TokenData
|
|
from src.auth.dependencies import require_auth, verify_tenant_access
|
|
|
|
router = APIRouter(prefix="/api/{tenant_id}/templates", tags=["templates"])
|
|
|
|
|
|
VALID_SUBJECT_TYPES = ("equipment", "product")
|
|
VALID_SCHEDULE_TYPES = ("daily", "weekly", "monthly", "yearly", "ad_hoc")
|
|
VALID_INSPECTION_MODES = ("checklist", "measurement", "monitoring")
|
|
VALID_DATA_TYPES = ("numeric", "boolean", "text", "select")
|
|
|
|
|
|
class ItemCreate(BaseModel):
|
|
name: str
|
|
category: Optional[str] = None
|
|
data_type: str
|
|
unit: Optional[str] = None
|
|
spec_min: Optional[float] = None
|
|
spec_max: Optional[float] = None
|
|
warning_min: Optional[float] = None
|
|
warning_max: Optional[float] = None
|
|
trend_window: Optional[int] = None
|
|
select_options: Optional[list] = None
|
|
equipment_part_id: Optional[str] = None
|
|
is_required: bool = True
|
|
|
|
|
|
class ItemUpdate(BaseModel):
|
|
name: Optional[str] = None
|
|
category: Optional[str] = None
|
|
data_type: Optional[str] = None
|
|
unit: Optional[str] = None
|
|
spec_min: Optional[float] = None
|
|
spec_max: Optional[float] = None
|
|
warning_min: Optional[float] = None
|
|
warning_max: Optional[float] = None
|
|
trend_window: Optional[int] = None
|
|
select_options: Optional[list] = None
|
|
equipment_part_id: Optional[str] = None
|
|
is_required: Optional[bool] = None
|
|
|
|
|
|
class TemplateCreate(BaseModel):
|
|
name: str
|
|
subject_type: str
|
|
machine_id: Optional[str] = None
|
|
product_code: Optional[str] = None
|
|
schedule_type: str
|
|
inspection_mode: str = "measurement"
|
|
items: List[ItemCreate] = []
|
|
|
|
|
|
class TemplateUpdate(BaseModel):
|
|
name: Optional[str] = None
|
|
subject_type: Optional[str] = None
|
|
machine_id: Optional[str] = None
|
|
product_code: Optional[str] = None
|
|
schedule_type: Optional[str] = None
|
|
inspection_mode: Optional[str] = None
|
|
|
|
|
|
class ReorderRequest(BaseModel):
|
|
item_ids: List[str]
|
|
|
|
|
|
def _format_ts(val) -> Optional[str]:
|
|
if val is None:
|
|
return None
|
|
return val.isoformat() if hasattr(val, "isoformat") else str(val)
|
|
|
|
|
|
def _item_to_dict(item: InspectionTemplateItem) -> dict:
|
|
return {
|
|
"id": str(item.id),
|
|
"template_id": str(item.template_id),
|
|
"sort_order": item.sort_order,
|
|
"name": str(item.name),
|
|
"category": str(item.category) if item.category else None,
|
|
"data_type": str(item.data_type),
|
|
"unit": str(item.unit) if item.unit else None,
|
|
"spec_min": float(item.spec_min) if item.spec_min is not None else None,
|
|
"spec_max": float(item.spec_max) if item.spec_max is not None else None,
|
|
"warning_min": float(item.warning_min)
|
|
if item.warning_min is not None
|
|
else None,
|
|
"warning_max": float(item.warning_max)
|
|
if item.warning_max is not None
|
|
else None,
|
|
"trend_window": int(item.trend_window)
|
|
if item.trend_window is not None
|
|
else None,
|
|
"select_options": item.select_options,
|
|
"equipment_part_id": str(item.equipment_part_id)
|
|
if item.equipment_part_id
|
|
else None,
|
|
"is_required": bool(item.is_required),
|
|
"created_at": _format_ts(item.created_at),
|
|
}
|
|
|
|
|
|
def _template_to_dict(
|
|
t: InspectionTemplate, items_count: int = 0, include_items: bool = False
|
|
) -> dict:
|
|
result = {
|
|
"id": str(t.id),
|
|
"tenant_id": str(t.tenant_id),
|
|
"name": str(t.name),
|
|
"subject_type": str(t.subject_type),
|
|
"machine_id": str(t.machine_id) if t.machine_id else None,
|
|
"product_code": str(t.product_code) if t.product_code else None,
|
|
"schedule_type": str(t.schedule_type),
|
|
"inspection_mode": str(t.inspection_mode or "measurement"),
|
|
"version": int(t.version or 1),
|
|
"is_active": bool(t.is_active),
|
|
"items_count": items_count,
|
|
"created_at": _format_ts(t.created_at),
|
|
"updated_at": _format_ts(t.updated_at),
|
|
}
|
|
if include_items:
|
|
result["items"] = [_item_to_dict(i) for i in (t.items or [])]
|
|
return result
|
|
|
|
|
|
def _validate_subject_type(subject_type: str):
|
|
if subject_type not in VALID_SUBJECT_TYPES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"subject_type은 {', '.join(VALID_SUBJECT_TYPES)} 중 하나여야 합니다.",
|
|
)
|
|
|
|
|
|
def _validate_schedule_type(schedule_type: str):
|
|
if schedule_type not in VALID_SCHEDULE_TYPES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"schedule_type은 {', '.join(VALID_SCHEDULE_TYPES)} 중 하나여야 합니다.",
|
|
)
|
|
|
|
|
|
def _validate_inspection_mode(mode: str):
|
|
if mode not in VALID_INSPECTION_MODES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"inspection_mode는 {', '.join(VALID_INSPECTION_MODES)} 중 하나여야 합니다.",
|
|
)
|
|
|
|
|
|
def _validate_data_type(data_type: str):
|
|
if data_type not in VALID_DATA_TYPES:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"data_type은 {', '.join(VALID_DATA_TYPES)} 중 하나여야 합니다.",
|
|
)
|
|
|
|
|
|
async def _get_template(
|
|
db: AsyncSession, tenant_id: str, template_id: UUID, load_items: bool = False
|
|
) -> InspectionTemplate:
|
|
stmt = select(InspectionTemplate).where(
|
|
InspectionTemplate.id == template_id,
|
|
InspectionTemplate.tenant_id == tenant_id,
|
|
InspectionTemplate.is_active == True,
|
|
)
|
|
if load_items:
|
|
stmt = stmt.options(selectinload(InspectionTemplate.items))
|
|
result = await db.execute(stmt)
|
|
template = result.scalar_one_or_none()
|
|
if not template:
|
|
raise HTTPException(status_code=404, detail="검사 템플릿을 찾을 수 없습니다.")
|
|
return template
|
|
|
|
|
|
@router.get("")
|
|
async def list_templates(
|
|
tenant_id: str = Path(...),
|
|
subject_type: Optional[str] = None,
|
|
schedule_type: Optional[str] = None,
|
|
machine_id: Optional[str] = None,
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
|
|
stmt = (
|
|
select(
|
|
InspectionTemplate,
|
|
func.count(InspectionTemplateItem.id).label("items_count"),
|
|
)
|
|
.outerjoin(
|
|
InspectionTemplateItem,
|
|
InspectionTemplateItem.template_id == InspectionTemplate.id,
|
|
)
|
|
.where(
|
|
InspectionTemplate.tenant_id == tenant_id,
|
|
InspectionTemplate.is_active == True,
|
|
)
|
|
.group_by(InspectionTemplate.id)
|
|
.order_by(InspectionTemplate.name)
|
|
)
|
|
|
|
if subject_type:
|
|
stmt = stmt.where(InspectionTemplate.subject_type == subject_type)
|
|
if schedule_type:
|
|
stmt = stmt.where(InspectionTemplate.schedule_type == schedule_type)
|
|
if machine_id:
|
|
stmt = stmt.where(InspectionTemplate.machine_id == UUID(machine_id))
|
|
|
|
result = await db.execute(stmt)
|
|
rows = result.all()
|
|
|
|
return {"templates": [_template_to_dict(t, count) for t, count in rows]}
|
|
|
|
|
|
@router.post("")
|
|
async def create_template(
|
|
body: TemplateCreate,
|
|
tenant_id: str = Path(...),
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
_validate_subject_type(body.subject_type)
|
|
_validate_schedule_type(body.schedule_type)
|
|
_validate_inspection_mode(body.inspection_mode)
|
|
|
|
if body.subject_type == "equipment" and body.machine_id:
|
|
machine_stmt = select(Machine).where(
|
|
Machine.id == UUID(body.machine_id), Machine.tenant_id == tenant_id
|
|
)
|
|
if not (await db.execute(machine_stmt)).scalar_one_or_none():
|
|
raise HTTPException(status_code=404, detail="설비를 찾을 수 없습니다.")
|
|
|
|
template = InspectionTemplate(
|
|
tenant_id=tenant_id,
|
|
name=body.name,
|
|
subject_type=body.subject_type,
|
|
machine_id=UUID(body.machine_id) if body.machine_id else None,
|
|
product_code=body.product_code,
|
|
schedule_type=body.schedule_type,
|
|
inspection_mode=body.inspection_mode,
|
|
)
|
|
db.add(template)
|
|
await db.flush()
|
|
|
|
for idx, item_data in enumerate(body.items):
|
|
_validate_data_type(item_data.data_type)
|
|
item = InspectionTemplateItem(
|
|
template_id=template.id,
|
|
sort_order=idx + 1,
|
|
name=item_data.name,
|
|
category=item_data.category,
|
|
data_type=item_data.data_type,
|
|
unit=item_data.unit,
|
|
spec_min=item_data.spec_min,
|
|
spec_max=item_data.spec_max,
|
|
warning_min=item_data.warning_min,
|
|
warning_max=item_data.warning_max,
|
|
trend_window=item_data.trend_window,
|
|
select_options=item_data.select_options,
|
|
equipment_part_id=UUID(item_data.equipment_part_id)
|
|
if item_data.equipment_part_id
|
|
else None,
|
|
is_required=item_data.is_required,
|
|
)
|
|
db.add(item)
|
|
|
|
await db.commit()
|
|
|
|
stmt = (
|
|
select(InspectionTemplate)
|
|
.options(selectinload(InspectionTemplate.items))
|
|
.where(InspectionTemplate.id == template.id)
|
|
)
|
|
result = await db.execute(stmt)
|
|
created = result.scalar_one()
|
|
|
|
return _template_to_dict(created, len(created.items), include_items=True)
|
|
|
|
|
|
@router.get("/{template_id}")
|
|
async def get_template(
|
|
tenant_id: str = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
template = await _get_template(db, tenant_id, template_id, load_items=True)
|
|
return _template_to_dict(template, len(template.items), include_items=True)
|
|
|
|
|
|
@router.put("/{template_id}")
|
|
async def update_template(
|
|
body: TemplateUpdate,
|
|
tenant_id: str = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
template = await _get_template(db, tenant_id, template_id, load_items=True)
|
|
|
|
if body.subject_type is not None:
|
|
_validate_subject_type(body.subject_type)
|
|
template.subject_type = body.subject_type
|
|
if body.schedule_type is not None:
|
|
_validate_schedule_type(body.schedule_type)
|
|
template.schedule_type = body.schedule_type
|
|
if body.inspection_mode is not None:
|
|
_validate_inspection_mode(body.inspection_mode)
|
|
template.inspection_mode = body.inspection_mode
|
|
if body.name is not None:
|
|
template.name = body.name
|
|
if body.machine_id is not None:
|
|
template.machine_id = UUID(body.machine_id) if body.machine_id else None
|
|
if body.product_code is not None:
|
|
template.product_code = body.product_code
|
|
|
|
template.version = (template.version or 1) + 1
|
|
|
|
await db.commit()
|
|
await db.refresh(template)
|
|
|
|
return _template_to_dict(template, len(template.items), include_items=True)
|
|
|
|
|
|
@router.delete("/{template_id}")
|
|
async def delete_template(
|
|
tenant_id: str = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
template = await _get_template(db, tenant_id, template_id)
|
|
|
|
template.is_active = False
|
|
await db.commit()
|
|
|
|
return {"status": "success", "message": "검사 템플릿이 비활성화되었습니다."}
|
|
|
|
|
|
@router.post("/{template_id}/items")
|
|
async def create_item(
|
|
body: ItemCreate,
|
|
tenant_id: str = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
template = await _get_template(db, tenant_id, template_id, load_items=True)
|
|
_validate_data_type(body.data_type)
|
|
|
|
max_order = max((i.sort_order for i in template.items), default=0)
|
|
|
|
item = InspectionTemplateItem(
|
|
template_id=template.id,
|
|
sort_order=max_order + 1,
|
|
name=body.name,
|
|
category=body.category,
|
|
data_type=body.data_type,
|
|
unit=body.unit,
|
|
spec_min=body.spec_min,
|
|
spec_max=body.spec_max,
|
|
warning_min=body.warning_min,
|
|
warning_max=body.warning_max,
|
|
trend_window=body.trend_window,
|
|
select_options=body.select_options,
|
|
equipment_part_id=UUID(body.equipment_part_id)
|
|
if body.equipment_part_id
|
|
else None,
|
|
is_required=body.is_required,
|
|
)
|
|
db.add(item)
|
|
|
|
template.version = (template.version or 1) + 1
|
|
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
|
|
return _item_to_dict(item)
|
|
|
|
|
|
@router.put("/{template_id}/items/reorder")
|
|
async def reorder_items(
|
|
body: ReorderRequest,
|
|
tenant_id: str = Path(...),
|
|
template_id: UUID = Path(...),
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
template = await _get_template(db, tenant_id, template_id, load_items=True)
|
|
|
|
item_map = {str(i.id): i for i in template.items}
|
|
|
|
if set(body.item_ids) != set(item_map.keys()):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="item_ids는 해당 템플릿의 모든 항목 ID를 포함해야 합니다.",
|
|
)
|
|
|
|
for idx, item_id_str in enumerate(body.item_ids):
|
|
item_map[item_id_str].sort_order = idx + 1
|
|
|
|
template.version = (template.version or 1) + 1
|
|
|
|
await db.commit()
|
|
db.expire_all()
|
|
|
|
stmt = (
|
|
select(InspectionTemplate)
|
|
.options(selectinload(InspectionTemplate.items))
|
|
.where(InspectionTemplate.id == template_id)
|
|
)
|
|
result = await db.execute(stmt)
|
|
refreshed = result.scalar_one()
|
|
|
|
return _template_to_dict(refreshed, len(refreshed.items), include_items=True)
|
|
|
|
|
|
@router.put("/{template_id}/items/{item_id}")
|
|
async def update_item(
|
|
body: ItemUpdate,
|
|
tenant_id: str = Path(...),
|
|
template_id: UUID = Path(...),
|
|
item_id: UUID = Path(...),
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
await _get_template(db, tenant_id, template_id)
|
|
|
|
stmt = select(InspectionTemplateItem).where(
|
|
InspectionTemplateItem.id == item_id,
|
|
InspectionTemplateItem.template_id == template_id,
|
|
)
|
|
result = await db.execute(stmt)
|
|
item = result.scalar_one_or_none()
|
|
if not item:
|
|
raise HTTPException(status_code=404, detail="검사 항목을 찾을 수 없습니다.")
|
|
|
|
if body.name is not None:
|
|
item.name = body.name
|
|
if body.category is not None:
|
|
item.category = body.category
|
|
if body.data_type is not None:
|
|
_validate_data_type(body.data_type)
|
|
item.data_type = body.data_type
|
|
if body.unit is not None:
|
|
item.unit = body.unit
|
|
if body.spec_min is not None:
|
|
item.spec_min = body.spec_min
|
|
if body.spec_max is not None:
|
|
item.spec_max = body.spec_max
|
|
if body.warning_min is not None:
|
|
item.warning_min = body.warning_min
|
|
if body.warning_max is not None:
|
|
item.warning_max = body.warning_max
|
|
if body.trend_window is not None:
|
|
item.trend_window = body.trend_window
|
|
if body.select_options is not None:
|
|
item.select_options = body.select_options
|
|
if body.equipment_part_id is not None:
|
|
item.equipment_part_id = (
|
|
UUID(body.equipment_part_id) if body.equipment_part_id else None
|
|
)
|
|
if body.is_required is not None:
|
|
item.is_required = body.is_required
|
|
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
|
|
return _item_to_dict(item)
|
|
|
|
|
|
@router.delete("/{template_id}/items/{item_id}")
|
|
async def delete_item(
|
|
tenant_id: str = Path(...),
|
|
template_id: UUID = Path(...),
|
|
item_id: UUID = Path(...),
|
|
current_user: TokenData = Depends(require_auth),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
verify_tenant_access(tenant_id, current_user)
|
|
await _get_template(db, tenant_id, template_id)
|
|
|
|
stmt = select(InspectionTemplateItem).where(
|
|
InspectionTemplateItem.id == item_id,
|
|
InspectionTemplateItem.template_id == template_id,
|
|
)
|
|
result = await db.execute(stmt)
|
|
item = result.scalar_one_or_none()
|
|
if not item:
|
|
raise HTTPException(status_code=404, detail="검사 항목을 찾을 수 없습니다.")
|
|
|
|
deleted_order = item.sort_order
|
|
await db.delete(item)
|
|
await db.flush()
|
|
|
|
remaining_stmt = (
|
|
select(InspectionTemplateItem)
|
|
.where(
|
|
InspectionTemplateItem.template_id == template_id,
|
|
InspectionTemplateItem.sort_order > deleted_order,
|
|
)
|
|
.order_by(InspectionTemplateItem.sort_order)
|
|
)
|
|
remaining_result = await db.execute(remaining_stmt)
|
|
for remaining_item in remaining_result.scalars().all():
|
|
remaining_item.sort_order -= 1
|
|
|
|
template_stmt = select(InspectionTemplate).where(
|
|
InspectionTemplate.id == template_id
|
|
)
|
|
template = (await db.execute(template_stmt)).scalar_one()
|
|
template.version = (template.version or 1) + 1
|
|
|
|
await db.commit()
|
|
|
|
return {"status": "success", "message": "검사 항목이 삭제되었습니다."}
|