WorkBuddy+Python设计商品库存管理系统：从需求拆分到核心代码落地

在实际开发中“用 WorkBuddyPython 设计商品库存管理系统”并不是让 AI 自动把所有代码写完而是先借助 WorkBuddy 这类 AI 工作台把一句自然语言需求拆成可落地的表结构、服务规则和接口协议再用 Python 把它们实现成一个可以运行、可以测试、可以继续迭代的最小系统。库存管理表面上只是商品的增删改查真正决定工程质量的是库存不足怎么处理、出库入库怎么记录、并发下单时库存是否会被扣成负数、数据和金额用浮点还是精确数值等边界问题。这篇文章会从一句需求开始给出提示词写法、FastAPI SQLAlchemy SQLite 的最小实现、接口验证方式、常见坑排查和生产化改造方向适合正在学习 Python Web 开发或准备把 AI 生成的需求文档落成代码的读者。1. 先理解“一句话设计”背后的完整链路1.1 一句话需求不等于一张数据表标题里的“一句话设计”非常容易让人误解。很多人以为对 AI 工作台说一句“帮我做一个商品库存管理系统”就能拿到一个生产可用的系统。真实情况是这句话只确定了业务方向并没有确定边界系统是给仓库管理员用还是给运营看报表需要支持扫码枪吗库存是按单品数量扣减还是按多属性规格扣减出库后是否允许负库存这些问题如果不提前定义AI 生成的结果只能是“看起来完整实际不能用”的通用模板。把“设计”两个字落到技术上至少需要拆出几个部分商品数据商品编码、名称、分类、单位、默认价格。库存数据当前库存数量、占用数量、可用数量、库存预警阈值。库存流水每次入库、出库、盘点调整都必须记录来源和去向。操作规则库存不足能否出库退货是否回填库存并发扣减如何避免超卖。对外接口前端页面、订单系统或移动端扫码时通过什么 API 发起操作。如果这五类信息没有在一开始对齐写出来的库存表只是“一堆字段”不是可靠的数据结构。1.2 最小可用库存系统应该包含什么功能学习阶段不应该一上来就做多仓、多币种、成本核算、批次追踪那会让最简单的一条链路上叠加大量无关复杂度。适合作为练手项目的“最小可用库存系统”只需要承担以下职责功能说明核心问题商品建档新增商品维护 SKU、名称、价格、单位商品编码不能重复入库采购入库或退货入库增加库存同一商品多次入库要累加数量出库销售出库或领用出库扣减库存库存不足时要明确报错策略库存查询查看当前每个商品的库存快照快照与流水必须一致流水追溯每一次数量变化都能看到操作时间、类型、数量、关联单号修改库存时必须同时写流水安全库存提示低于阈值时提示补货阈值字段建议可配置这也是后面实现时所有代码和接口的最小范围。超出这个范围的功能建议放到二期来做。1.3 为什么商品表、库存表和流水表要分开新手常见做法是把“库存数量”直接放到商品表入库就quantity 1出库就quantity - 1最后整个系统只有一张商品表。这种方式对纯展示作业没问题但对真实库存管理有很大隐患你不知道当前库存是哪些操作累积出来的也无法排查“为什么库存对不上”。正确的做法是分成两层当前库存快照层保存最新状态便于查询。库存变动流水层保存每一次操作记录作为审计和重建依据。快照可以被修改流水只能追加。如果发现快照错误不能直接改商品表里的库存数字而是先写一条“盘点调整”流水再更新快照。这样每一次库存变化都有迹可循这也是“商品库存管理系统”和“普通商品管理”的本质区别。注意库存系统里快照数据是结果操作流水才是证据。任何绕过流水直接修改库存的做法都会让系统的可追溯性失效。2. 向 WorkBuddy 这类 AI 工作台下出“能落地”的提示词2.1 提示词里要把约束写清楚而不是只写目标WorkBuddy 使用过程中最容易出现的问题是用户对 AI 工作台描述业务时过于笼统。把“帮我设计商品库存管理系统”扔进对话框和把一段包含用户角色、业务规则、输出格式、禁止事项的提示词扔进对话框得到的结果质量差别非常大。针对库存项目我建议把提示词拆成四段分别描述背景、对象、规则和输出格式。下面是一个可以直接复制的提示词模板实际使用时需要根据你安装的 WorkBuddy 版本和交互界面调整我希望你以一个高级产品经理兼后端架构师的身份帮我完成「商品库存管理系统」的需求分析和数据结构设计。 项目背景 - 这是一个面向小团队仓储场景的库存管理系统。 - 管理员需要维护商品信息完成采购入库、销售出库、库存查询。 - 需要记录每次库存变化的流水。 - 需要支持安全库存提醒。 业务规则 - 商品编码唯一不允许重复建档。 - 出库时如果可用库存不足系统必须拒绝并返回明确错误消息。 - 数量必须使用正整型或精确数值类型禁止使用浮点数累计库存。 - 每次入库/出库/盘点调整都必须生成一条流水记录。 - 系统需要区分“当前库存快照”和“库存流水明细”两张不同职责的表。 请输出 1. 一份需求清单按 P0/P1/P2 标注优先级。 2. 数据库表设计字段需要包含类型、约束、说明。 3. 3 个核心 API 接口的请求和响应示例。 4. 至少列出 5 个最容易出错的设计点。这段提示词的关键点不是字数多而是把“系统做成什么样”换成了“系统在什么约束下工作”。AI 工作台无法知道你的业务规则但你能明确告诉它“如果库存不足怎么办”“如果商品编码重复怎么办”。当规则足够清楚时生成结果才具备实现参考价值。2.2 拿到 AI 输出后先检查哪些内容把 AI 工作台给出的设计直接当成终稿是另一种常见错误。AI 生成内容的主要价值是提供结构、字段思路和 API 草案但它不会为这段代码在你的实际业务里是否正确负责。拿到生成结果后建议按下面的清单检查检查项不通过时如何处理是否区分商品表和库存表不区分则说明设计仍然停留在 CRUD没有库存语义是否包含流水表缺失流水会导致无法审计数量字段类型是否精确使用浮点数时要求改为 DECIMAL 或 int是否有库存不足校验没有就补上服务层校验API 是否包含操作人、操作类型、备注缺少审计字段会在排障时很被动是否说明并发情况下如何扣库存没有说明则需要对实现方案再确认检查这一步决定了后面的 Python 代码是否可靠。不要在模糊的设计上直接写代码否则返工成本会转嫁到后端服务层。2.3 常见的提示词翻车现象很多人使用 AI 工作台时会遇到三种情况第一是生成的代码里大量使用不存在的依赖或过度封装。解决方式是要求 AI 明确输出 requirements.txt再和你本地的 Python 版本核对。第二是接口设计过于理想化比如一次出库请求同时要求更新商品表、插入历史表、刷新缓存但给的代码里根本没有事务控制。解决方式是把“必须在事务内完成所有写操作”加入提示词约束。第三是生成的数据字段类型和你要用的数据库不匹配。比如把在 MySQL 里常用的auto_increment直接带进 PostgreSQL 语法。这并不代表 AI 不聪明而是你没有在提示词里说明实际使用的数据库类型。因此在提示词末尾补充“请使用 SQLAlchemy 2.x SQLite 作为本地开发数据库”会让结果更贴近可运行代码。3. Python 环境与库存项目结构3.1 先确认 Python 安装状态项目标题里虽然强调设计但真正验证设计是否合理仍然要把代码跑起来。建议本地安装 Python 3.10 或更高版本因为后续使用的 Pydantic 2.x 和 FastAPI 新版本对 3.10 更新支持更好。在终端执行下面命令确认 Python 是否已安装python --versionWindows 系统中如果命令提示找不到 Python可以使用py --versionmacOS / Linux 系统里如果同时存在多个 Python 版本建议先确认python3指向哪个版本python3 --version如果没有安装去 Python 官网下载对应系统安装包即可。安装时要注意勾选“Add Python to PATH”否则后续在终端里使用命令行会非常麻烦。3.2 创建并激活虚拟环境每个 Python 项目都应该拥有独立的虚拟环境而不是把依赖安装到全局环境。原因是不同项目对 FastAPI、SQLAlchemy、Pydantic 的版本要求可能不一致一旦全局冲突排查起来要花很多时间。创建项目目录并进入mkdir workbuddy-inventory cd workbuddy-inventory在项目根目录创建虚拟环境python -m venv venvWindows 激活方式venv\Scripts\activatemacOS / Linux 激活方式source venv/bin/activate激活后终端前面会出现(venv)标记。如果你明明在执行命令却没有看到这个标记说明虚拟环境没有激活后面安装的依赖很可能装到了系统解释器里。注意不要直接运行pip install而跳过虚拟环境。库存项目用到 FastAPI、Uvicorn、SQLAlchemy、Pydantic这些包一旦装进系统级 Python后续升级和卸载都可能影响其他项目。3.3 安装项目依赖在虚拟环境激活状态下创建requirements.txtfastapi0.111.0 uvicorn[standard]0.30.1 SQLAlchemy2.0.30 pydantic2.8.2然后安装依赖pip install -r requirements.txtFastAPI 负责 Web 接口层Uvicorn 负责启动本地服务SQLAlchemy 负责数据库访问Pydantic 负责请求参数和响应数据的校验。这套组合在重量级框架面前足够轻量同时也保留了后续切换 MySQL、PostgreSQL 的空间。安装完成后可以执行pip list看到版本号后说明依赖安装完成。如果这里出现网络超时可以使用国内镜像源安装例如清华或阿里云的 PyPI 镜像。3.4 项目目录结构为了避免把所有代码写进一个main.py我建议按模块拆分。学习项目不需要太重但至少要区分数据库层、模型层、服务层和路由层workbuddy-inventory/ ├── venv/ ├── requirements.txt ├── app/ │ ├── __init__.py │ ├── database.py │ ├── models.py │ ├── schemas.py │ ├── services.py │ ├── main.py │ └── routers/ │ ├── __init__.py │ └── inventory.py └── stock.dbmodels.py放 SQLAlchemy 的 ORM 模型schemas.py放 Pydantic 的请求和响应模型services.py放核心库存操作逻辑routers/inventory.py放 HTTP 路由main.py创建 FastAPI 实例并注册路由。这样拆分的好处是当后续需要加入订单逻辑、商品分类、库存预警时不需要重写现有模块。4. 数据库模型与表关系设计4.1 Product 与 Stock 拆分的原因前面已经说过商品基础信息和库存快照要分开。下面这种设计是最基本的字段Product商品表Stock库存表主键idid唯一编码skuproduct_id 外键业务信息name, category, unitquantity金额信息price无状态字段is_active无Product保存商品的静态信息SKU 唯一。Stock保存动态库存数量一个商品对应一条库存记录。Stock中不重复保存商品名称、价格等静态数据否则商品改名时还得同步修改多个表。下面再看完整字段表表字段类型说明productsidInteger PK自增主键productsskuString(64) unique商品编码productsnameString(128)商品名称productscategoryString(64)分类productsunitString(16)单位productspriceNumeric(10,2)参考售价productsis_activeBoolean是否启用stockidInteger PK自增主键stockproduct_idInteger FK关联商品stockquantityInteger可用库存数量stocksafety_stockInteger安全库存阈值库存数量在最小版本里使用整数。如果实际业务需要支持小数重量或容积单位可以把数量列改为Numeric(14,3)但要同步修改所有涉及判断“库存是否足够”的逻辑。4.2 SQLAlchemy 模型代码在models.py中定义 ORM 模型from datetime import datetime from sqlalchemy import ( Boolean, DateTime, ForeignKey, Integer, Numeric, String, ) from sqlalchemy.orm import Mapped, mapped_column, relationship from .database import Base class Product(Base): __tablename__ products id: Mapped[int] mapped_column(primary_keyTrue, indexTrue) sku: Mapped[str] mapped_column(String(64), uniqueTrue, indexTrue) name: Mapped[str] mapped_column(String(128)) category: Mapped[str] mapped_column(String(64), default) unit: Mapped[str] mapped_column(String(16), default件) price: Mapped[float] mapped_column(Numeric(10, 2), default0) is_active: Mapped[bool] mapped_column(Boolean, defaultTrue) stock: Mapped[Stock] relationship( back_populatesproduct, cascadeall, delete-orphan, ) class Stock(Base): __tablename__ stock id: Mapped[int] mapped_column(primary_keyTrue, indexTrue) product_id: Mapped[int] mapped_column( ForeignKey(products.id), uniqueTrue, indexTrue ) quantity: Mapped[int] mapped_column(Integer, default0) safety_stock: Mapped[int] mapped_column(Integer, default0) product: Mapped[Product] relationship(back_populatesstock) class StockMovement(Base): __tablename__ stock_movements id: Mapped[int] mapped_column(primary_keyTrue, indexTrue) product_id: Mapped[int] mapped_column( ForeignKey(products.id), indexTrue ) change_type: Mapped[str] mapped_column(String(16)) quantity: Mapped[int] mapped_column(Integer) before_quantity: Mapped[int] mapped_column(Integer, default0) after_quantity: Mapped[int] mapped_column(Integer, default0) operator: Mapped[str] mapped_column(String(64), default) remark: Mapped[str] mapped_column(String(255), default) created_at: Mapped[datetime] mapped_column( DateTime, defaultdatetime.utcnow )这里有一个设计细节stock_movements表保存了before_quantity和after_quantity。很多库存流水表只保存变动数量不保存变动前后的快照这会导致以后重建现场时要慢慢推算。把操作前数量和操作后数量同时写进流水排障能省很多时间。4.3 初始化数据库连接在database.py中创建引擎和会话from sqlalchemy import create_engine from sqlalchemy.orm import DeclarativeBase, sessionmaker DATABASE_URL sqlite:///./stock.db engine create_engine( DATABASE_URL, connect_args{check_same_thread: False}, ) SessionLocal sessionmaker(bindengine, autoflushFalse, expire_on_commitFalse) class Base(DeclarativeBase): passSQLite 设置check_same_threadFalse是为了允许 FastAPI 在不同线程中访问同一个 SQLite 连接对象。虽然这是本地开发通用写法但如果把项目切换到 PostgreSQL需要删除这个参数并引入真正的连接池配置。还需要在main.py或启动脚本中创建所有表。最简单的方式是在 FastAPI 启动前调用一次Base.metadata.create_all(bindengine)from fastapi import FastAPI from .database import Base, engine from .routers import inventory Base.metadata.create_all(bindengine) app FastAPI(titleWorkBuddy Inventory API) app.include_router(inventory.router, prefix/api/v1, tags[inventory])这个初始化方式适合学习项目。生产项目建议使用alembic这类迁移工具管理表结构变更而不是每次启动都执行create_all。4.4 为什么库存数量不用浮点数价格字段使用Numeric(10, 2)而不是 Python 原生float原因是浮点数在存储十进制小数时存在误差。举个最简单的例子0.1 0.2在 Python 中的结果是0.30000000000000004。如果库存数量或金额长期使用浮点数参与运算最终会出现对账不平、精度丢失等问题而且很难定位到底是哪一笔操作造成的。学习阶段如果只维护整数库存可以直接用Integer。一旦涉及到价格或小数数量数据库字段就要用Numeric/DECIMALPython 侧用Decimal解析避免在边界上踩浮点坑。5. 核心库存服务入库、出库、流水5.1 服务层为什么要单独存在如果直接把数据库读写逻辑写在路由函数里一个小项目也能跑通但后续会出问题。比如业务上新增一个“只有管理员才能出库”的规则你要在多个路由里重复添加权限判断如果库存不足的校验逻辑写了两份其中一份漏掉就会出现库存负数。因此库存核心逻辑必须收敛到services.py中。这一层的核心职责是根据 SKU 找到商品。检查商品状态。检查库存数量是否满足出库。在同一个事务中更新库存并写入流水。出现异常时回滚整个事务。5.2 入库服务创建商品时可以同时创建库存记录这样新增商品后默认库存为零。执行入库时直接增加stock.quantity并写一条stock_movements记录from datetime import datetime from sqlalchemy.orm import Session from .models import Product, Stock, StockMovement def create_product_with_stock( db: Session, sku: str, name: str, category: str, unit: str, price: float, ) - Product: exist_product db.query(Product).filter(Product.sku sku).first() if exist_product: raise ValueError(f商品 SKU 已存在: {sku}) product Product( skusku, namename, categorycategory, unitunit, priceprice, ) db.add(product) db.flush() stock Stock(product_idproduct.id, quantity0) db.add(stock) db.commit() db.refresh(product) return product def stock_in( db: Session, sku: str, quantity: int, operator: str admin, remark: str , ) - Stock: if quantity 0: raise ValueError(入库数量必须大于 0) product db.query(Product).filter(Product.sku sku).first() if not product: raise ValueError(f商品不存在: {sku}) stock db.query(Stock).filter(Stock.product_id product.id).first() if not stock: stock Stock(product_idproduct.id, quantity0) db.add(stock) db.flush() before_quantity stock.quantity stock.quantity quantity movement StockMovement( product_idproduct.id, change_typeIN, quantityquantity, before_quantitybefore_quantity, after_quantitystock.quantity, operatoroperator, remarkremark, ) db.add(movement) db.commit() db.refresh(stock) return stockchange_type在最小版本中可以约定为IN表示入库OUT表示出库ADJUST表示盘点调整。使用字符串是为了方便阅读如果项目更复杂可以把它改成枚举常量类避免散落魔法值。5.3 出库服务与库存不足校验出库逻辑的关键是判断“扣减后是否低于零”。最简单可靠的校验方式是在内存中判断if stock.quantity quantity: raise ValueError(库存不足)但这种方式在单进程学习中没问题并发场景里同一时间可能有两个请求同时读到quantity10同时执行扣减最终导致库存变成负数或超卖。要解决并发问题不能只依赖这里的代码还需要配合数据库行锁或原子更新后面第 8 节会专门讲。先看最小版本def stock_out( db: Session, sku: str, quantity: int, operator: str admin, remark: str , ) - Stock: if quantity 0: raise ValueError(出库数量必须大于 0) product db.query(Product).filter(Product.sku sku).first() if not product: raise ValueError(f商品不存在: {sku}) stock db.query(Stock).filter(Stock.product_id product.id).first() if not stock: raise ValueError(f商品库存记录不存在: {sku}) if stock.quantity quantity: raise ValueError( f库存不足商品 {sku} 当前库存 {stock.quantity} f本次出库 {quantity} ) before_quantity stock.quantity stock.quantity - quantity movement StockMovement( product_idproduct.id, change_typeOUT, quantityquantity, before_quantitybefore_quantity, after_quantitystock.quantity, operatoroperator, remarkremark, ) db.add(movement) db.commit() db.refresh(stock) return stock这里有两个容易忽略的点。第一quantity必须在使用前先转换成正整数否则如果接口传来负数负数入库会减少库存负数出库会增加库存语义完全颠倒。第二before_quantity必须在修改stock.quantity之前取出来这样才能保证流水里记录的是操作发生前的真实快照。如果先改库存再记录历史数据就全错了。5.4 查询库存与流水查询当前库存可以按 SKU 精准查询也可以列出所有商品和库存。为了方便前端展示我通常把商品信息和库存信息合并返回def get_stock_by_sku(db: Session, sku: str) - dict: product db.query(Product).filter(Product.sku sku).first() if not product: raise ValueError(f商品不存在: {sku}) stock ( db.query(Stock).filter(Stock.product_id product.id).first() ) quantity stock.quantity if stock else 0 safety_stock stock.safety_stock if stock else 0 return { sku: product.sku, name: product.name, quantity: quantity, safety_stock: safety_stock, status: LOW if quantity safety_stock else OK, }查询流水时按商品 ID 过滤并限制返回条目数避免一次把全表数据读到内存def list_movements( db: Session, sku: str, limit: int 20, offset: int 0 ) - list[StockMovement]: product db.query(Product).filter(Product.sku sku).first() if not product: raise ValueError(f商品不存在: {sku}) return ( db.query(StockMovement) .filter(StockMovement.product_id product.id) .order_by(StockMovement.id.desc()) .limit(limit) .offset(offset) .all() )查询接口不需要把服务层所有细节暴露给调用方但在返回值里保留status字段可以让前端在库存低于安全阈值时更容易做颜色标记和补货提醒。5.5 使用数据库事务的边界上面的代码里每次操作都调用了db.commit()。这意味着一次入库或出库要么全成功要么全失败。如果更新库存成功而写入流水失败SQLAlchemy 的同一事务会自动回滚不会出现“库存变了但没有流水”的中间状态。注意不要在服务层函数内使用裸的try...except吞掉异常后继续提交。正确的做法是让异常向上抛出由 FastAPI 全局异常处理器统一转换成 HTTP 错误这样客户端能拿到稳定结构的错误返回。6. API 层、运行和接口验证6.1 路由定义接口不用设计得太多。先把核心动作暴露出来创建商品POST /api/v1/products查询库存GET /api/v1/stock/{sku}入库POST /api/v1/stock/in出库POST /api/v1/stock/out查询流水GET /api/v1/products/{sku}/movements在routers/inventory.py中实现from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field from sqlalchemy.orm import Session from ..database import SessionLocal from ..services import ( create_product_with_stock, get_stock_by_sku, list_movements, stock_in, stock_out, ) router APIRouter() class ProductCreate(BaseModel): sku: str Field(..., min_length1, max_length64) name: str Field(..., min_length1, max_length128) category: str unit: str 件 price: float 0 class StockChange(BaseModel): sku: str quantity: int Field(..., gt0) operator: str admin remark: str def get_db(): db SessionLocal() try: yield db finally: db.close() def handle_service_error(exc: Exception) - HTTPException: return HTTPException(status_code400, detailstr(exc)) router.post(/products) def create_product(body: ProductCreate, db: Session Depends(get_db)): try: product create_product_with_stock( db, skubody.sku, namebody.name, categorybody.category, unitbody.unit, pricebody.price, ) except ValueError as exc: raise handle_service_error(exc) return { id: product.id, sku: product.sku, name: product.name, unit: product.unit, } router.get(/stock/{sku}) def get_stock(sku: str, db: Session Depends(get_db)): try: return get_stock_by_sku(db, sku) except ValueError as exc: raise handle_service_error(exc) router.post(/stock/in) def stock_in_api(body: StockChange, db: Session Depends(get_db)): try: stock stock_in( db, skubody.sku, quantitybody.quantity, operatorbody.operator, remarkbody.remark, ) except ValueError as exc: raise handle_service_error(exc) return {sku: body.sku, quantity: stock.quantity} router.post(/stock/out) def stock_out_api(body: StockChange, db: Session Depends(get_db)): try: stock stock_out( db, skubody.sku, quantitybody.quantity, operatorbody.operator, remarkbody.remark, ) except ValueError as exc: raise handle_service_error(exc) return {sku: body.sku, quantity: stock.quantity} router.get(/products/{sku}/movements) def movements(sku: str, db: Session Depends(get_db)): try: items list_movements(db, sku) except ValueError as exc: raise handle_service_error(exc) return [ { id: item.id, type: item.change_type, quantity: item.quantity, before_quantity: item.before_quantity, after_quantity: item.after_quantity, operator: item.operator, remark: item.remark, created_at: item.created_at.isoformat(), } for item in items ]这个路由文件里多写了一些自定义异常转换因为services.py使用ValueError表达业务校验失败而 FastAPI 不会自动把ValueError变成 400 响应。所以函数在出口处把它们变成统一的HTTPException。6.2 Pydantic 模型和输入校验ProductCreate和StockChange中已经使用了 Pydantic 字段校验。Field(..., gt0)表示quantity必须大于 0。如果调用方传了 0 或负数FastAPI 会在进入业务代码之前直接返回 422 校验失败不需要再在服务层重复判断非正整数。BaseModel里价格字段仍然写成了float这是示例中的简化写法。在实际生产代码中请求和响应的价格字段应定义为字符串或 Decimal然后由 Pydantic 校验格式避免浮点误差提前进入系统。6.3 启动服务在项目根目录执行uvicorn app.main:app --reload --host 0.0.0.0 --port 8000看到类似下面日志说明服务已经启动INFO: Uvicorn running on http://0.0.0.0:8000 INFO: Application startup complete.浏览器访问http://127.0.0.1:8000/docs可以看到 FastAPI 自动生成的 Swagger 文档页面。这是验证接口非常方便的工具不需要额外安装 Postman。6.4 用 curl 跑一遍完整库存链路下面用一个完成链路验证系统是否满足核心规则。创建商品curl -X POST http://127.0.0.1:8000/api/v1/products \ -H Content-Type: application/json \ -d {\sku\: \SKU001\, \name\: \机械键盘\, \category\: \外设\, \unit\: \件\, \price\: 299}预期返回{ id: 1, sku: SKU001, name: 机械键盘, unit: 件 }入库 100 件curl -X POST http://127.0.0.1:8000/api/v1/stock/in \ -H Content-Type: application/json \ -d {\sku\: \SKU001\, \quantity\: 100, \operator\: \purchaser\, \remark\: \采购入库\}预期返回{ sku: SKU001, quantity: 100 }查询库存curl http://127.0.0.1:8000/api/v1/stock/SKU001预期返回{ sku: SKU001, name: 机械键盘, quantity: 100, safety_stock: 0, status: OK }出库 30 件curl -X POST http://127.0.0.1:8000/api/v1/stock/out \ -H Content-Type: application/json \ -d {\sku\: \SKU001\, \quantity\: 30, \operator\: \seller\, \remark\: \销售出库\}预期库存变成 70。再次出库 100 件此时服务层会拒绝curl -X POST http://127.0.0.1:8000/api/v1/stock/out \ -H Content-Type: application/json \ -d {\sku\: \SKU001\, \quantity\: 100, \operator\: \seller\, \remark\: \测试超卖\}预期返回 HTTP 400{ detail: 库存不足商品 SKU001 当前库存 70本次出库 100 }这个失败用例非常重要因为它验证了库存系统最关键的一条规则不允许库存扣成负数。如果真实环境返回了成功说明services.py里的校验没有生效或者调用路径没有经过服务层。查询流水curl http://127.0.0.1:8000/api/v1/products/SKU001/movements预期返回两条流水一条入库、一条出库且每条都记录了before_quantity和after_quantity。6.5 准备一个最小单元测试除了人工 curl还可以用 pytest 为库存服务写一个自动化测试。先把 pytest 写入依赖pytest8.2.2安装后创建test_services.pyfrom app.database import Base, SessionLocal, engine from app.services import create_product_with_stock, stock_in, stock_out def setup_function(): Base.metadata.drop_all(bindengine) Base.metadata.create_all(bindengine) def test_stock_out_reject_when_insufficient(): db SessionLocal() create_product_with_stock( db, skuTEST001, name测试商品, category测试, unit件, price10, ) stock_in(db, TEST001, 5) try: stock_out(db, TEST001, 6) except ValueError as exc: assert 库存不足 in str(exc) else: raise AssertionError(库存不足时应抛出 ValueError) db.close()这个测试覆盖了最容易出现的业务故障。写 CI 时这样的小测试可以防止后续改代码把库存不足校验误删掉。7. 库存系统最容易踩的五个坑7.1 主键自增和 SKU 唯一性问题现象创建第二个相同 SKU 商品时系统没有报错反而生成了两条商品记录。原因要么数据库表没有给sku字段加唯一约束要么代码里先执行了db.commit()再用同一个 session 查询导致约束错误被吞掉。处理方式数据库models.py中给sku加uniqueTrue。在创建商品前先查询一次存在则直接抛出业务错误。如果希望数据库层面兜住并发重复创建可以进一步捕获IntegrityError再返回友好提示。7.2 使用浮点数累加库存和金额现象系统正常跑了很久某天对账时发现某些商品的库存或金额总和比流水的数学结果多了一点点或少了一点点。原因models.py中数量字段用了Numeric但 Python 侧用float赋值或直接在代码里执行浮点运算。处理方式库存数量用int金额用Numeric(10, 2)在 Pydantic 请求模型中使用字符串或 Decimal 接收价格并在服务层用 Decimal 做计算。检查方式在 Python REPL 中执行0.1 0.2如果看到0.30000000000000004就不要在核心账务逻辑里使用 float。7.3 库存不足判断和扣减不在同一事务现象在低并发测试看不出问题一旦用脚本同时发 100 个出库请求库存会变成负数。原因代码先if stock.quantity quantity判断再执行扣减但两个操作之间没有防止其他请求修改同一条记录也没有使用数据库行锁。处理方式学习阶段至少保证判断和扣减在同一个数据库事务中。生产环境把查询换成SELECT ... FOR UPDATE锁住库存行再执行扣减。7.4 新增商品时没有同时创建库存记录现象商品创建成功了但执行入库时报“库存记录不存在”。原因商品表和库存表分开后新增接口只插入了商品没有插入默认库存记录或没有在服务层自动补齐。处理方式在create_product_with_stock中调用db.flush()和db.add(stock)保证每个新商品都有一条初始库存记录。同时在入库、出库服务中如果查询到的是空库存仍然允许初始化一条quantity0的记录。7.5 SQLite 并发写出现 database is locked现象压测或同时操作多个窗口时API 偶尔返回database is locked。原因SQLite 对并发写支持有限。多个线程同时写同一数据库文件后到的写事务会被锁拒绝。处理方式学习阶段使用本地 SQLite 没有问题。部署到生产或多进程环境时把DATABASE_URL切换到 PostgreSQL 或 MySQL并增加连接池配置。8. 从学习项目到生产级库存系统还需要做什么8.1 数据库从 SQLite 切换到 PostgreSQLSQLite 的优势是零配置、单文件适合学习和本地验证。生产环境写并发场景建议切换到 PostgreSQL。切换时只需要修改database.py中的DATABASE_URL但有几个细节需要一并处理SQLite 的check_same_threadFalse不再需要。SQLAlchemy 模型中的日期默认值建议使用数据库端的CURRENT_TIMESTAMP。建表和字段变更改用 Alembic 迁移脚本不要在启动时create_all。确认数据库连接的连接池大小、超时时间与业务并发量匹配。DATABASE_URL改成 PostgreSQL 后类似DATABASE_URL postgresqlpsycopg://user:password127.0.0.1:5432/inventory安装驱动pip install psycopg[binary]注意要在requirements.txt里补充新依赖。8.2 解决并发扣减问题最简单的鲁棒方案是为出库操作增加行锁。在 SQLAlchemy 中可以使用with_for_update()stock ( db.query(Stock) .filter(Stock.product_id product.id) .with_for_update() .first() )这段代码对应的 SQL 效果是SELECT ... FOR UPDATE它会锁住库存行直到当前事务提交或回滚。第二个并发出库请求只能等待第一个事务完成随后重新读取最新库存因此不会出现两个请求都基于旧库存做判断的情况。同时出库接口需要真正开一个数据库事务不能在每次查询后自动提交。8.3 增加操作鉴权和操作人身份当前接口的operator字段由调用方直接传入这很容易被伪造。生产环境应该通过登录态、Token 或内部服务鉴权获得操作人而不是信任请求体里的字符串。最小落地方案是加一个简单的X-Operator请求头由网关或认证中间件填充后传给服务层。更进一步需要引入 JWT、RBAC、操作审计日志。8.4 补充日志、监控和预警库存系统需要三类日志请求日志谁在什么时候通过哪个接口改了什么库存。业务日志库存不足、库存过低等业务事件。数据一致性日志周期性的库存快照比对发现差异时告警。除了日志还应该把“库存低于安全阈值”“扣减失败次数”“数据库锁等待时间”暴露成监控指标。生产发布前至少确认日志有统一采集不能只写在本地文件里。8.5 生产发布前检查清单下面这份清单可以直接复制到发布文档里逐项确认检查项完成标准数据库迁移使用 Alembic 管理表结构不在代码中直接 create_all唯一约束SKU 和商品 ID 有数据库唯一索引库存不足校验出库必须校验库存并有自动化测试覆盖金额类型金额和价格使用 Decimal禁止 float操作审计每一次变更都有操作人、操作类型、操作前后值事务边界一次出库/入库的所有写操作在同一个事务中并发方案生产环境使用行锁或原子更新避免超卖日志采集业务日志集中管理错误日志能按时间定位权限控制创建商品、出库、入库接口都有鉴权和权限校验回滚方案数据库有备份代码版本支持平滑回滚8.6 后续扩展方向最小版本完成后还有很多可以继续深入的方向增加盘点功能用ADJUST类型记录盘盈盘亏。增加“占用库存”和“可用库存”字段销售下单时先锁定库存支付后再真正扣减。增加多仓库模型把库存拆成warehouse_id product_id维度。增加条码、扫码枪对接、Excel 导入导出。增加 Redis 缓存商品热数据降低数据库压力。把 WorkBuddy 生成的提示词和项目文档沉淀为团队内部模板让后续新系统设计更规范。实际项目里最值得花时间打磨的不是 CRUD 代码而是库存不足、流水追溯、并发控制、金额精度和权限审计这些边缘场景。把这些练熟后再遇到订单系统、财务系统或供应链系统你会更容易判断哪些逻辑需要放进服务层哪些规则必须提前写清楚。真正有经验的后端工程师往往不是能写出更快的 SQL而是能在设计阶段就预见数据会怎么变、订单会怎么并发、账目要怎么追溯。这个商品库存项目就是一个很适合用来建立这种判断力的起点。