fastAPI/docs/plans/2026-05-11-fix-409-response.md

修复 409 响应格式 — 第一阶段：FastAPI

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: 将 409 重复上架的响应改为统一错误格式 {"error_code", "message", "location_code", "registered_at"}。

Architecture: 用自定义异常 DuplicateLocationError 替代 HTTPException，通过自定义 exception handler 返回扁平的统一错误结构。同时更新 200 成功响应和 Swagger 文档。

Tech Stack: FastAPI, SQLAlchemy, pytest

---

目标响应格式

200 成功（不变）

{
  "zongpai_no": "26B1",
  "location_code": "A01-02-03",
  "created_at": "2026-05-11T10:30:00"
}

409 重复上架（修改后）

{
  "error_code": "DUPLICATE_LOCATION",
  "message": "该总排号已存在货位记录",
  "location_code": "A01-02-03",
  "registered_at": "2026-05-11T10:30:00"
}

400 校验失败（修改后）

{
  "error_code": "INVALID_ZONGPAI",
  "message": "INVALID_ZONGPAI"
}

---

Task 1: 自定义异常 + handler + 测试更新

Files:

• Modify: apps/fastAPI/app/services/location_service.py
• Modify: apps/fastAPI/app/main.py
• Modify: apps/fastAPI/app/api/v1/location.py
• Modify: apps/fastAPI/app/schemas/common.py
• Modify: apps/fastAPI/tests/test_location_api.py

Step 1: 修改 app/services/location_service.py — 用自定义异常替换 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)
        .filter(FinishedGoodsLocation.zongpai_no == req.zongpai_no)
        .first()
    )
    if existing:
        raise DuplicateLocationError(
            location_code=existing.location_code,
            registered_at=existing.created_at.isoformat(),
        )

    record = FinishedGoodsLocation(
        zongpai_no=req.zongpai_no,
        location_code=req.location_code,
    )
    db.add(record)
    db.commit()
    db.refresh(record)
    return record

Step 2: 修改 app/schemas/common.py — 更新 ErrorResponse 匹配新格式

from pydantic import BaseModel


class ErrorResponse(BaseModel):
    error_code: str
    message: str

Step 3: 修改 app/main.py — 添加 DuplicateLocationError handler，更新 ValueError handler

from fastapi import FastAPI, Request
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(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    error_code = str(exc)
    return JSONResponse(
        status_code=400,
        content={
            "error_code": error_code,
            "message": error_code,
        },
    )


@app.get("/")
async def root():
    return {"message": "Welcome to CargoTrace API"}

Step 4: 修改 app/api/v1/location.py — 更新 Swagger 409 文档

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,
    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(
        zongpai_no=record.zongpai_no,
        location_code=record.location_code,
        created_at=record.created_at.isoformat(),
    )

Step 5: 修改 tests/test_location_api.py — 更新断言

from fastapi.testclient import TestClient


def test_register_location_success(client: TestClient):
    """正常上架 — 应返回 200"""
    resp = client.post(
        "/CargoTrace/location",
        json={"zongpai_no": "26B999", "location_code": "A01-02-03"},
    )
    assert resp.status_code == 200
    data = resp.json()
    assert data["zongpai_no"] == "26B999"
    assert data["location_code"] == "A01-02-03"


def test_register_location_duplicate(client: TestClient):
    """重复上架 — 应返回 409，包含统一错误格式"""
    client.post(
        "/CargoTrace/location",
        json={"zongpai_no": "26C999", "location_code": "A01-02-03"},
    )
    resp = client.post(
        "/CargoTrace/location",
        json={"zongpai_no": "26C999", "location_code": "A02-01-01"},
    )
    assert resp.status_code == 409
    data = resp.json()
    assert data["error_code"] == "DUPLICATE_LOCATION"
    assert data["location_code"] == "A01-02-03"
    assert "registered_at" in data


def test_register_location_invalid_zongpai(client: TestClient):
    """无效总排号 — 应返回 400"""
    resp = client.post(
        "/CargoTrace/location",
        json={"zongpai_no": "INVALID", "location_code": "A01-02-03"},
    )
    assert resp.status_code == 400
    assert resp.json()["error_code"] == "INVALID_ZONGPAI"


def test_register_location_invalid_location(client: TestClient):
    """无效货位号 — 应返回 400"""
    resp = client.post(
        "/CargoTrace/location",
        json={"zongpai_no": "26T999", "location_code": "bad-location"},
    )
    assert resp.status_code == 400
    assert resp.json()["error_code"] == "INVALID_LOCATION"


def test_register_location_trans_location(client: TestClient):
    """转运特殊货位 — 应返回 200"""
    resp = client.post(
        "/CargoTrace/location",
        json={"zongpai_no": "26BW0999", "location_code": "TRANS-01"},
    )
    assert resp.status_code == 200
    assert resp.json()["location_code"] == "TRANS-01"

注意变更：

• INVALID_ZONGPAI / INVALID_LOCATION 现在返回 400（不再是 422），由自定义 ValueError handler 处理
• 409 响应断言改为验证 error_code、location_code、registered_at 三个字段

Step 6: 运行测试

Run: cd D:/projects/CargoTrace/apps/fastAPI && .venv/Scripts/python.exe -m pytest tests/ -v Expected: 5 tests PASSED

Step 7: Commit

git add app/services/location_service.py app/main.py app/api/v1/location.py app/schemas/common.py tests/test_location_api.py
git commit -m "fix: unify error response format with error_code, message and domain fields"