尧图网站设计 尧图网站设计YAOTU DESIGN
ARTICLE DETAIL

资讯详情

深耕网站设计与一线实操的经验洞察。

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

用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 张表表名核心字段作用productid, barcode, name, category_id, price, cost_price, stock, safety_stock商品档案和实时库存categoryid, name, parent_id商品分类方便按类目统计supplierid, name, contact, phone, address供应商信息进货时关联purchase_orderid, supplier_id, total_amount, status, created_at进货单主表purchase_itemid, order_id, product_id, quantity, price, amount进货单明细一单多品sale_orderid, order_no, member_id, total_amount, cashier_id, created_at销售单主表sale_itemid, order_id, product_id, quantity, price, amount销售明细memberid, name, phone, points, balance会员用于折扣和积分stock_logid, product_id, change_type, quantity, before_stock, after_stock, created_at库存变动流水初学者最常见的错误是只建 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_keyTrue) supplier_id: Mapped[int] mapped_column(ForeignKey(supplier.id)) order_no: Mapped[str] mapped_column(String(32), uniqueTrue) total_amount: Mapped[float] mapped_column(Numeric(10, 2)) status: Mapped[int] mapped_column(Integer, default0) # 0待入库, 1已入库, 2已作废 created_at: Mapped[datetime] mapped_column(DateTime, server_defaultfunc.now()) items: Mapped[list[PurchaseItem]] relationship( back_populatesorder, cascadeall, delete-orphan ) class PurchaseItem(Base): __tablename__ purchase_item id: Mapped[int] mapped_column(Integer, primary_keyTrue) 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_populatesitems) product: Mapped[Product] relationship()逻辑说明Numeric(10, 2)用于金额字段避免 Float 的二进制精度问题这在超市结算场景中会直接影响对账结果。cascadeall, delete-orphan表示删除进货单时同时删除明细但前提是进货单处于未入库状态这个校验写业务层而不是数据库层。relationship()会自动处理外键关联查询进货单时通过order.items即可拿到全部明细无需手写 join。参数说明进货单状态status我习惯用整数枚举而不是字符串因为后续如果用 FastAPI 做接口前端下拉框传数字比传中文更不容易出错入库操作时通过事务同时修改purchase_order.status和商品库存。2.3 库存表为什么不单独设计而是挂在 product 上有些管理系统会把库存单独拆成 inventory 表理由是商品档案和库存数量生命周期不同。但对超市这种小规模场景单独拆表会让代码多一层 join而且并发扣库存时还得处理两行数据的锁。我一般直接把stock字段放在 product 表上同时用 stock_log 表记录每一次变动这样既能实时查库存又能回溯每笔变动。入库、销售、盘点调整都写 stock_logchange_type 用数字约定change_type含义1进货入库2销售出库3盘点调整4退货入库5报损出库这样设置的收益是搜索超市管理系统 库存流水这个关键词的人最终都会意识到一条变动手写一条日志比直接改库存数字更可靠。后面所有报表统计也可以直接基于 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(bindengine, autoflushFalse, autocommitFalse)逻辑说明autocommitFalse是必须的它要求你在业务代码里显式调用db.commit()这样才能把多个写操作包进一个事务。autoflushFalse意思是查询前不自动把待提交的改动刷到数据库避免在复杂业务逻辑中产生预期之外的 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_onlyTrue是判断是否需要补货的接口超市管理里这个功能高频使用。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_code404, detail进货单不存在) if order.status ! 0: raise HTTPException(status_code400, 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_iditem.product_id, change_type1, quantityitem.quantity, before_stockold_stock, after_stockproduct.stock, )) order.status 1 db.commit() except Exception: db.rollback() raise HTTPException(status_code500, 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(barcodeitem.barcode).first() if not product: raise HTTPException(status_code404, detailf条码 {item.barcode} 不存在) if product.stock item.quantity: raise HTTPException(status_code400, detailf{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(phonepayload.member_phone).first() if member: discount_rate 0.95 # 会员默认 95 折 actual_amount round(total_amount * discount_rate, 2) # 4. 创建销售单和明细 order SaleOrder( order_noorder_no, total_amountactual_amount, member_idmember.id if member else None, ) db.add(order) db.flush() # 获取自增主键 for data in sale_items: product data[product] db.add(SaleItem( order_idorder.id, product_idproduct.id, quantitydata[quantity], priceproduct.price, amountdata[amount], )) # 5. 扣库存 old_stock product.stock product.stock - data[quantity] db.add(StockLog( product_idproduct.id, change_type2, quantitydata[quantity], before_stockold_stock, after_stockproduct.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 fSO{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, jsonpayload) return resp.status_code, resp.json() with concurrent.futures.ThreadPoolExecutor(max_workers10) 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_type3则代表手动修正实际盘点数量以调整后的为准。这段 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_optsopts.LabelOpts(is_showFalse)) 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_typetext/csv, headers{Content-Disposition: attachment; filenamesales.csv}, )StreamingResponse 直接把 CSV 内容作为响应体返回浏览器会触发下载。用StringIO而非临时文件避免磁盘清理问题。这一步单独拿出来做成通用导出工具后面供应商对账、采购明细导出都复用同一个函数。6.4 最后值得单列的技巧定期备份 SQLite 文件SQLite 数据库复制文件即可备份。最稳妥的方式是定时把.db文件复制一份并带上日期后缀不需要像 MySQL 那样再导 SQL 文件。如果你用 MySQL 部署则在接口层写一个简单的备份调度每天凌晨执行一次mysqldump。课设答辩时演示一下备份功能属于典型的加分项。整理文档时把备份命令也写进说明这样整套系统的完整性会明显高于平均水准。本文还有配套的精品资源点击获取
返回列表