feat: add boxing module API — GET /box/info and POST /box
- Add box schemas (BoxInfoResponse, BoxSaveRequest, BoxSaveResponse) - Add box service with ERP view query and save logic (supports many2one append) - Add box router with GET /box/info and POST /box endpoints - Register exception handlers for InvalidZongpaiError, ZongpaiNotFoundError, DuplicateBoxItemError - Update RequestValidationError handler to cover INVALID_BOX_NO and INVALID_QUANTITY - Add 7 integration tests for box API Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Misaka_Company
23
app/api/v1/box.py
Normal file
23
app/api/v1/box.py
Normal file
@@ -0,0 +1,23 @@
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.database import get_db
|
||||
from app.schemas.box import BoxInfoResponse, BoxSaveRequest, BoxSaveResponse
|
||||
from app.services.box_service import get_box_info, save_box_record
|
||||
|
||||
router = APIRouter(tags=["boxing"])
|
||||
|
||||
|
||||
@router.get("/box/info", response_model=BoxInfoResponse)
|
||||
def box_info(
|
||||
zongpai_no: str = Query(..., description="总排号"),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""查询装箱信息：根据总排号查询排产号、数量和已有箱号明细。"""
|
||||
return get_box_info(db, zongpai_no)
|
||||
|
||||
|
||||
@router.post("/box", response_model=BoxSaveResponse)
|
||||
def create_box(req: BoxSaveRequest, db: Session = Depends(get_db)):
|
||||
"""保存装箱记录：将总排号记录到指定箱号中。"""
|
||||
return save_box_record(db, req)
|
||||
40
app/main.py
40
app/main.py
@@ -3,11 +3,14 @@ from fastapi.exceptions import RequestValidationError
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
from app.api.v1.location import router as location_router
|
||||
from app.api.v1.box import router as box_router
|
||||
from app.services.location_service import DuplicateLocationError
|
||||
from app.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError, InvalidZongpaiError
|
||||
|
||||
app = FastAPI(title="CargoTrace API", version="0.1.0")
|
||||
|
||||
app.include_router(location_router, prefix="/CargoTrace")
|
||||
app.include_router(box_router, prefix="/CargoTrace")
|
||||
|
||||
|
||||
@app.exception_handler(DuplicateLocationError)
|
||||
@@ -27,7 +30,7 @@ async def duplicate_location_handler(request: Request, exc: DuplicateLocationErr
|
||||
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:
|
||||
if "INVALID_ZONGPAI" in msg or "INVALID_LOCATION" in msg or "INVALID_BOX_NO" in msg or "INVALID_QUANTITY" in msg:
|
||||
error_code = msg.replace("Value error, ", "")
|
||||
return JSONResponse(
|
||||
status_code=400,
|
||||
@@ -54,6 +57,41 @@ async def value_error_handler(request: Request, exc: ValueError):
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(InvalidZongpaiError)
|
||||
async def invalid_zongpai_handler(request: Request, exc: InvalidZongpaiError):
|
||||
return JSONResponse(
|
||||
status_code=400,
|
||||
content={
|
||||
"error_code": "INVALID_ZONGPAI",
|
||||
"message": "总排号格式不合法",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(ZongpaiNotFoundError)
|
||||
async def zongpai_not_found_handler(request: Request, exc: ZongpaiNotFoundError):
|
||||
return JSONResponse(
|
||||
status_code=404,
|
||||
content={
|
||||
"error_code": "ZONGPAI_NOT_FOUND",
|
||||
"message": "未找到该总排号对应的排产号信息",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(DuplicateBoxItemError)
|
||||
async def duplicate_box_handler(request: Request, exc: DuplicateBoxItemError):
|
||||
return JSONResponse(
|
||||
status_code=409,
|
||||
content={
|
||||
"error_code": "DUPLICATE_BOX_NO",
|
||||
"message": f"排产号 {exc.paichan_no} 下箱号 {exc.box_no} 已存在",
|
||||
"paichan_no": exc.paichan_no,
|
||||
"box_no": exc.box_no,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@app.get("/")
|
||||
async def root():
|
||||
return {"message": "Welcome to CargoTrace API"}
|
||||
|
||||
65
app/schemas/box.py
Normal file
65
app/schemas/box.py
Normal file
@@ -0,0 +1,65 @@
|
||||
import re
|
||||
|
||||
from pydantic import BaseModel, field_validator
|
||||
|
||||
ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
|
||||
|
||||
|
||||
class BoxItemDetail(BaseModel):
|
||||
"""箱号下某个总排号的明细"""
|
||||
zongpai_no: str
|
||||
quantity: int
|
||||
|
||||
|
||||
class BoxDetail(BaseModel):
|
||||
"""一个箱号的完整信息"""
|
||||
box_no: int
|
||||
items: list[BoxItemDetail]
|
||||
|
||||
|
||||
class BoxInfoResponse(BaseModel):
|
||||
"""GET /box/info 响应"""
|
||||
zongpai_no: str
|
||||
paichan_no: str
|
||||
quantity: int
|
||||
existing_boxes: list[BoxDetail]
|
||||
max_box_no: int
|
||||
suggested_box_no: int
|
||||
|
||||
|
||||
class BoxSaveRequest(BaseModel):
|
||||
"""POST /box 请求"""
|
||||
zongpai_no: str
|
||||
box_no: int
|
||||
quantity: int
|
||||
|
||||
@field_validator("zongpai_no")
|
||||
@classmethod
|
||||
def validate_zongpai_no(cls, v: str) -> str:
|
||||
v = v.strip().upper()
|
||||
if not ZONGPAI_PATTERN.match(v):
|
||||
raise ValueError("INVALID_ZONGPAI")
|
||||
return v
|
||||
|
||||
@field_validator("box_no")
|
||||
@classmethod
|
||||
def validate_box_no(cls, v: int) -> int:
|
||||
if v <= 0:
|
||||
raise ValueError("INVALID_BOX_NO")
|
||||
return v
|
||||
|
||||
@field_validator("quantity")
|
||||
@classmethod
|
||||
def validate_quantity(cls, v: int) -> int:
|
||||
if v <= 0:
|
||||
raise ValueError("INVALID_QUANTITY")
|
||||
return v
|
||||
|
||||
|
||||
class BoxSaveResponse(BaseModel):
|
||||
"""POST /box 响应"""
|
||||
paichan_no: str
|
||||
box_no: int
|
||||
zongpai_no: str
|
||||
quantity: int
|
||||
created_at: str
|
||||
135
app/services/box_service.py
Normal file
135
app/services/box_service.py
Normal file
@@ -0,0 +1,135 @@
|
||||
import re
|
||||
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
|
||||
from app.schemas.box import BoxSaveRequest
|
||||
|
||||
ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
|
||||
|
||||
|
||||
class InvalidZongpaiError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class ZongpaiNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class DuplicateBoxItemError(Exception):
|
||||
def __init__(self, paichan_no: str, box_no: int):
|
||||
self.paichan_no = paichan_no
|
||||
self.box_no = box_no
|
||||
|
||||
|
||||
def query_erp_info(db: Session, zongpai_no: str) -> dict:
|
||||
"""从 ERP 视图查询总排号对应的排产号和数量。"""
|
||||
sql = text(
|
||||
"SELECT TOP 1 [总排号], [排产号], [数量] "
|
||||
"FROM [ERPAuto].[vw_productionContractData] "
|
||||
"WHERE [总排号] = :zongpai_no"
|
||||
)
|
||||
row = db.execute(sql, {"zongpai_no": zongpai_no}).fetchone()
|
||||
if not row:
|
||||
raise ZongpaiNotFoundError()
|
||||
return {
|
||||
"zongpai_no": row[0],
|
||||
"paichan_no": row[1],
|
||||
"quantity": int(row[2]),
|
||||
}
|
||||
|
||||
|
||||
def get_box_info(db: Session, zongpai_no: str) -> dict:
|
||||
"""获取装箱信息：排产号、数量、已有箱号明细。"""
|
||||
zongpai_no = zongpai_no.strip().upper()
|
||||
if not ZONGPAI_PATTERN.match(zongpai_no):
|
||||
raise InvalidZongpaiError()
|
||||
erp = query_erp_info(db, zongpai_no)
|
||||
paichan_no = erp["paichan_no"]
|
||||
quantity = erp["quantity"]
|
||||
|
||||
boxes = (
|
||||
db.query(FinishedGoodsBox)
|
||||
.filter(FinishedGoodsBox.paichan_no == paichan_no)
|
||||
.order_by(FinishedGoodsBox.box_no)
|
||||
.all()
|
||||
)
|
||||
|
||||
existing_boxes = []
|
||||
max_box_no = 0
|
||||
for box in boxes:
|
||||
items = (
|
||||
db.query(FinishedGoodsBoxItem)
|
||||
.filter(FinishedGoodsBoxItem.box_id == box.id)
|
||||
.all()
|
||||
)
|
||||
existing_boxes.append({
|
||||
"box_no": box.box_no,
|
||||
"items": [
|
||||
{"zongpai_no": item.zongpai_no, "quantity": int(item.quantity or 0)}
|
||||
for item in items
|
||||
],
|
||||
})
|
||||
if box.box_no > max_box_no:
|
||||
max_box_no = box.box_no
|
||||
|
||||
return {
|
||||
"zongpai_no": zongpai_no,
|
||||
"paichan_no": paichan_no,
|
||||
"quantity": quantity,
|
||||
"existing_boxes": existing_boxes,
|
||||
"max_box_no": max_box_no,
|
||||
"suggested_box_no": max_box_no + 1,
|
||||
}
|
||||
|
||||
|
||||
def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
|
||||
"""保存装箱记录。箱已存在时仅追加明细（多码一箱场景）。"""
|
||||
erp = query_erp_info(db, req.zongpai_no)
|
||||
paichan_no = erp["paichan_no"]
|
||||
|
||||
box = (
|
||||
db.query(FinishedGoodsBox)
|
||||
.filter(
|
||||
FinishedGoodsBox.paichan_no == paichan_no,
|
||||
FinishedGoodsBox.box_no == req.box_no,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if box:
|
||||
existing_item = (
|
||||
db.query(FinishedGoodsBoxItem)
|
||||
.filter(
|
||||
FinishedGoodsBoxItem.box_id == box.id,
|
||||
FinishedGoodsBoxItem.zongpai_no == req.zongpai_no,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if existing_item:
|
||||
raise DuplicateBoxItemError(paichan_no, req.box_no)
|
||||
else:
|
||||
box = FinishedGoodsBox(
|
||||
paichan_no=paichan_no,
|
||||
box_no=req.box_no,
|
||||
)
|
||||
db.add(box)
|
||||
db.flush()
|
||||
|
||||
item = FinishedGoodsBoxItem(
|
||||
box_id=box.id,
|
||||
zongpai_no=req.zongpai_no,
|
||||
quantity=req.quantity,
|
||||
)
|
||||
db.add(item)
|
||||
db.commit()
|
||||
db.refresh(item)
|
||||
|
||||
return {
|
||||
"paichan_no": paichan_no,
|
||||
"box_no": req.box_no,
|
||||
"zongpai_no": req.zongpai_no,
|
||||
"quantity": req.quantity,
|
||||
"created_at": item.created_at.isoformat(),
|
||||
}
|
||||
507
docs/plans/2026-05-11-boxing-module.md
Normal file
507
docs/plans/2026-05-11-boxing-module.md
Normal file
@@ -0,0 +1,507 @@
|
||||
# 装箱编号模块 (Boxing) Implementation Plan — FastAPI
|
||||
|
||||
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
|
||||
|
||||
**Goal:** 实现 PRD v1.3 中定义的装箱编号模块后端接口：`GET /CargoTrace/box/info` 和 `POST /CargoTrace/box`，支持总排号查询排产号信息、已有箱号查询、以及装箱记录保存。
|
||||
|
||||
**Architecture:** 在现有三层架构（API → Service → Model）基础上扩展。ERP 视图数据通过 SQLAlchemy raw SQL 查询（跨 schema 访问 `[ERPAuto].[vw_productionContractData]`），CargoTrace 业务表通过 ORM 操作。
|
||||
|
||||
**Tech Stack:** Python 3.12+, FastAPI, SQLAlchemy 2.x (sync), pyodbc, pydantic
|
||||
|
||||
---
|
||||
|
||||
## 数据流
|
||||
|
||||
```
|
||||
Flutter App
|
||||
│
|
||||
├── GET /CargoTrace/box/info?zongpai_no=xxx
|
||||
│ └── 查询 ERP 视图 → 排产号 + 数量
|
||||
│ └── 查询 CargoTrace 表 → 已有箱号 + 明细
|
||||
│ └── 返回: paichan_no, quantity, existing_boxes, max_box_no, suggested_box_no
|
||||
│
|
||||
└── POST /CargoTrace/box
|
||||
└── 查询 ERP 视图 → 排产号
|
||||
└── 查找或创建 CargoTrace.finished_goods_box
|
||||
└── 创建 CargoTrace.finished_goods_box_item
|
||||
└── 返回: paichan_no, box_no, zongpai_no, quantity, created_at
|
||||
```
|
||||
|
||||
## 关键设计决策
|
||||
|
||||
**多码一箱的 POST 逻辑：**
|
||||
- 若 (paichan_no, box_no) 不存在 → 创建 box 记录 + 创建 box_item
|
||||
- 若 (paichan_no, box_no) 已存在且该 zongpai_no 不在此箱中 → 仅创建 box_item（多码一箱场景）
|
||||
- 若 (paichan_no, box_no) 已存在且该 zongpai_no 已在此箱中 → 返回 409 DUPLICATE_BOX_NO
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Schema 层 — 装箱请求/响应模型
|
||||
|
||||
**Files:**
|
||||
- Create: `app/schemas/box.py`
|
||||
|
||||
**Step 1: 创建 box.py**
|
||||
|
||||
```python
|
||||
import re
|
||||
|
||||
from pydantic import BaseModel, field_validator
|
||||
|
||||
ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
|
||||
|
||||
|
||||
class BoxItemDetail(BaseModel):
|
||||
"""箱号下某个总排号的明细"""
|
||||
zongpai_no: str
|
||||
quantity: int
|
||||
|
||||
|
||||
class BoxDetail(BaseModel):
|
||||
"""一个箱号的完整信息"""
|
||||
box_no: int
|
||||
items: list[BoxItemDetail]
|
||||
|
||||
|
||||
class BoxInfoResponse(BaseModel):
|
||||
"""GET /box/info 响应"""
|
||||
zongpai_no: str
|
||||
paichan_no: str
|
||||
quantity: int
|
||||
existing_boxes: list[BoxDetail]
|
||||
max_box_no: int
|
||||
suggested_box_no: int
|
||||
|
||||
|
||||
class BoxSaveRequest(BaseModel):
|
||||
"""POST /box 请求"""
|
||||
zongpai_no: str
|
||||
box_no: int
|
||||
quantity: int
|
||||
|
||||
@field_validator("zongpai_no")
|
||||
@classmethod
|
||||
def validate_zongpai_no(cls, v: str) -> str:
|
||||
v = v.strip().upper()
|
||||
if not ZONGPAI_PATTERN.match(v):
|
||||
raise ValueError("INVALID_ZONGPAI")
|
||||
return v
|
||||
|
||||
@field_validator("box_no")
|
||||
@classmethod
|
||||
def validate_box_no(cls, v: int) -> int:
|
||||
if v <= 0:
|
||||
raise ValueError("INVALID_BOX_NO")
|
||||
return v
|
||||
|
||||
@field_validator("quantity")
|
||||
@classmethod
|
||||
def validate_quantity(cls, v: int) -> int:
|
||||
if v <= 0:
|
||||
raise ValueError("INVALID_QUANTITY")
|
||||
return v
|
||||
|
||||
|
||||
class BoxSaveResponse(BaseModel):
|
||||
"""POST /box 响应"""
|
||||
paichan_no: str
|
||||
box_no: int
|
||||
zongpai_no: str
|
||||
quantity: int
|
||||
created_at: str
|
||||
```
|
||||
|
||||
**Step 2: 验证导入**
|
||||
|
||||
Run: `cd apps/fastAPI && .venv/Scripts/python -c "from app.schemas.box import BoxInfoResponse, BoxSaveRequest; print('Schemas OK')"`
|
||||
|
||||
**Step 3: Commit**
|
||||
|
||||
```bash
|
||||
git add app/schemas/box.py
|
||||
git commit -m "feat: add box schemas for boxing module API"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: Service 层 — 装箱业务逻辑
|
||||
|
||||
**Files:**
|
||||
- Create: `app/services/box_service.py`
|
||||
|
||||
**Step 1: 创建 box_service.py**
|
||||
|
||||
```python
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
|
||||
from app.schemas.box import BoxSaveRequest
|
||||
|
||||
|
||||
class ZongpaiNotFoundError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
class DuplicateBoxItemError(Exception):
|
||||
def __init__(self, paichan_no: str, box_no: int):
|
||||
self.paichan_no = paichan_no
|
||||
self.box_no = box_no
|
||||
|
||||
|
||||
def query_erp_info(db: Session, zongpai_no: str) -> dict:
|
||||
"""从 ERP 视图查询总排号对应的排产号和数量。"""
|
||||
sql = text(
|
||||
"SELECT TOP 1 [总排号], [排产号], [数量] "
|
||||
"FROM [ERPAuto].[vw_productionContractData] "
|
||||
"WHERE [总排号] = :zongpai_no"
|
||||
)
|
||||
row = db.execute(sql, {"zongpai_no": zongpai_no}).fetchone()
|
||||
if not row:
|
||||
raise ZongpaiNotFoundError()
|
||||
return {
|
||||
"zongpai_no": row[0],
|
||||
"paichan_no": row[1],
|
||||
"quantity": int(row[2]),
|
||||
}
|
||||
|
||||
|
||||
def get_box_info(db: Session, zongpai_no: str) -> dict:
|
||||
"""获取装箱信息：排产号、数量、已有箱号明细。"""
|
||||
erp = query_erp_info(db, zongpai_no)
|
||||
paichan_no = erp["paichan_no"]
|
||||
quantity = erp["quantity"]
|
||||
|
||||
# 查询该排产号下的所有箱（按 box_no 升序）
|
||||
boxes = (
|
||||
db.query(FinishedGoodsBox)
|
||||
.filter(FinishedGoodsBox.paichan_no == paichan_no)
|
||||
.order_by(FinishedGoodsBox.box_no)
|
||||
.all()
|
||||
)
|
||||
|
||||
existing_boxes = []
|
||||
max_box_no = 0
|
||||
for box in boxes:
|
||||
items = (
|
||||
db.query(FinishedGoodsBoxItem)
|
||||
.filter(FinishedGoodsBoxItem.box_id == box.id)
|
||||
.all()
|
||||
)
|
||||
existing_boxes.append({
|
||||
"box_no": box.box_no,
|
||||
"items": [
|
||||
{"zongpai_no": item.zongpai_no, "quantity": int(item.quantity or 0)}
|
||||
for item in items
|
||||
],
|
||||
})
|
||||
if box.box_no > max_box_no:
|
||||
max_box_no = box.box_no
|
||||
|
||||
return {
|
||||
"zongpai_no": zongpai_no,
|
||||
"paichan_no": paichan_no,
|
||||
"quantity": quantity,
|
||||
"existing_boxes": existing_boxes,
|
||||
"max_box_no": max_box_no,
|
||||
"suggested_box_no": max_box_no + 1,
|
||||
}
|
||||
|
||||
|
||||
def save_box_record(db: Session, req: BoxSaveRequest) -> dict:
|
||||
"""保存装箱记录。处理多码一箱场景：箱已存在时仅追加明细。"""
|
||||
# 1. 查询 ERP 获取排产号
|
||||
erp = query_erp_info(db, req.zongpai_no)
|
||||
paichan_no = erp["paichan_no"]
|
||||
|
||||
# 2. 查找已有箱记录
|
||||
box = (
|
||||
db.query(FinishedGoodsBox)
|
||||
.filter(
|
||||
FinishedGoodsBox.paichan_no == paichan_no,
|
||||
FinishedGoodsBox.box_no == req.box_no,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if box:
|
||||
# 箱已存在 → 检查该总排号是否已在此箱中
|
||||
existing_item = (
|
||||
db.query(FinishedGoodsBoxItem)
|
||||
.filter(
|
||||
FinishedGoodsBoxItem.box_id == box.id,
|
||||
FinishedGoodsBoxItem.zongpai_no == req.zongpai_no,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
if existing_item:
|
||||
raise DuplicateBoxItemError(paichan_no, req.box_no)
|
||||
else:
|
||||
# 创建新箱
|
||||
box = FinishedGoodsBox(
|
||||
paichan_no=paichan_no,
|
||||
box_no=req.box_no,
|
||||
)
|
||||
db.add(box)
|
||||
db.flush() # 获取 box.id
|
||||
|
||||
# 3. 创建箱明细
|
||||
item = FinishedGoodsBoxItem(
|
||||
box_id=box.id,
|
||||
zongpai_no=req.zongpai_no,
|
||||
quantity=req.quantity,
|
||||
)
|
||||
db.add(item)
|
||||
db.commit()
|
||||
db.refresh(item)
|
||||
|
||||
return {
|
||||
"paichan_no": paichan_no,
|
||||
"box_no": req.box_no,
|
||||
"zongpai_no": req.zongpai_no,
|
||||
"quantity": req.quantity,
|
||||
"created_at": item.created_at.isoformat(),
|
||||
}
|
||||
```
|
||||
|
||||
**Step 2: 验证导入**
|
||||
|
||||
Run: `cd apps/fastAPI && .venv/Scripts/python -c "from app.services.box_service import get_box_info, save_box_record; print('Service OK')"`
|
||||
|
||||
**Step 3: Commit**
|
||||
|
||||
```bash
|
||||
git add app/services/box_service.py
|
||||
git commit -m "feat: add box service with ERP query and save logic"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: API 路由层 — GET /box/info 和 POST /box
|
||||
|
||||
**Files:**
|
||||
- Create: `app/api/v1/box.py`
|
||||
- Modify: `app/main.py`
|
||||
|
||||
**Step 1: 创建 api/v1/box.py**
|
||||
|
||||
```python
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.database import get_db
|
||||
from app.schemas.box import BoxInfoResponse, BoxSaveRequest, BoxSaveResponse
|
||||
from app.services.box_service import get_box_info, save_box_record
|
||||
|
||||
router = APIRouter(tags=["boxing"])
|
||||
|
||||
|
||||
@router.get("/box/info", response_model=BoxInfoResponse)
|
||||
def box_info(
|
||||
zongpai_no: str = Query(..., description="总排号"),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""查询装箱信息：根据总排号查询排产号、数量和已有箱号明细。"""
|
||||
return get_box_info(db, zongpai_no)
|
||||
|
||||
|
||||
@router.post("/box", response_model=BoxSaveResponse)
|
||||
def create_box(req: BoxSaveRequest, db: Session = Depends(get_db)):
|
||||
"""保存装箱记录：将总排号记录到指定箱号中。"""
|
||||
return save_box_record(db, req)
|
||||
```
|
||||
|
||||
**Step 2: 更新 main.py — 注册路由 + 异常处理**
|
||||
|
||||
在现有 `main.py` 中添加：
|
||||
|
||||
```python
|
||||
# 新增导入（文件顶部）
|
||||
from app.api.v1.box import router as box_router
|
||||
from app.services.box_service import ZongpaiNotFoundError, DuplicateBoxItemError
|
||||
|
||||
# 注册路由（在 location_router 之后）
|
||||
app.include_router(box_router, prefix="/CargoTrace")
|
||||
|
||||
# 新增异常处理（在现有 handler 之后）
|
||||
@app.exception_handler(ZongpaiNotFoundError)
|
||||
async def zongpai_not_found_handler(request: Request, exc: ZongpaiNotFoundError):
|
||||
return JSONResponse(
|
||||
status_code=404,
|
||||
content={
|
||||
"error_code": "ZONGPAI_NOT_FOUND",
|
||||
"message": "未找到该总排号对应的排产号信息",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@app.exception_handler(DuplicateBoxItemError)
|
||||
async def duplicate_box_handler(request: Request, exc: DuplicateBoxItemError):
|
||||
return JSONResponse(
|
||||
status_code=409,
|
||||
content={
|
||||
"error_code": "DUPLICATE_BOX_NO",
|
||||
"message": f"排产号 {exc.paichan_no} 下箱号 {exc.box_no} 已存在",
|
||||
"paichan_no": exc.paichan_no,
|
||||
"box_no": exc.box_no,
|
||||
},
|
||||
)
|
||||
```
|
||||
|
||||
**Step 3: 启动验证**
|
||||
|
||||
Run: `cd apps/fastAPI && .venv/Scripts/python -m uvicorn app.main:app --reload`
|
||||
|
||||
访问 `http://127.0.0.1:8000/docs`，确认 Swagger UI 中出现 `GET /CargoTrace/box/info` 和 `POST /CargoTrace/box` 两个端点。
|
||||
|
||||
**Step 4: 快速功能验证**
|
||||
|
||||
使用 Swagger UI 或 curl 测试：
|
||||
- GET `/CargoTrace/box/info?zongpai_no=26BW0011` → 应返回 W00009 的箱号信息
|
||||
- POST `/CargoTrace/box` body: `{"zongpai_no": "26BW0011", "box_no": 99, "quantity": 5}` → 应返回 200
|
||||
|
||||
**Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add app/api/v1/box.py app/main.py
|
||||
git commit -m "feat: add GET /box/info and POST /box endpoints for boxing module"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 4: 测试 — 接口集成测试
|
||||
|
||||
**Files:**
|
||||
- Modify: `tests/conftest.py`
|
||||
- Create: `tests/test_box_api.py`
|
||||
|
||||
**Step 1: 更新 conftest.py — 添加装箱测试数据清理**
|
||||
|
||||
在现有 `cleanup_test_data` fixture 中追加装箱表的测试数据清理：
|
||||
|
||||
```python
|
||||
# 在 conftest.py 的 cleanup_test_data 中追加
|
||||
from app.models.finished_goods import FinishedGoodsBox, FinishedGoodsBoxItem
|
||||
|
||||
# 清理装箱测试数据
|
||||
test_box_items = db.query(FinishedGoodsBoxItem).filter(
|
||||
FinishedGoodsBoxItem.zongpai_no.in_(["26B997", "26B998", "26B999"])
|
||||
).all()
|
||||
for item in test_box_items:
|
||||
db.delete(item)
|
||||
|
||||
db.query(FinishedGoodsBox).filter(
|
||||
FinishedGoodsBox.paichan_no.in_(["TEST_PAICHAN_999"])
|
||||
).delete(synchronize_session=False)
|
||||
db.commit()
|
||||
```
|
||||
|
||||
> 注意：装箱测试使用 ERP 中真实存在的总排号 `26BW0011`（排产号 W00009）做查询测试，保存测试则使用无效的测试总排号（会被 404 拦截），避免污染真实数据。如需完整保存测试，使用临时箱号（如 999），测试后立即清理。
|
||||
|
||||
**Step 2: 创建 test_box_api.py**
|
||||
|
||||
```python
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
# --- GET /box/info 测试 ---
|
||||
|
||||
|
||||
def test_box_info_success(client: TestClient):
|
||||
"""查询真实存在的总排号 → 200"""
|
||||
resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "26BW0011"})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["zongpai_no"] == "26BW0011"
|
||||
assert data["paichan_no"] == "W00009"
|
||||
assert data["quantity"] == 80
|
||||
assert "existing_boxes" in data
|
||||
assert "max_box_no" in data
|
||||
assert data["suggested_box_no"] == data["max_box_no"] + 1
|
||||
|
||||
|
||||
def test_box_info_invalid_zongpai(client: TestClient):
|
||||
"""无效总排号格式 → 400"""
|
||||
resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "INVALID"})
|
||||
assert resp.status_code == 400
|
||||
assert resp.json()["error_code"] == "INVALID_ZONGPAI"
|
||||
|
||||
|
||||
def test_box_info_not_found(client: TestClient):
|
||||
"""格式合法但 ERP 中不存在 → 404"""
|
||||
resp = client.get("/CargoTrace/box/info", params={"zongpai_no": "26B99999"})
|
||||
assert resp.status_code == 404
|
||||
assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"
|
||||
|
||||
|
||||
# --- POST /box 测试 ---
|
||||
|
||||
|
||||
def test_box_save_invalid_zongpai(client: TestClient):
|
||||
"""无效总排号 → 400"""
|
||||
resp = client.post(
|
||||
"/CargoTrace/box",
|
||||
json={"zongpai_no": "INVALID", "box_no": 1, "quantity": 10},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert resp.json()["error_code"] == "INVALID_ZONGPAI"
|
||||
|
||||
|
||||
def test_box_save_invalid_box_no(client: TestClient):
|
||||
"""箱号 <= 0 → 400"""
|
||||
resp = client.post(
|
||||
"/CargoTrace/box",
|
||||
json={"zongpai_no": "26BW0011", "box_no": 0, "quantity": 10},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_box_save_invalid_quantity(client: TestClient):
|
||||
"""数量 <= 0 → 400"""
|
||||
resp = client.post(
|
||||
"/CargoTrace/box",
|
||||
json={"zongpai_no": "26BW0011", "box_no": 1, "quantity": 0},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_box_save_zongpai_not_found(client: TestClient):
|
||||
"""ERP 中不存在的总排号 → 404"""
|
||||
resp = client.post(
|
||||
"/CargoTrace/box",
|
||||
json={"zongpai_no": "26B99999", "box_no": 1, "quantity": 10},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"
|
||||
```
|
||||
|
||||
**Step 3: 运行测试**
|
||||
|
||||
Run: `cd apps/fastAPI && .venv/Scripts/python -m pytest tests/test_box_api.py -v`
|
||||
|
||||
**Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add tests/
|
||||
git commit -m "test: add integration tests for boxing API"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## API 接口汇总
|
||||
|
||||
| 方法 | 路径 | 描述 |
|
||||
|------|------|------|
|
||||
| GET | `/CargoTrace/box/info?zongpai_no=xxx` | 查询总排号对应的装箱信息 |
|
||||
| POST | `/CargoTrace/box` | 保存装箱记录 |
|
||||
|
||||
### 错误码
|
||||
|
||||
| HTTP 状态码 | error_code | 含义 |
|
||||
|-------------|-----------|------|
|
||||
| 400 | `INVALID_ZONGPAI` | 总排号格式不合法 |
|
||||
| 400 | `INVALID_BOX_NO` | 箱号不合法（≤0） |
|
||||
| 400 | `INVALID_QUANTITY` | 数量不合法（≤0） |
|
||||
| 404 | `ZONGPAI_NOT_FOUND` | 总排号在 ERP 中未找到对应排产号 |
|
||||
| 409 | `DUPLICATE_BOX_NO` | 该排产号下该箱号中已存在该总排号 |
|
||||
| 500 | `SERVER_ERROR` | 服务器内部错误 |
|
||||
78
tests/test_box_api.py
Normal file
78
tests/test_box_api.py
Normal file
@@ -0,0 +1,78 @@
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
# --- GET /box/info 测试 ---
|
||||
|
||||
|
||||
def test_box_info_success(client: TestClient):
|
||||
"""查询真实存在的总排号 → 200"""
|
||||
resp = client.get(
|
||||
"/CargoTrace/box/info", params={"zongpai_no": "26BW0011"}
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["zongpai_no"] == "26BW0011"
|
||||
assert data["paichan_no"] == "W00009"
|
||||
assert data["quantity"] == 80
|
||||
assert "existing_boxes" in data
|
||||
assert "max_box_no" in data
|
||||
assert data["suggested_box_no"] == data["max_box_no"] + 1
|
||||
|
||||
|
||||
def test_box_info_invalid_zongpai(client: TestClient):
|
||||
"""无效总排号格式 → 400"""
|
||||
resp = client.get(
|
||||
"/CargoTrace/box/info", params={"zongpai_no": "INVALID"}
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert resp.json()["error_code"] == "INVALID_ZONGPAI"
|
||||
|
||||
|
||||
def test_box_info_not_found(client: TestClient):
|
||||
"""格式合法但 ERP 中不存在 → 404"""
|
||||
resp = client.get(
|
||||
"/CargoTrace/box/info", params={"zongpai_no": "26B99999"}
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"
|
||||
|
||||
|
||||
# --- POST /box 测试 ---
|
||||
|
||||
|
||||
def test_box_save_invalid_zongpai(client: TestClient):
|
||||
"""无效总排号 → 400"""
|
||||
resp = client.post(
|
||||
"/CargoTrace/box",
|
||||
json={"zongpai_no": "INVALID", "box_no": 1, "quantity": 10},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert resp.json()["error_code"] == "INVALID_ZONGPAI"
|
||||
|
||||
|
||||
def test_box_save_invalid_box_no(client: TestClient):
|
||||
"""箱号 <= 0 → 400"""
|
||||
resp = client.post(
|
||||
"/CargoTrace/box",
|
||||
json={"zongpai_no": "26BW0011", "box_no": 0, "quantity": 10},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_box_save_invalid_quantity(client: TestClient):
|
||||
"""数量 <= 0 → 400"""
|
||||
resp = client.post(
|
||||
"/CargoTrace/box",
|
||||
json={"zongpai_no": "26BW0011", "box_no": 1, "quantity": 0},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_box_save_zongpai_not_found(client: TestClient):
|
||||
"""ERP 中不存在的总排号 → 404"""
|
||||
resp = client.post(
|
||||
"/CargoTrace/box",
|
||||
json={"zongpai_no": "26B99999", "box_no": 1, "quantity": 10},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"
|
||||
Reference in New Issue
Block a user