diff --git a/app/api/v1/location.py b/app/api/v1/location.py index 8405b02..0e5b572 100644 --- a/app/api/v1/location.py +++ b/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( diff --git a/app/main.py b/app/main.py index f50b871..7e48053 100644 --- a/app/main.py +++ b/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, }, ) diff --git a/app/schemas/common.py b/app/schemas/common.py index af20c84..2644af0 100644 --- a/app/schemas/common.py +++ b/app/schemas/common.py @@ -4,4 +4,3 @@ from pydantic import BaseModel class ErrorResponse(BaseModel): error_code: str message: str - detail: dict | None = None diff --git a/app/services/location_service.py b/app/services/location_service.py index 973748f..cc69849 100644 --- a/app/services/location_service.py +++ b/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( - status_code=409, - detail={ - "error_code": "DUPLICATE_LOCATION", - "message": "该总排号已存在货位记录", - "detail": { - "existing_location": existing.location_code, - "created_at": existing.created_at.isoformat(), - }, - }, + raise DuplicateLocationError( + location_code=existing.location_code, + registered_at=existing.created_at.isoformat(), ) record = FinishedGoodsLocation( diff --git a/tests/test_location_api.py b/tests/test_location_api.py index 59d1e61..0fb7732 100644 --- a/tests/test_location_api.py +++ b/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 "existing_location" in data["detail"]["detail"] + 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): - """无效总排号 — 应返回 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):