267 lines
7.5 KiB
Markdown
267 lines
7.5 KiB
Markdown
# 修复 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 成功（不变）
|
||
|
||
```json
|
||
{
|
||
"zongpai_no": "26B1",
|
||
"location_code": "A01-02-03",
|
||
"created_at": "2026-05-11T10:30:00"
|
||
}
|
||
```
|
||
|
||
### 409 重复上架（修改后）
|
||
|
||
```json
|
||
{
|
||
"error_code": "DUPLICATE_LOCATION",
|
||
"message": "该总排号已存在货位记录",
|
||
"location_code": "A01-02-03",
|
||
"registered_at": "2026-05-11T10:30:00"
|
||
}
|
||
```
|
||
|
||
### 400 校验失败（修改后）
|
||
|
||
```json
|
||
{
|
||
"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**
|
||
|
||
```python
|
||
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 匹配新格式**
|
||
|
||
```python
|
||
from pydantic import BaseModel
|
||
|
||
|
||
class ErrorResponse(BaseModel):
|
||
error_code: str
|
||
message: str
|
||
```
|
||
|
||
**Step 3: 修改 `app/main.py` — 添加 DuplicateLocationError handler，更新 ValueError handler**
|
||
|
||
```python
|
||
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 文档**
|
||
|
||
```python
|
||
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` — 更新断言**
|
||
|
||
```python
|
||
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**
|
||
|
||
```bash
|
||
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"
|
||
```
|