CargoTrace/fastAPI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(box): include item total quantity

Browse Source
This commit is contained in:
Misaka_Company
2026-05-13 15:14:14 +08:00
parent afc540005c
commit bd926050c2
4 changed files with 167 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
app/schemas/box.py
View File

@@ -11,6 +11,7 @@ class BoxItemDetail(BaseModel):
zongpai_no: str
work_order_no: str | None = None
quantity: int
total_quantity: int | None = None
class CurrentZongpaiBox(BaseModel):

25
app/services/box_service.py
View File

@@ -68,21 +68,21 @@ def _erp_info_sql(db: Session) -> str:
)
def query_erp_work_order_map(db: Session, zongpai_nos: list[str]) -> dict[str, str | None]:
"""批量查询总排号对应的工令号。"""
def query_erp_item_info_map(db: Session, zongpai_nos: list[str]) -> dict[str, dict]:
"""批量查询总排号对应的工令号和 ERP 总数量。"""
if not zongpai_nos:
return {}
dialect_name = db.bind.dialect.name if db.bind is not None else ""
if dialect_name == "postgresql":
sql = text(
'SELECT "总排号", "工令号" '
'SELECT "总排号", "工令号", "数量" '
'FROM "ERPAuto"."vw_productionContractData" '
'WHERE "总排号" IN :zongpai_nos'
)
else:
sql = text(
"SELECT [总排号], [工令号] "
"SELECT [总排号], [工令号], [数量] "
"FROM [ERPAuto].[vw_productionContractData] "
"WHERE [总排号] IN :zongpai_nos"
)
@@ -90,7 +90,13 @@ def query_erp_work_order_map(db: Session, zongpai_nos: list[str]) -> dict[str, s
sql.bindparams(bindparam("zongpai_nos", expanding=True)),
{"zongpai_nos": sorted(set(zongpai_nos))},
).fetchall()
return {row[0]: row[1] for row in rows}
return {
row[0]: {
"work_order_no": row[1],
"total_quantity": int(row[2]) if row[2] is not None else None,
}
for row in rows
}
def get_box_info(db: Session, zongpai_no: str) -> dict:
@@ -120,7 +126,7 @@ def get_box_info(db: Session, zongpai_no: str) -> dict:
box_items[box.id] = items
item_zongpai_nos.extend(item.zongpai_no for item in items)
work_order_map = query_erp_work_order_map(db, item_zongpai_nos)
erp_item_info_map = query_erp_item_info_map(db, item_zongpai_nos)
existing_boxes = []
current_zongpai_boxes = []
@@ -140,8 +146,13 @@ def get_box_info(db: Session, zongpai_no: str) -> dict:
{
"box_item_id": item.id,
"zongpai_no": item.zongpai_no,
"work_order_no": work_order_map.get(item.zongpai_no),
"work_order_no": erp_item_info_map.get(
item.zongpai_no, {}
).get("work_order_no"),
"quantity": int(item.quantity or 0),
"total_quantity": erp_item_info_map.get(
item.zongpai_no, {}
).get("total_quantity"),
}
for item in items
],

22
tests/conftest.py
View File

@@ -4,7 +4,11 @@ from sqlalchemy.orm import Session
from app.core.database import SessionLocal
from app.main import app
from app.models.finished_goods import FinishedGoodsLocation
from app.models.finished_goods import (
FinishedGoodsBox,
FinishedGoodsBoxItem,
FinishedGoodsLocation,
)
@pytest.fixture
@@ -34,4 +38,20 @@ def cleanup_test_data(db: Session):
db.query(FinishedGoodsLocation).filter(
FinishedGoodsLocation.zongpai_no.in_(test_zongpai_nos)
).delete(synchronize_session=False)
test_boxes = (
db.query(FinishedGoodsBox)
.filter(
FinishedGoodsBox.paichan_no == "W00009",
FinishedGoodsBox.box_no >= 900,
)
.all()
)
test_box_ids = [box.id for box in test_boxes]
if test_box_ids:
db.query(FinishedGoodsBoxItem).filter(
FinishedGoodsBoxItem.box_id.in_(test_box_ids)
).delete(synchronize_session=False)
db.query(FinishedGoodsBox).filter(
FinishedGoodsBox.id.in_(test_box_ids)
).delete(synchronize_session=False)
db.commit()

127
tests/test_box_api.py
View File

@@ -16,7 +16,11 @@ def test_box_info_success(client: TestClient):
assert "work_order_no" in data
assert data["work_order_no"] is None or isinstance(data["work_order_no"], str)
assert data["quantity"] == 80
assert "current_zongpai_boxes" in data
assert "existing_boxes" in data
for box in data["existing_boxes"]:
for item in box["items"]:
assert "box_item_id" in item
assert "max_box_no" in data
assert data["suggested_box_no"] == data["max_box_no"] + 1
@@ -78,3 +82,126 @@ def test_box_save_zongpai_not_found(client: TestClient):
)
assert resp.status_code == 404
assert resp.json()["error_code"] == "ZONGPAI_NOT_FOUND"
def test_box_save_many_to_one_appends_different_zongpais(client: TestClient):
"""多码一箱：同一排产号同一箱号可追加不同总排号。"""
box_no = 901
first = client.post(
"/CargoTrace/box",
json={
"zongpai_no": "26BW0011",
"box_no": box_no,
"quantity": 10,
"box_mode": "many-to-one",
},
)
assert first.status_code == 200
assert first.json()["box_item_id"] is not None
second = client.post(
"/CargoTrace/box",
json={
"zongpai_no": "26BW0012",
"box_no": box_no,
"quantity": 8,
"box_mode": "many-to-one",
},
)
assert second.status_code == 200
assert second.json()["box_no"] == box_no
assert second.json()["zongpai_no"] == "26BW0012"
info = client.get("/CargoTrace/box/info", params={"zongpai_no": "26BW0011"})
assert info.status_code == 200
target_box = next(
box for box in info.json()["existing_boxes"] if box["box_no"] == box_no
)
assert {item["zongpai_no"] for item in target_box["items"]} == {
"26BW0011",
"26BW0012",
}
for item in target_box["items"]:
assert "work_order_no" in item
assert isinstance(item["total_quantity"], int)
def test_box_save_many_to_one_duplicate_same_zongpai(client: TestClient):
"""多码一箱：同箱同总排重复录入返回 DUPLICATE_BOX_NO。"""
box_no = 902
first = client.post(
"/CargoTrace/box",
json={
"zongpai_no": "26BW0011",
"box_no": box_no,
"quantity": 10,
"box_mode": "many-to-one",
},
)
assert first.status_code == 200
duplicate = client.post(
"/CargoTrace/box",
json={
"zongpai_no": "26BW0011",
"box_no": box_no,
"quantity": 5,
"box_mode": "many-to-one",
},
)
assert duplicate.status_code == 409
assert duplicate.json()["error_code"] == "DUPLICATE_BOX_NO"
def test_box_update_same_box_no_does_not_conflict(client: TestClient):
"""PATCH 原箱号未变时，应排除当前 box_item_id，避免误判重复。"""
box_no = 903
created = client.post(
"/CargoTrace/box",
json={
"zongpai_no": "26BW0011",
"box_no": box_no,
"quantity": 10,
"box_mode": "one-to-many",
},
)
assert created.status_code == 200
box_item_id = created.json()["box_item_id"]
updated = client.patch(
f"/CargoTrace/box/{box_item_id}",
json={"box_no": box_no, "quantity": 7, "box_mode": "one-to-many"},
)
assert updated.status_code == 200
assert updated.json()["quantity"] == 7
def test_box_delete_removes_empty_box(client: TestClient):
"""DELETE 删除最后一条明细后，同步清理空箱号。"""
box_no = 904
created = client.post(
"/CargoTrace/box",
json={
"zongpai_no": "26BW0011",
"box_no": box_no,
"quantity": 10,
"box_mode": "many-to-one",
},
)
assert created.status_code == 200
box_item_id = created.json()["box_item_id"]
deleted = client.delete(f"/CargoTrace/box/{box_item_id}")
assert deleted.status_code == 200
assert deleted.json() == {"box_item_id": box_item_id, "deleted": True}
recreated = client.post(
"/CargoTrace/box",
json={
"zongpai_no": "26BW0012",
"box_no": box_no,
"quantity": 8,
"box_mode": "one-to-many",
},
)
assert recreated.status_code == 200