feat: add accessory API routes and exception handlers

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app/api/v1/accessory.py

@@ -0,0 +1,116 @@
+from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
+
+from app.core.database import get_db
+from app.schemas.accessory import (
+    AccessoryCreateRequest,
+    AccessoryListResponse,
+    AccessoryResponse,
+    AccessoryTypeCreateRequest,
+    AccessoryTypeListResponse,
+    AccessoryTypeUpdateRequest,
+    AccessoryUpdateRequest,
+    WorkOrderQueryResponse,
+)
+from app.services.accessory_service import (
+    create_accessory,
+    create_accessory_type,
+    delete_accessory,
+    delete_accessory_type,
+    list_accessories,
+    list_accessory_types,
+    update_accessory,
+    update_accessory_type,
+    query_work_orders,
+)
+
+router = APIRouter(tags=["accessory"])
+
+
+@router.get(
+    "/accessory/work-orders",
+    response_model=WorkOrderQueryResponse,
+)
+def work_orders(
+    paichan_no: str = Query(..., description="排产号"),
+    db: Session = Depends(get_db),
+):
+    """查询工单信息：根据排产号查询工令号及对应总排号。"""
+    return query_work_orders(db, paichan_no)
+
+
+@router.get(
+    "/accessory",
+    response_model=AccessoryListResponse,
+)
+def accessory_list(
+    paichan_no: str = Query(..., description="排产号"),
+    db: Session = Depends(get_db),
+):
+    """查询附件列表：根据排产号查询所有附件记录。"""
+    return list_accessories(db, paichan_no)
+
+
+@router.post(
+    "/accessory",
+    response_model=AccessoryResponse,
+)
+def accessory_create(req: AccessoryCreateRequest, db: Session = Depends(get_db)):
+    """创建附件记录。"""
+    record = create_accessory(db, req)
+    return {
+        "id": record.id,
+        "paichan_no": record.paichan_no,
+        "zongpai_no": record.zongpai_no,
+        "accessory_type": record.accessory_type,
+        "quantity": record.quantity,
+        "location_code": record.location_code,
+        "is_boxed": False,
+        "created_at": record.created_at.isoformat(),
+    }
+
+
+@router.patch(
+    "/accessory/{id}",
+    response_model=AccessoryResponse,
+)
+def accessory_update(id: int, req: AccessoryUpdateRequest, db: Session = Depends(get_db)):
+    """更新附件记录。"""
+    return update_accessory(db, id, req)
+
+
+@router.delete("/accessory/{id}")
+def accessory_delete(id: int, db: Session = Depends(get_db)):
+    """删除附件记录。"""
+    return delete_accessory(db, id)
+
+
+@router.get(
+    "/accessory-type",
+    response_model=AccessoryTypeListResponse,
+)
+def accessory_type_list(db: Session = Depends(get_db)):
+    """查询所有附件类型。"""
+    return list_accessory_types(db)
+
+
+@router.post("/accessory-type")
+def accessory_type_create(
+    req: AccessoryTypeCreateRequest, db: Session = Depends(get_db)
+):
+    """创建附件类型。"""
+    return create_accessory_type(db, req)
+
+
+@router.patch("/accessory-type/{id}")
+def accessory_type_update(
+    id: int, req: AccessoryTypeUpdateRequest, db: Session = Depends(get_db)
+):
+    """更新附件类型。"""
+    return update_accessory_type(db, id, req)
+
+
+@router.delete("/accessory-type/{id}")
+def accessory_type_delete(id: int, db: Session = Depends(get_db)):
+    """删除附件类型。"""
+    return delete_accessory_type(db, id)

app/main.py

@@ -18,6 +18,7 @@ except Exception as e:
 
 from app.api.v1.location import router as location_router
 from app.api.v1.box import router as box_router
+from app.api.v1.accessory import router as accessory_router
 from app.services.location_service import (
     AlreadyOffShelfError,
     DuplicateLocationError,
@@ -31,11 +32,22 @@ from app.services.box_service import (
     InvalidZongpaiError,
     ZongpaiNotFoundError,
 )
+from app.services.accessory_service import (
+    AccessoryAlreadyBoxedError,
+    AccessoryAlreadyOffShelfError,
+    AccessoryDuplicateLocationError,
+    AccessoryNotFoundError,
+    AccessoryTypeNotFoundError,
+    DuplicateAccessoryTypeNameError,
+    InvalidPaichanError,
+    PaichanNotFoundError as AccessoryPaichanNotFoundError,
+)
 
 app = FastAPI(title="CargoTrace API", version="0.1.0")
 
 app.include_router(location_router, prefix="/CargoTrace")
 app.include_router(box_router, prefix="/CargoTrace")
+app.include_router(accessory_router, prefix="/CargoTrace")
 
 
 @app.exception_handler(DuplicateLocationError)
@@ -68,7 +80,7 @@ async def already_off_shelf_handler(request: Request, exc: AlreadyOffShelfError)
 async def validation_error_handler(request: Request, exc: RequestValidationError):
     for err in exc.errors():
         msg = err.get("msg", "")
-        if "INVALID_ZONGPAI" in msg or "INVALID_LOCATION" in msg or "INVALID_BOX_NO" in msg or "INVALID_QUANTITY" in msg:
+        if "INVALID_ZONGPAI" in msg or "INVALID_LOCATION" in msg or "INVALID_BOX_NO" in msg or "INVALID_QUANTITY" in msg or "INVALID_PAICHAN" in msg or "INVALID_TYPE_NAME" in msg:
             error_code = msg.replace("Value error, ", "")
             return JSONResponse(
                 status_code=400,
@@ -174,6 +186,110 @@ async def invalid_box_item_handler(request: Request, exc: InvalidBoxItemError):
     )
 
 
+@app.exception_handler(InvalidPaichanError)
+async def invalid_paichan_handler(request: Request, exc: InvalidPaichanError):
+    return JSONResponse(
+        status_code=400,
+        content={
+            "error_code": "INVALID_PAICHAN",
+            "message": "排产号格式不合法",
+        },
+    )
+
+
+@app.exception_handler(AccessoryPaichanNotFoundError)
+async def accessory_paichan_not_found_handler(
+    request: Request, exc: AccessoryPaichanNotFoundError
+):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "PAICHA_NOT_FOUND",
+            "message": "未找到该排产号信息",
+        },
+    )
+
+
+@app.exception_handler(AccessoryNotFoundError)
+async def accessory_not_found_handler(request: Request, exc: AccessoryNotFoundError):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "ACCESSORY_NOT_FOUND",
+            "message": "附件记录不存在",
+        },
+    )
+
+
+@app.exception_handler(AccessoryAlreadyBoxedError)
+async def accessory_already_boxed_handler(
+    request: Request, exc: AccessoryAlreadyBoxedError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "ALREADY_BOXED",
+            "message": "该附件已装箱不可操作",
+        },
+    )
+
+
+@app.exception_handler(AccessoryDuplicateLocationError)
+async def accessory_duplicate_location_handler(
+    request: Request, exc: AccessoryDuplicateLocationError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_LOCATION",
+            "message": "该附件已存在货位记录",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
+@app.exception_handler(AccessoryAlreadyOffShelfError)
+async def accessory_already_off_shelf_handler(
+    request: Request, exc: AccessoryAlreadyOffShelfError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "ALREADY_OFF_SHELF",
+            "message": "该附件已下架不可重新上架",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
+@app.exception_handler(AccessoryTypeNotFoundError)
+async def accessory_type_not_found_handler(
+    request: Request, exc: AccessoryTypeNotFoundError
+):
+    return JSONResponse(
+        status_code=404,
+        content={
+            "error_code": "ACCESSORY_TYPE_NOT_FOUND",
+            "message": "附件类型不存在",
+        },
+    )
+
+
+@app.exception_handler(DuplicateAccessoryTypeNameError)
+async def duplicate_accessory_type_name_handler(
+    request: Request, exc: DuplicateAccessoryTypeNameError
+):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_TYPE_NAME",
+            "message": "附件类型名称已存在",
+        },
+    )
+
+
 @app.get("/")
 async def root():
     return {"message": "Welcome to CargoTrace API"}