fastAPI/docs/plans/2026-05-11-shelf-registration.md

上架登记模块 (Shelf Registration) Implementation Plan

For Claude: REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.

Goal: 实现 PRD v1.0 中定义的上架登记后端接口 POST /CargoTrace/location，支持总排号与货位号的绑定。

Architecture: FastAPI + SQLAlchemy (同步模式) 连接 SQL Server。三层结构：API 路由层 → Service 业务层 → Model 数据层。使用 pydantic-settings 管理 .env 配置，Alembic 管理数据库迁移（当前版本表已存在，仅做初始化）。

Tech Stack: Python 3.12+, FastAPI, SQLAlchemy 2.x (sync), pyodbc, pydantic-settings, python-dotenv

---

项目目录结构（最终目标）

apps/fastAPI/
├── .env                        # 数据库连接配置
├── .gitignore
├── requirements.txt
├── README.md
├── alembic.ini
├── alembic/
├── app/
│   ├── __init__.py
│   ├── main.py                 # FastAPI 应用入口
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py           # pydantic-settings 配置
│   │   └── database.py         # SQLAlchemy engine & session
│   ├── models/
│   │   ├── __init__.py
│   │   └── finished_goods.py   # 3 张表的 ORM 模型
│   ├── schemas/
│   │   ├── __init__.py
│   │   ├── location.py         # 上架登记请求/响应 schema
│   │   └── common.py           # 通用错误响应 schema
│   ├── api/
│   │   ├── __init__.py
│   │   └── v1/
│   │       ├── __init__.py
│   │       └── location.py     # POST /CargoTrace/location
│   └── services/
│       ├── __init__.py
│       └── location_service.py # 上架登记业务逻辑
├── tests/
│   ├── __init__.py
│   ├── conftest.py             # 测试 fixtures (TestClient, mock DB)
│   └── test_location_api.py    # 接口测试

---

Task 1: 项目基础设施 — 配置与环境

Files:

• Create: apps/fastAPI/.env
• Create: apps/fastAPI/.gitignore
• Modify: apps/fastAPI/requirements.txt
• Create: apps/fastAPI/app/core/__init__.py
• Create: apps/fastAPI/app/core/config.py
• Create: apps/fastAPI/app/core/database.py

Step 1: 更新 requirements.txt

fastapi>=0.115.0
uvicorn[standard]>=0.34.0
sqlalchemy>=2.0.0
pyodbc>=5.2.0
pydantic-settings>=2.0.0
python-dotenv>=1.0.0

Step 2: 创建 .env 文件

# SQL Server
SQL_SERVER_HOST=192.168.110.114
SQL_SERVER_PORT=1433
SQL_SERVER_DATABASE=CompanyDB
SQL_SERVER_USERNAME=peng
SQL_SERVER_PASSWORD=Cqbld123456.
SQL_SERVER_DRIVER={ODBC Driver 18 for SQL Server}
SQL_SERVER_TRUST_SERVER_CERTIFICATE=yes

Step 3: 创建 .gitignore

.venv/
__pycache__/
*.pyc
.env
*.egg-info/
dist/
build/
.pytest_cache/

Step 4: 创建 app/core/config.py

from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    SQL_SERVER_HOST: str
    SQL_SERVER_PORT: int = 1433
    SQL_SERVER_DATABASE: str
    SQL_SERVER_USERNAME: str
    SQL_SERVER_PASSWORD: str
    SQL_SERVER_DRIVER: str = "{ODBC Driver 18 for SQL Server}"
    SQL_SERVER_TRUST_SERVER_CERTIFICATE: str = "yes"

    @property
    def database_url(self) -> str:
        return (
            f"mssql+pyodbc://{self.SQL_SERVER_USERNAME}:{self.SQL_SERVER_PASSWORD}"
            f"@{self.SQL_SERVER_HOST},{self.SQL_SERVER_PORT}/{self.SQL_SERVER_DATABASE}"
            f"?driver={self.SQL_SERVER_DRIVER}"
            f"&TrustServerCertificate={self.SQL_SERVER_TRUST_SERVER_CERTIFICATE}"
        )

    model_config = {"env_file": ".env", "extra": "ignore"}


settings = Settings()

Step 5: 创建 app/core/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase, sessionmaker

from app.core.config import settings

engine = create_engine(settings.database_url)
SessionLocal = sessionmaker(bind=engine)


class Base(DeclarativeBase):
    pass


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

Step 6: 验证数据库连接

Run: cd apps/fastAPI && source .venv/bin/activate && python -c "from app.core.database import engine; engine.connect(); print('DB OK')" Expected: DB OK

Step 7: Commit

git add .env .gitignore requirements.txt app/core/
git commit -m "feat: add project config and database connection"

---

Task 2: ORM 模型 — 三张表映射

Files:

• Create: apps/fastAPI/app/models/__init__.py
• Create: apps/fastAPI/app/models/finished_goods.py

Step 1: 创建 models/finished_goods.py

from datetime import datetime

from sqlalchemy import BigInteger, DateTime, Index, Numeric, String, func
from sqlalchemy.orm import Mapped, mapped_column

from app.core.database import Base

SCHEMA = "CargoTrace"


class FinishedGoodsLocation(Base):
    __tablename__ = "finished_goods_location"
    __table_args__ = (
        Index("uk_zongpai_no", "zongpai_no", unique=True),
        Index("idx_fgl_location_code", "location_code"),
        {"schema": SCHEMA},
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
    location_code: Mapped[str] = mapped_column(String(64), nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, server_default=func.getdate()
    )


class FinishedGoodsBox(Base):
    __tablename__ = "finished_goods_box"
    __table_args__ = (
        Index("uk_paichan_box", "paichan_no", "box_no", unique=True),
        Index("idx_fgb_paichan_no", "paichan_no"),
        {"schema": SCHEMA},
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    paichan_no: Mapped[str] = mapped_column(String(64), nullable=False)
    box_no: Mapped[int] = mapped_column(nullable=False)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, server_default=func.getdate()
    )


class FinishedGoodsBoxItem(Base):
    __tablename__ = "finished_goods_box_item"
    __table_args__ = (
        Index("idx_fgbi_box_id", "box_id"),
        Index("idx_fgbi_zongpai_no", "zongpai_no"),
        {"schema": SCHEMA},
    )

    id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
    box_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
    zongpai_no: Mapped[str] = mapped_column(String(64), nullable=False)
    quantity: Mapped[float | None] = mapped_column(Numeric(18, 3), nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime, nullable=False, server_default=func.getdate()
    )

Step 2: 创建 models/init.py

from app.models.finished_goods import (
    FinishedGoodsBox,
    FinishedGoodsBoxItem,
    FinishedGoodsLocation,
)

__all__ = [
    "FinishedGoodsLocation",
    "FinishedGoodsBox",
    "FinishedGoodsBoxItem",
]

Step 3: 验证模型可正常导入

Run: cd apps/fastAPI && python -c "from app.models import FinishedGoodsLocation; print('Models OK')" Expected: Models OK

Step 4: Commit

git add app/models/
git commit -m "feat: add ORM models for finished goods tables"

---

Task 3: Pydantic Schemas — 请求/响应模型

Files:

• Create: apps/fastAPI/app/schemas/__init__.py
• Create: apps/fastAPI/app/schemas/common.py
• Create: apps/fastAPI/app/schemas/location.py

Step 1: 创建 schemas/common.py — 通用错误响应

from pydantic import BaseModel


class ErrorResponse(BaseModel):
    error_code: str
    message: str
    detail: dict | None = None

Step 2: 创建 schemas/location.py — 上架登记请求/响应

import re

from pydantic import BaseModel, field_validator

# 总排号正则: 两位年份 + 类型字母(B/C/T) + 流水号 或 两位年份 + 温度计(BW/CW) + 4位流水号
ZONGPAI_PATTERN = re.compile(r"^(\d{2}(B|C|T)\d+|\d{2}(BW|CW)\d{4})$")
# 普通货位号: 区域货架-层-格 (全大写字母+数字, 横杠分隔)
LOCATION_NORMAL_PATTERN = re.compile(r"^[A-Z]+\d+-\d+-\d+$")
# 转运特殊货位号: TRANS-序号
LOCATION_TRANS_PATTERN = re.compile(r"^TRANS-\d+")


class LocationRequest(BaseModel):
    zongpai_no: str
    location_code: str

    @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("location_code")
    @classmethod
    def validate_location_code(cls, v: str) -> str:
        v = v.strip().upper()
        if not LOCATION_NORMAL_PATTERN.match(v) and not LOCATION_TRANS_PATTERN.match(v):
            raise ValueError("INVALID_LOCATION")
        return v


class LocationResponse(BaseModel):
    zongpai_no: str
    location_code: str
    created_at: str


class DuplicateLocationDetail(BaseModel):
    existing_location: str
    created_at: str

Step 3: 创建 schemas/init.py

Step 4: Commit

git add app/schemas/
git commit -m "feat: add pydantic schemas for location API"

---

Task 4: Service 层 — 上架登记业务逻辑

Files:

• Create: apps/fastAPI/app/services/__init__.py
• Create: apps/fastAPI/app/services/location_service.py

Step 1: 创建 services/location_service.py

from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.models.finished_goods import FinishedGoodsLocation
from app.schemas.location import LocationRequest


def register_location(db: Session, req: LocationRequest) -> FinishedGoodsLocation:
    existing = (
        db.query(FinishedGoodsLocation)
        .filter(FinishedGoodsLocation.zongpai_no == req.zongpai_no)
        .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(),
                },
            },
        )

    record = FinishedGoodsLocation(
        zongpai_no=req.zongpai_no,
        location_code=req.location_code,
    )
    db.add(record)
    db.commit()
    db.refresh(record)
    return record

Step 2: 创建 services/init.py

Step 3: Commit

git add app/services/
git commit -m "feat: add location registration service"

---

Task 5: API 路由层 — POST /CargoTrace/location

Files:

• Create: apps/fastAPI/app/api/__init__.py
• Create: apps/fastAPI/app/api/v1/__init__.py
• Create: apps/fastAPI/app/api/v1/location.py
• Modify: apps/fastAPI/app/main.py

Step 1: 创建 api/v1/location.py

from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from app.core.database import get_db
from app.schemas.location import LocationRequest, LocationResponse
from app.services.location_service import register_location

router = APIRouter()


@router.post("/location", response_model=LocationResponse)
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 2: 更新 main.py — 挂载路由

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

from app.api.v1.location import router as location_router

app = FastAPI(title="CargoTrace API", version="0.1.0")

app.include_router(location_router, prefix="/CargoTrace")


@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,
        content={
            "error_code": error_code,
            "message": str(exc),
        },
    )


@app.get("/")
async def root():
    return {"message": "Welcome to CargoTrace API"}

Step 3: 启动验证

Run: cd apps/fastAPI && uvicorn app.main:app --reload Expected: 服务启动成功，访问 http://127.0.0.1:8000/docs 可看到 Swagger UI

Step 4: Commit

git add app/api/ app/main.py
git commit -m "feat: add POST /CargoTrace/location endpoint"

---

Task 6: 测试 — 接口集成测试

Files:

• Modify: apps/fastAPI/requirements.txt (添加测试依赖)
• Create: apps/fastAPI/tests/__init__.py
• Create: apps/fastAPI/tests/conftest.py
• Create: apps/fastAPI/tests/test_location_api.py

Step 1: 添加测试依赖到 requirements.txt

追加重定内容:

pytest>=8.0.0
httpx>=0.28.0

Step 2: 创建 tests/conftest.py — 测试 fixtures，使用真实数据库

import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session

from app.core.database import SessionLocal
from app.main import app
from app.models.finished_goods import FinishedGoodsLocation


@pytest.fixture
def client():
    return TestClient(app)


@pytest.fixture
def db():
    session = SessionLocal()
    yield session
    session.close()


@pytest.fixture(autouse=True)
def cleanup_test_data(db: Session):
    yield
    # 清理测试数据：删除测试用总排号记录
    test_zongpai_nos = ["26B999", "26C999", "26BW0999", "26T999"]
    db.query(FinishedGoodsLocation).filter(
        FinishedGoodsLocation.zongpai_no.in_(test_zongpai_nos)
    ).delete(synchronize_session=False)
    db.commit()

Step 3: 创建 tests/test_location_api.py

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["detail"]["error_code"] == "DUPLICATE_LOCATION"
    assert "existing_location" in data["detail"]["detail"]


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 == 422  # Pydantic validation error


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 == 422


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"

Step 4: 运行测试

Run: cd apps/fastAPI && pytest tests/ -v Expected: 5 tests passed

Step 5: Commit

git add requirements.txt tests/
git commit -m "test: add integration tests for location API"

---

API 接口汇总

方法	路径	描述
POST	/CargoTrace/location	上架登记 — 绑定总排号与货位号

错误码

HTTP 状态码	error_code	含义
409	DUPLICATE_LOCATION	该总排号已存在货位记录
422	INVALID_ZONGPAI	总排号格式不合法
422	INVALID_LOCATION	货位号格式不合法
500	SERVER_ERROR	服务器内部错误