fix: unify error response format with error_code, message and domain fields

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

app/api/v1/location.py

@@ -2,13 +2,27 @@ from fastapi import APIRouter, Depends
 from sqlalchemy.orm import Session
 
 from app.core.database import get_db
+from app.schemas.common import ErrorResponse
 from app.schemas.location import LocationRequest, LocationResponse
 from app.services.location_service import register_location
 
 router = APIRouter()
 
 
-@router.post("/location", response_model=LocationResponse)
+@router.post(
+    "/location",
+    response_model=LocationResponse,
+    responses={
+        400: {
+            "description": "总排号或货位号格式不合法",
+            "model": ErrorResponse,
+        },
+        409: {
+            "description": "该总排号已存在货位记录",
+            "model": ErrorResponse,
+        },
+    },
+)
 def create_location(req: LocationRequest, db: Session = Depends(get_db)):
     record = register_location(db, req)
     return LocationResponse(

app/main.py

@@ -1,26 +1,55 @@
 from fastapi import FastAPI, Request
+from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse
 
 from app.api.v1.location import router as location_router
+from app.services.location_service import DuplicateLocationError
 
 app = FastAPI(title="CargoTrace API", version="0.1.0")
 
 app.include_router(location_router, prefix="/CargoTrace")
 
 
+@app.exception_handler(DuplicateLocationError)
+async def duplicate_location_handler(request: Request, exc: DuplicateLocationError):
+    return JSONResponse(
+        status_code=409,
+        content={
+            "error_code": "DUPLICATE_LOCATION",
+            "message": "该总排号已存在货位记录",
+            "location_code": exc.location_code,
+            "registered_at": exc.registered_at,
+        },
+    )
+
+
+@app.exception_handler(RequestValidationError)
+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:
+            error_code = msg.replace("Value error, ", "")
+            return JSONResponse(
+                status_code=400,
+                content={
+                    "error_code": error_code,
+                    "message": error_code,
+                },
+            )
+    return JSONResponse(
+        status_code=422,
+        content={"detail": exc.errors()},
+    )
+
+
 @app.exception_handler(ValueError)
 async def value_error_handler(request: Request, exc: ValueError):
     error_code = str(exc)
-    status_map = {
-        "INVALID_ZONGPAI": 400,
-        "INVALID_LOCATION": 400,
-    }
-    status = status_map.get(error_code, 400)
     return JSONResponse(
-        status_code=status,
+        status_code=400,
         content={
             "error_code": error_code,
-            "message": str(exc),
+            "message": error_code,
         },
     )
 

app/schemas/common.py

@@ -4,4 +4,3 @@ from pydantic import BaseModel
 class ErrorResponse(BaseModel):
     error_code: str
     message: str
-    detail: dict | None = None

app/services/location_service.py

@@ -1,10 +1,17 @@
-from fastapi import HTTPException
 from sqlalchemy.orm import Session
 
 from app.models.finished_goods import FinishedGoodsLocation
 from app.schemas.location import LocationRequest
 
 
+class DuplicateLocationError(Exception):
+    """总排号已存在货位记录。"""
+
+    def __init__(self, location_code: str, registered_at: str):
+        self.location_code = location_code
+        self.registered_at = registered_at
+
+
 def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocation:
     existing = (
         db.query(FinishedGoodsLocation)
@@ -12,16 +19,9 @@ def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocatio
         .first()
     )
     if existing:
-        raise HTTPException(
+        raise DuplicateLocationError(
-            status_code=409,
+            location_code=existing.location_code,
-            detail={
+            registered_at=existing.created_at.isoformat(),
-                "error_code": "DUPLICATE_LOCATION",
-                "message": "该总排号已存在货位记录",
-                "detail": {
-                    "existing_location": existing.location_code,
-                    "created_at": existing.created_at.isoformat(),
-                },
-            },
         )
 
     record = FinishedGoodsLocation(

tests/test_location_api.py

@@ -14,7 +14,7 @@ def test_register_location_success(client: TestClient):
 
 
 def test_register_location_duplicate(client: TestClient):
-    """重复上架 — 应返回 409"""
+    """重复上架 — 应返回 409，包含统一错误格式"""
     client.post(
         "/CargoTrace/location",
         json={"zongpai_no": "26C999", "location_code": "A01-02-03"},
@@ -25,26 +25,29 @@ def test_register_location_duplicate(client: TestClient):
     )
     assert resp.status_code == 409
     data = resp.json()
-    assert data["detail"]["error_code"] == "DUPLICATE_LOCATION"
+    assert data["error_code"] == "DUPLICATE_LOCATION"
-    assert "existing_location" in data["detail"]["detail"]
+    assert data["location_code"] == "A01-02-03"
+    assert "registered_at" in data
 
 
 def test_register_location_invalid_zongpai(client: TestClient):
-    """无效总排号 — 应返回 422"""
+    """无效总排号 — 应返回 400"""
     resp = client.post(
         "/CargoTrace/location",
         json={"zongpai_no": "INVALID", "location_code": "A01-02-03"},
     )
-    assert resp.status_code == 422
+    assert resp.status_code == 400
+    assert resp.json()["error_code"] == "INVALID_ZONGPAI"
 
 
 def test_register_location_invalid_location(client: TestClient):
-    """无效货位号 — 应返回 422"""
+    """无效货位号 — 应返回 400"""
     resp = client.post(
         "/CargoTrace/location",
         json={"zongpai_no": "26T999", "location_code": "bad-location"},
     )
-    assert resp.status_code == 422
+    assert resp.status_code == 400
+    assert resp.json()["error_code"] == "INVALID_LOCATION"
 
 
 def test_register_location_trans_location(client: TestClient):