diff --git a/app/api/v1/attachment.py b/app/api/v1/attachment.py new file mode 100644 index 0000000..0c7962e --- /dev/null +++ b/app/api/v1/attachment.py @@ -0,0 +1,22 @@ +from fastapi import APIRouter, Depends +from sqlalchemy.orm import Session + +from app.core.database import get_db +from app.models.attachment import AttachmentCategory +from app.schemas.attachment import CategoryResponse + +router = APIRouter(tags=["attachment"]) + + +@router.get("/attachment/categories", response_model=list[CategoryResponse]) +def list_categories(db: Session = Depends(get_db)): + """列出启用中的附件类型，按 sort_order、name 排序。""" + rows = ( + db.query(AttachmentCategory) + .filter(AttachmentCategory.is_active == True) + .order_by(AttachmentCategory.sort_order, AttachmentCategory.name) + .all() + ) + return [ + CategoryResponse(id=r.id, name=r.name, sort_order=r.sort_order) for r in rows + ] diff --git a/app/main.py b/app/main.py index 52b71a3..beb0689 100644 --- a/app/main.py +++ b/app/main.py @@ -18,6 +18,7 @@ except Exception as e: from app.api.v1.location import router as location_router from app.api.v1.box import router as box_router +from app.api.v1.attachment import router as attachment_router from app.services.location_service import ( AlreadyOffShelfError, DuplicateLocationError, @@ -36,6 +37,7 @@ app = FastAPI(title="CargoTrace API", version="0.1.0") app.include_router(location_router, prefix="/CargoTrace") app.include_router(box_router, prefix="/CargoTrace") +app.include_router(attachment_router, prefix="/CargoTrace") @app.exception_handler(DuplicateLocationError) diff --git a/app/schemas/attachment.py b/app/schemas/attachment.py new file mode 100644 index 0000000..2fb142d --- /dev/null +++ b/app/schemas/attachment.py @@ -0,0 +1,169 @@ +from pydantic import BaseModel, Field, field_validator + + +class CategoryResponse(BaseModel): + id: int + name: str + sort_order: int + + +class AttachmentStatusItem(BaseModel): + category_id: int + name: str + expected_qty: int + boxed_qty: int + location_code: str | None = None + complete: bool + + +class AttachmentStatusResponse(BaseModel): + zongpai_no: str + determination: str + all_complete: bool + items: list[AttachmentStatusItem] = Field(default_factory=list) + + +class AttachmentConfigItem(BaseModel): + category_id: int + expected_qty: int + + @field_validator("expected_qty") + @classmethod + def _positive(cls, v: int) -> int: + if v <= 0: + raise ValueError("INVALID_QUANTITY") + return v + + +class AttachmentConfigRequest(BaseModel): + zongpai_no: str + determination: str + items: list[AttachmentConfigItem] = Field(default_factory=list) + + @field_validator("zongpai_no") + @classmethod + def _zongpai(cls, v: str) -> str: + import re + + v = v.strip().upper() + if not re.match(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$", v): + raise ValueError("INVALID_ZONGPAI") + return v + + @field_validator("determination") + @classmethod + def _det(cls, v: str) -> str: + if v not in ("none", "has"): + raise ValueError("INVALID_DETERMINATION") + return v + + +class AttachmentLocationRequest(BaseModel): + zongpai_no: str + category_id: int + location_code: str + + @field_validator("zongpai_no") + @classmethod + def _zongpai(cls, v: str) -> str: + import re + + v = v.strip().upper() + if not re.match(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$", v): + raise ValueError("INVALID_ZONGPAI") + return v + + @field_validator("location_code") + @classmethod + def _location(cls, v: str) -> str: + import re + + v = v.strip().upper() + normal = re.match(r"^[A-Z]+\d+-\d+-\d+$", v) + transit = re.match(r"^TRANS-\d+", v) + if not normal and not transit: + raise ValueError("INVALID_LOCATION") + return v + + +class AttachmentLocationResponse(BaseModel): + zongpai_no: str + category_id: int + location_code: str + created_at: str + previous_location: str | None = None + + +class AttachmentBoxSaveRequest(BaseModel): + zongpai_no: str + category_id: int + box_no: int + quantity: int + + @field_validator("zongpai_no") + @classmethod + def _zongpai(cls, v: str) -> str: + import re + + v = v.strip().upper() + if not re.match(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$", v): + raise ValueError("INVALID_ZONGPAI") + return v + + @field_validator("box_no") + @classmethod + def _box_no(cls, v: int) -> int: + if v <= 0: + raise ValueError("INVALID_BOX_NO") + return v + + @field_validator("quantity") + @classmethod + def _qty(cls, v: int) -> int: + if v <= 0: + raise ValueError("INVALID_QUANTITY") + return v + + +class AttachmentBoxSaveResponse(BaseModel): + box_item_id: int + paichan_no: str + box_no: int + zongpai_no: str + category_id: int + quantity: int + created_at: str + + +class AttachmentBoxUpdateRequest(BaseModel): + box_no: int + quantity: int + + @field_validator("box_no") + @classmethod + def _box_no(cls, v: int) -> int: + if v <= 0: + raise ValueError("INVALID_BOX_NO") + return v + + @field_validator("quantity") + @classmethod + def _qty(cls, v: int) -> int: + if v <= 0: + raise ValueError("INVALID_QUANTITY") + return v + + +class AttachmentBoxUpdateResponse(BaseModel): + box_item_id: int + paichan_no: str + box_no: int + zongpai_no: str + category_id: int + quantity: int + updated_at: str + + +class AttachmentBoxDeleteResponse(BaseModel): + box_item_id: int + deleted: bool diff --git a/tests/test_attachment_api.py b/tests/test_attachment_api.py new file mode 100644 index 0000000..1057765 --- /dev/null +++ b/tests/test_attachment_api.py @@ -0,0 +1,30 @@ +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.models.attachment import AttachmentCategory + + +def _seed_categories(db: Session) -> list[int]: + cats = [ + AttachmentCategory(name="TEST_CERT", sort_order=1, is_active=True), + AttachmentCategory(name="TEST_EXTRA", sort_order=2, is_active=True), + AttachmentCategory(name="TEST_OFF", sort_order=3, is_active=False), + ] + for c in cats: + db.add(c) + db.commit() + for c in cats: + db.refresh(c) + return [c.id for c in cats] + + +def test_categories_lists_active_only(client: TestClient, db: Session): + _seed_categories(db) + resp = client.get("/CargoTrace/attachment/categories") + assert resp.status_code == 200 + names = [c["name"] for c in resp.json()] + assert "TEST_CERT" in names + assert "TEST_EXTRA" in names + assert "TEST_OFF" not in names + # sorted by sort_order then name + assert names.index("TEST_CERT") < names.index("TEST_EXTRA")