用Python和FastAPI打造超市管理系统：表设计、事务与并发控制

简介：一份基于Python开发的超市管理系统毕业设计资料包，面向计算机相关专业在校学生，可用于毕业设计、课程设计、项目初期演示，也适合新手学习Python项目从设计到落地的完整流程。资源覆盖系统设计、功能实现与文档整理，能帮助读者快速理解超市管理系统的代码结构和业务流程，并在此基础上继续完善或二次开发。

压缩包共9个文件，整体约9.89MB，包含Python源码、可直接运行的exe程序、Word用户手册与大作业说明文档，以及TXT和Markdown格式的说明文件，既便于在本地演示系统效果，也方便对照文档阅读代码。资源同时保留了一个zip子包，供进一步解压查看内部资料。整体容量较小，下载与使用都比较轻量。

目前已有391人学习或下载该资源。对正在筹备毕业设计、课程设计或作业的学生来说，这套资料提供了可复用的完整案例：代码经过运行测试，文档覆盖操作说明与设计说明，能够帮助快速搭建一个可演示的超市管理系统，同时也能作为Python应用开发入门的参考资料。

1. 一个被低估的课设题目，才是练 Python 后端的好机会

超市管理系统在毕业设计里出现频率极高，但多数人拿到题目后的第一反应是"太土了"。实际上，这个题目覆盖了 Python 后端开发最核心的几条主线：关系型数据库建模、ORM 操作、分层架构设计、权限控制和增删改查之外的事务处理。如果你把收银结算、库存扣减、供应商结算这三件事真正想清楚，写出来的系统比很多空谈微服务的项目更扎实。

本文不讨论论文怎么排版、答辩 PPT 怎么做，只讲技术层面如何把一个超市管理系统从零搭到可演示：库存、商品、收银、会员、供应商这几张表怎么设计，FastAPI 和 SQLAlchemy 怎么组织，收银台结账时库存和流水怎么保证一致，以及哪些细节决定了答辩时能不能扛住老师的追问。适合正在做课设或想拿完整项目练手的开发者。

2. 先把数据模型立住：超市管理系统最关键的 6 张表和它们的关系

2.1 表结构设计不是越多越好，而是恰好覆盖业务闭环

超市管理系统的业务闭环是"进货 → 上架 → 销售 → 结算 → 补货"。围绕这条链路，最少需要 6 张表：

初学者最常见的错误是只建 product 和 sale_order 两张表，把购买明细直接塞进一个 text 字段。这样写 demo 可以，但老师一旦问"怎么统计某个商品的月销量"，就只能现写字符串解析代码，属于给自己挖坑。上述 9 张表已经覆盖了从进货到销售再到库存追溯的完整闭环，且每张表都有明确职责，后续写 SQL 统计会非常直接。

2.2 用 SQLAlchemy 定义模型时，重点处理三个关系

在models.py中定义数据模型，最常见的做法是使用 SQLAlchemy 2.x 的 Declarative 方式。以下代码是 purchase_order 与 purchase_item 的一对多关系，以及 sku 表的多对多骨架：

from sqlalchemy import ( ForeignKey, Numeric, Integer, String, Text, DateTime, func ) from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship from datetime import datetime class Base(DeclarativeBase): pass class PurchaseOrder(Base): __tablename__ = "purchase_order" id: Mapped[int] = mapped_column(Integer, primary_key=True) supplier_id: Mapped[int] = mapped_column(ForeignKey("supplier.id")) order_no: Mapped[str] = mapped_column(String(32), unique=True) total_amount: Mapped[float] = mapped_column(Numeric(10, 2)) status: Mapped[int] = mapped_column(Integer, default=0) # 0=待入库, 1=已入库, 2=已作废 created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) items: Mapped[list["PurchaseItem"]] = relationship( back_populates="order", cascade="all, delete-orphan" ) class PurchaseItem(Base): __tablename__ = "purchase_item" id: Mapped[int] = mapped_column(Integer, primary_key=True) order_id: Mapped[int] = mapped_column(ForeignKey("purchase_order.id")) product_id: Mapped[int] = mapped_column(ForeignKey("product.id")) quantity: Mapped[int] = mapped_column(Integer) price: Mapped[float] = mapped_column(Numeric(10, 2)) amount: Mapped[float] = mapped_column(Numeric(10, 2)) order: Mapped["PurchaseOrder"] = relationship(back_populates="items") product: Mapped["Product"] = relationship()

逻辑说明：Numeric(10, 2)用于金额字段，避免 Float 的二进制精度问题，这在超市结算场景中会直接影响对账结果。cascade="all, delete-orphan"表示删除进货单时同时删除明细，但前提是进货单处于未入库状态，这个校验写业务层而不是数据库层。relationship()会自动处理外键关联，查询进货单时通过order.items即可拿到全部明细，无需手写 join。

参数说明：进货单状态status我习惯用整数枚举而不是字符串，因为后续如果用 FastAPI 做接口，前端下拉框传数字比传中文更不容易出错；入库操作时通过事务同时修改purchase_order.status和商品库存。

2.3 库存表为什么不单独设计，而是挂在 product 上

有些管理系统会把库存单独拆成 inventory 表，理由是"商品档案"和"库存数量"生命周期不同。但对超市这种小规模场景，单独拆表会让代码多一层 join，而且并发扣库存时还得处理两行数据的锁。我一般直接把stock字段放在 product 表上，同时用 stock_log 表记录每一次变动，这样既能实时查库存，又能回溯每笔变动。

入库、销售、盘点调整都写 stock_log，change_type 用数字约定：

这样设置的收益是：搜索"超市管理系统 库存流水"这个关键词的人，最终都会意识到一条变动手写一条日志比直接改库存数字更可靠。后面所有报表统计也可以直接基于 stock_log 聚合，不必去翻订单明细。真正写代码时，这条日志的插入必须和库存变更放在同一个数据库事务里，否则会出现库存改了但日志没记，或者日志记了但库存没改的不一致状态。

3. 用 FastAPI 把核心接口写出来：入库、商品管理、收银台、会员折扣

3.1 项目目录和启动方式，让环境可复现

动手写代码前先把目录固定下来。以下是我常用的分层结构：

app/ ├── main.py # FastAPI 入口 ├── models.py # SQLAlchemy 模型 ├── schemas.py # Pydantic 请求/响应模型 ├── database.py # 数据库连接和会话 ├── routers/ │ ├── product.py # 商品管理 │ ├── purchase.py # 进货入库 │ ├── sale.py # 收银台 │ └── supplier.py # 供应商 ├── services/ │ ├── stock_service.py # 库存变动逻辑 │ └── sale_service.py # 结算逻辑 └── utils/ └── order_no.py # 订单号生成

启动方式很常规，本地用 SQLite 足够，演示时无需额外装数据库；部署答辩再切换 MySQL，只需改database.py里的连接串。SQLite 和 MySQL 在 SQLAlchemy 下切换成本极低，这也是选 SQLAlchemy 而不是直接拼 SQL 的原因。

database.py里的核心会话设置如下：

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker DATABASE_URL = "sqlite:///./supermarket.db" engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

逻辑说明：autocommit=False是必须的，它要求你在业务代码里显式调用db.commit()，这样才能把多个写操作包进一个事务。autoflush=False意思是查询前不自动把待提交的改动刷到数据库，避免在复杂业务逻辑中产生预期之外的 SQL。

3.2 商品管理接口：条码查询、库存状态、价格修改

商品管理是整个系统的入口，条码是超市场景里的核心检索字段。以下接口实现了通过条码精确查询和模糊搜索：

from fastapi import APIRouter, Depends, HTTPException from sqlalchemy.orm import Session @router.get("/products") def list_products( keyword: str | None = None, category_id: int | None = None, low_stock_only: bool = False, db: Session = Depends(get_db), ): query = db.query(Product) if keyword: # 条码精确匹配优先，名称模糊搜索兜底 query = query.filter( (Product.barcode == keyword) | (Product.name.like(f"%{keyword}%")) ) if category_id: query = query.filter(Product.category_id == category_id) if low_stock_only: query = query.filter(Product.stock <= Product.safety_stock) return query.order_by(Product.id.desc()).limit(100).all()

参数说明：low_stock_only=True是判断是否需要补货的接口，超市管理里这个功能高频使用。keyword先比对条码再做名称模糊匹配，是因为扫码枪输入的就是条码，精确匹配效率远高于 like。返回值建议只返回前 100 条，超市商品总量一般不会超过几万，limit 足够演示分页思路。

修改库存价格时要写操作日志，价格变化直接影响毛利统计，所以 product 表要加updated_at时间戳。

3.3 入库接口：进货单 + 明细 + 库存变更放在一个事务里

入库是培训新手理解事务的绝佳场景。进货单创建时只写入库单数据，点"确认入库"时才动库存，这样设计是避免前端网络请求超时时库存已经改了但用户以为没成功。

@router.post("/purchase_orders/{order_id}/confirm") def confirm_purchase(order_id: int, db: Session = Depends(get_db)): order = db.get(PurchaseOrder, order_id) if not order: raise HTTPException(status_code=404, detail="进货单不存在") if order.status != 0: raise HTTPException(status_code=400, detail="只有待入库状态才能确认入库") try: for item in order.items: product = db.get(Product, item.product_id) old_stock = product.stock product.stock += item.quantity db.add(StockLog( product_id=item.product_id, change_type=1, quantity=item.quantity, before_stock=old_stock, after_stock=product.stock, )) order.status = 1 db.commit() except Exception: db.rollback() raise HTTPException(status_code=500, detail="入库失败，事务已回滚") return {"message": "入库成功", "order_id": order_id}

逻辑说明：db.get()按主键取对象，取不到直接抛 404。product.stock += item.quantity是在 Python 内存层面修改对象属性，提交时 SQLAlchemy 会生成 update 语句。StockLog 记录变动前后的库存值，这一步的意义在于后续可以审计每一次入库的来源。

注意，这里务必要用db.commit()一次性提交所有改动。如果中途抛异常，rollback()会让前面已修改的 product 和已插入的 stock_log 全部撤销，库存和日志保持一致。有一个常见误用是在 confirm 接口里不检查 status 就允许重复入库，这会导致库存翻倍，是答辩时老师最爱问的一句话。

3.4 收银台是系统复杂度最高的部分，设计结算逻辑

收银台接收购物车列表，把会员折扣、库存扣减、销售单生成三者合并到一个接口里。以下代码给出完整实现：

@router.post("/checkout") def checkout( payload: CheckoutPayload, db: Session = Depends(get_db), ): # 1. 生成唯一订单号 order_no = generate_order_no() # 2. 计算总金额并校验库存 sale_items = [] total_amount = 0.0 for item in payload.items: product = db.query(Product).filter_by(barcode=item.barcode).first() if not product: raise HTTPException(status_code=404, detail=f"条码 {item.barcode} 不存在") if product.stock < item.quantity: raise HTTPException(status_code=400, detail=f"{product.name} 库存不足") amount = product.price * item.quantity total_amount += amount sale_items.append({ "product": product, "quantity": item.quantity, "amount": amount, }) # 3. 会员折扣：按 95 折，积分抵现按实际规则 discount_rate = 1.0 if payload.member_phone: member = db.query(Member).filter_by(phone=payload.member_phone).first() if member: discount_rate = 0.95 # 会员默认 95 折 actual_amount = round(total_amount * discount_rate, 2) # 4. 创建销售单和明细 order = SaleOrder( order_no=order_no, total_amount=actual_amount, member_id=member.id if member else None, ) db.add(order) db.flush() # 获取自增主键 for data in sale_items: product = data["product"] db.add(SaleItem( order_id=order.id, product_id=product.id, quantity=data["quantity"], price=product.price, amount=data["amount"], )) # 5. 扣库存 old_stock = product.stock product.stock -= data["quantity"] db.add(StockLog( product_id=product.id, change_type=2, quantity=data["quantity"], before_stock=old_stock, after_stock=product.stock, )) db.commit() return { "order_no": order_no, "total_amount": actual_amount, "discount_amount": round(total_amount - actual_amount, 2), }

参数说明：db.flush()在提交前把 order 写入数据库拿到自增 id，这样 sale_item 才能引用order_id。此时事务未提交，失败回滚时这些 id 也随之作废。折扣逻辑直接用0.95硬编码演示，实际系统应该从会员等级表读取；把折扣率设计成可配置项，方便答辩时演示不同等级会员的不同折扣。

这段代码同时覆盖了库存校验、金额计算、日志记录，是整套系统里信息量最大的部分。设计成"先全部校验通过，再统一写入"，避免前几件商品扣了库存，后面某个条码不存在导致整体失败却留下了脏数据。

3.5 订单号生成逻辑，避免并发重复

订单号生成有一个反直觉的点：直接用数据库自增 id 做订单号，在超市这种并发场景中很容易因为多线程同时插入造成重复。更稳妥的办法是基于时间戳加随机数：

from datetime import datetime import random def generate_order_no(): current = datetime.now().strftime("%Y%m%d%H%M%S") suffix = str(random.randint(1000, 9999)) return f"SO{current}{suffix}"

逻辑说明：SO前缀区分销售单，PO前缀给进货单。时间戳精确到秒，同一秒内最多生成 9000 个不重复的单号，对单个超市足够了。随机数只是降低纯时间戳撞车概率，不必依赖全局锁。注意入库单号也要用同样的规则生成，不要两个模块各写一套，统一放到utils/order_no.py里。

4. 报表统计与可视化：用 SQL 别用 Python 循环去算毛利

4.1 销售统计的正确打开方式是 group by，而不是在 Python 里循环累加

写管理系统时最常见的坏味道是查出所有订单到 Python 里做循环求和。数据量小的时候感觉不到问题，但只要销量过千条，接口响应时间会明显变慢。SQLAlchemy 直接支持聚合查询，用 group_by 在数据库里就把结果算好。

下面是按日期汇总销售额和订单量：

from sqlalchemy import func @router.get("/reports/daily_sales") def daily_sales( start_date: str, end_date: str, db: Session = Depends(get_db), ): results = ( db.query( func.date(SaleOrder.created_at).label("day"), func.count(SaleOrder.id).label("order_count"), func.sum(SaleOrder.total_amount).label("sales_amount"), ) .filter(SaleOrder.created_at >= start_date) .filter(SaleOrder.created_at <= end_date + " 23:59:59") .group_by(func.date(SaleOrder.created_at)) .order_by(func.date(SaleOrder.created_at)) .all() ) return [dict(row._mapping) for row in results]

参数说明：func.date()把 datetime 字段截断到天，group by正是按天聚合的关键。end_date + " 23:59:59"是为了把结束日期当天的数据全部纳入统计，否则今天 00:00 之后到当前时间之间的订单会被漏掉。row._mapping是 SQLAlchemy 2.x 推荐的取列名方式，比_asdict()更安全。

4.2 商品毛利排行：关联订单明细和商品成本

很多人只在 product 表里存了price（售价），漏了cost_price（进价），导致毛利根本算不出来。有成本价后，排行逻辑简单且可靠：

SELECT p.name AS product_name, SUM(si.quantity) AS total_qty, SUM(si.amount) AS total_sales, SUM(si.quantity * p.cost_price) AS total_cost, (SUM(si.amount) - SUM(si.quantity * p.cost_price)) AS profit FROM sale_item si JOIN product p ON p.id = si.product_id JOIN sale_order so ON so.id = si.order_id WHERE so.created_at >= :start_date AND so.created_at <= :end_date GROUP BY p.id ORDER BY profit DESC LIMIT 20;

SQL 说明：联表查询在 SQLAlchemy 里可以写成一次 query，也可以直接用text()执行原生 SQL。二者在超市管理场景下性能差异不大，关键是 title 里提到"详细文档"时，你要能解释清楚 JOIN 的意义：sale_item 只有 product_id，商品名和成本在 product 表里，不 join 拿不到。成本用的是当前cost_price，如果进价经常波动，更严谨的做法是把成本也冗余进 sale_item，这就是回头改表时要考虑的优化方向了。

4.3 低库存预警在系统里如何体现

低库存预警不一定要做推送，先在页面上把列表做出来就够了。最常见的设计是查询库存小于安全库存的商品，并在后台首页显示数量角标。SQL 写法：

SELECT * FROM product WHERE stock <= safety_stock ORDER BY stock ASC;

在 dashboard 接口里，同时返回低库存商品列表和低库存总数。扩展到报损、退货等场景时可以共用一个stock_log模型，按 change_type 做区分。对课设而言，把低库存预警做成一个独立接口，比和其他数据混在一个响应里更容易讲清楚逻辑。

5. 收银高频场景的并发与数据一致性检查：先加悲观锁还是先扣库存

5.1 扫描枪连点导致超卖，问题出在"校验再扣减"这两步之间

本文前面给的收银代码在单线程演示环境没有问题，但如果有两个收银员同时卖同一件商品，就可能出现超卖。原因很简单：两个请求都读到了库存 3，都判断 3 > 1，然后各自扣减，库存变成 2 而不是 1。数据库层面的解决办法是加锁。

在 SQLAlchemy 中对 product 行加悲观锁，写法如下：

product = ( db.query(Product) .filter(Product.barcode == item.barcode) .with_for_update() .first() )

with_for_update()翻译成 SQL 就是SELECT ... FOR UPDATE，这一行数据在事务提交前被数据库锁住，其他事务的同类查询会等待。超市管理系统演示时一般不会触发这个锁，但答辩时回答"如何避免超卖"这个问题，这段代码能实打实撑住场面。

悲观锁的代价是并发性能下降，不过超市收银台规模远未到这个瓶颈。如果你向往上游走，可以改用乐观锁：在 product 表加version字段，扣减时拼上WHERE version = 上次读到的值，更新成功再version + 1。哪种更合适，取决于系统对吞吐量的预期。

5.2 一张脑图式的代码执行顺序，扣库存之前必须重查库存

即使加了FOR UPDATE，仍然要在修改前重新判断product.stock < item.quantity。因为加锁后读到的才是最新值，之前那个未加锁的查询结果可能已经过期。正确顺序是：

先锁 → 再读 → 判库存 → 扣减 → 记流水 → 提交

把这条顺序记在代码注释里，比什么都管用。任何跳过锁直接更新的写法，都会在并发测试下暴露超卖问题。如果答辩老师要求你"把并发安全考虑进去"，这一段内容可以直接背上。

5.3 数据一致性验证：写一个并发测试脚本，别靠肉眼

写一个并发脚本模拟两个收银员同时买同一件商品，是验证超卖是否被解决的最快方法。以下用 Python 的concurrent.futures模拟 10 个并发请求：

import concurrent.futures import requests barcode = "6901234567890" payload = { "items": [{"barcode": barcode, "quantity": 1}], "member_phone": None, } def checkout_once(i): resp = requests.post("http://localhost:8000/checkout", json=payload) return resp.status_code, resp.json() with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor: futures = [executor.submit(checkout_once, i) for i in range(10)] for f in concurrent.futures.as_completed(futures): print(f.result())

逻辑说明：正常情况应该有一部分请求成功、一部分返回"库存不足"。如果全部成功且最终库存为负，说明并发控制失效。这个脚本不需要很复杂，关键是卡在"校验"这一步能够并发触发，才能在实测中观察到锁的效果。实操时可以先跑不加锁的版本，再跑加锁版本，对比结果，比空口讲理论更有说服力。

5.4 在代码里记录一份 troubleshooting 自查表

给自己的项目写一份简短的排错清单，后面写文档时直接照抄。我常用的检查顺序是：

1. 确认数据库连接串指向的文件/库是否正确 2. 确认 SQLAlchemy 模型字段和数据库表结构是否一致（改过模型后要迁移） 3. 确认收银接口是否在同一个 db session 里完成所有写操作 4. 确认 stock_log 是否每笔变动都有记录 5. 用并发脚本压一遍，查看最终库存是否和流水对得上

第 4 条在答辩时是加分项。老师提问"如何保证数据一致"，你能答出"所有库存变动都写日志，并且与库存修改在同一个事务里提交"，基本就过关了。

6. 把库存流水调成一把万能尺：从日结对账到报表回溯的实操技巧

6.1 用 stock_log 做日结盘点，不需要另外开发

日结是超市每天必须做的动作。对账的核心公式是：昨日库存 + 今日入库 - 今日销售 = 今日库存。用 stock_log 直接查当天的所有变动，即可验证账面库存与实际盘点是否一致：

SELECT product_id, SUM(CASE WHEN change_type = 1 THEN quantity ELSE 0 END) AS in_qty, SUM(CASE WHEN change_type = 2 THEN quantity ELSE 0 END) AS out_qty, SUM(CASE WHEN change_type IN (1, 4) THEN quantity WHEN change_type IN (2, 3, 5) THEN -quantity ELSE 0 END) AS net_change FROM stock_log WHERE created_at >= :today_start AND created_at <= :today_end GROUP BY product_id;

逻辑说明：change_type为 1（进货）和 4（退货）时数量为正；2（销售）和 5（报损）时是出库，应取负。net_change可以直接加到昨日库存上得出今日账面库存，如果有盘点调整（change_type=3）则代表手动修正，实际盘点数量以调整后的为准。这段 SQL 是"详细文档"里最适合放进论文附录的东西。

6.2 仅用 Pyecharts 做一张图表，放到后台首页当门面

可视化没必要引入太重的前端框架。如果你是后端为主、前端只写简单 H5 页面，推荐用 Pyecharts 生成图表并输出 HTML 片段。以下是销售额趋势的柱状图和折线图组合：

from pyecharts import options as opts from pyecharts.charts import Bar, Line bar = Bar() bar.add_xaxis(dates) bar.add_yaxis("销售额", sales_amounts, label_opts=opts.LabelOpts(is_show=False)) line = Line() line.add_xaxis(dates) line.add_yaxis("订单量", order_counts) bar.overlap(line) html_content = bar.render_embed()

render_embed()返回一个可插入到页面模板的 HTML 片段，不用单独部署图表服务。把这一段放在后台首页，再附上低库存角标，整个系统的"结果呈现"就完整了。生成图时注意把日期做成连续时间轴，缺哪天补 0，否则图表中间断档很难看。

6.3 导出 CSV 报表的技巧：别用 print 而是用标准库 csv

财务或店长通常要求导出 Excel。用csv标准库在 FastAPI 中返回一个可下载文件，最简单且无需安装额外依赖：

from fastapi.responses import StreamingResponse import csv import io def export_sales(db: Session): buffer = io.StringIO() writer = csv.writer(buffer) writer.writerow(["日期", "订单数", "销售额"]) rows = get_daily_sales(db) # 复用前面的聚合查询 for row in rows: writer.writerow([row["day"], row["order_count"], row["sales_amount"]]) buffer.seek(0) return StreamingResponse( buffer, media_type="text/csv", headers={"Content-Disposition": "attachment; filename=sales.csv"}, )

StreamingResponse 直接把 CSV 内容作为响应体返回，浏览器会触发下载。用StringIO而非临时文件，避免磁盘清理问题。这一步单独拿出来做成通用导出工具，后面供应商对账、采购明细导出都复用同一个函数。

6.4 最后值得单列的技巧：定期备份 SQLite 文件

SQLite 数据库复制文件即可备份。最稳妥的方式是定时把.db文件复制一份并带上日期后缀，不需要像 MySQL 那样再导 SQL 文件。如果你用 MySQL 部署，则在接口层写一个简单的备份调度，每天凌晨执行一次mysqldump。课设答辩时演示一下备份功能，属于典型的"加分项"。整理文档时，把备份命令也写进说明，这样整套系统的完整性会明显高于平均水准。

本文还有配套的精品资源，点击获取