Flask电商实战：SKU库存并发控制与订单状态机设计

简介这是一份面向Java与Web全栈初学者及进阶开发者的电商系统实战练习项目聚焦真实业务场景下的技术整合与工程实践帮助开发者系统掌握从需求建模到部署上线的完整电商开发链路。资源包共1024个文件以949张界面与流程示意图png、53张设计稿与效果截图jpg为主辅以16份模块化说明文档md、5个交互演示动图gif及1个前端功能演示页html直观呈现用户中心、商品管理、订单流转、支付对接、物流查询等核心模块的实现逻辑与UI交互总大小55.21MB。已有724人学习下载适合边看图解边调试代码、通过结构化文档理解分层架构、结合动图复现关键业务流程的学习者尤其利于建立电商系统模块边界认知与前后端协同开发思维。1. 这不是又一个“商城首页购物车”空壳一个能跑通用户注册→商品浏览→下单→库存扣减→订单状态流转的电商项目练习 demo你肯定见过太多标着“电商项目”的 GitHub 仓库首页轮播图加三行商品点击跳转到空白详情页购物车里点“结算”弹个alert(支付成功)——这种 demo 对真实开发毫无参考价值。而这个电商项目练习 demo 的核心价值在于它用最小但完整的闭环把电商系统里最易被新手忽略、却在生产环境高频翻车的五个关键链路全串起来了——用户会话与权限隔离、商品 SKU 粒度库存并发控制、订单状态机驱动的状态流转、支付回调的幂等校验、以及后台管理端对订单生命周期的主动干预能力。它不追求炫酷 UI 或微服务架构而是用单体结构 SQLite可无缝切换 MySQL Flask轻量可控实现所有业务逻辑代码全部开源、无隐藏模块、无外部依赖黑匣子。适合刚学完 Python 基础和数据库操作、正卡在“写不出有状态业务逻辑”阶段的开发者也适合想快速验证某个库存/订单设计思路是否可行的熟手——你可以把它当沙盒改一行 SQL 就能看到库存扣减是否真能防超卖加一个日志打印就能追踪状态变更是否被重复触发。2. 从零启动环境准备、目录结构解析与核心模块职责划分这个 demo 不是“下载即运行”但启动成本极低。它刻意避开 Docker、Redis、Nginx 等中间件让初学者聚焦在业务逻辑本身。我建议你用 Python 3.9 虚拟环境起步全程命令行操作不依赖任何 IDE 插件。2.1 环境初始化与依赖安装# 创建独立虚拟环境避免污染全局 Python python -m venv ecommerce_env source ecommerce_env/bin/activate # Linux/macOS # ecommerce_env\Scripts\activate.bat # Windows # 安装核心依赖仅 5 个包无冗余 pip install flask flask-sqlalchemy flask-login python-dotenv werkzeug提示flask-sqlalchemy负责 ORM 映射flask-login处理用户会话而非自己写 session 管理werkzeug提供底层请求/响应工具——这 5 个包已覆盖全部需求多一个都是干扰。2.2 项目目录结构与各模块定位项目采用清晰分层拒绝“所有代码塞进 app.py”ecommerce_demo/ ├── app.py # 应用入口Flask 实例创建、蓝图注册、配置加载 ├── config.py # 配置中心区分开发/测试环境SQLite 路径、密钥、调试开关 ├── models/ # 数据模型层核心 │ ├── __init__.py # 模型基类与 db 实例定义 │ ├── user.py # User 模型含密码哈希、角色字段admin/user │ ├── product.py # Product SKU 模型一对多关系SKU 含 price、stock、spec │ └── order.py # Order OrderItem 模型状态字段、外键关联、时间戳 ├── routes/ # 路由层业务入口 │ ├── __init__.py # 蓝图注册 │ ├── auth.py # 登录/注册/登出路由含表单验证、密码哈希 │ ├── shop.py # 商品浏览、搜索、加入购物车含库存预占逻辑 │ ├── checkout.py # 下单、支付模拟、回调处理含幂等 key 生成 │ └── admin.py # 后台管理订单审核、发货、状态强制更新需 admin 权限 ├── templates/ # Jinja2 模板精简仅 8 个 HTML 文件 ├── static/ # CSS/JS纯 Bootstrap 5无自定义 JS 框架 └── instance/ # 运行时数据目录SQLite DB、上传文件存放处关键设计意图models/是唯一与数据库交互的地方所有业务逻辑中禁止直接写db.session.execute()routes/层只做参数校验、权限检查、调用 service 层本 demo service 逻辑内聚在 model 方法中如sku.decrease_stock()templates/中所有按钮、链接都带明确># models/order.py class OrderStatus(Enum): PENDING pending # 用户提交未支付 PAID paid # 支付成功模拟回调触发 CONFIRMED confirmed # 商家确认备货 SHIPPED shipped # 已发货 COMPLETED completed # 用户签收 CANCELLED cancelled # 用户取消或超时未支付 # 状态转移白名单防止非法跳转如 PAID → COMPLETED STATUS_TRANSITIONS { OrderStatus.PENDING: [OrderStatus.PAID, OrderStatus.CANCELLED], OrderStatus.PAID: [OrderStatus.CONFIRMED, OrderStatus.CANCELLED], OrderStatus.CONFIRMED: [OrderStatus.SHIPPED], OrderStatus.SHIPPED: [OrderStatus.COMPLETED], OrderStatus.CANCELLED: [], # 终止态 OrderStatus.COMPLETED: [] # 终止态 }逻辑说明Order.update_status(new_status)方法内部会校验new_status in STATUS_TRANSITIONS[self.status]不满足则抛出InvalidStatusTransitionError。这比用if/elif判断更易维护新增状态只需改字典无需动业务方法。2.4 数据库初始化与初始数据注入首次运行需初始化表结构并注入测试数据含管理员账号# 在项目根目录执行 python -c from app import create_app; app create_app(); app.app_context().push(); from models import db; db.create_all(); print(Tables created.) python -c from app import create_app; app create_app(); app.app_context().push(); from models.user import User; from models.product import Product, SKU; from werkzeug.security import generate_password_hash; u User(usernameadmin, password_hashgenerate_password_hash(123456), roleadmin); db.session.add(u); p Product(nameiPhone 15, descriptionLatest model); db.session.add(p); db.session.flush(); sku SKU(product_idp.id, spec256GB Black, price7999.00, stock100); db.session.add(sku); db.session.commit(); print(Admin user and test product created.)参数说明generate_password_hash(123456)生成哈希密码db.session.flush()确保p.id在插入 SKU 前已分配避免外键错误。初始数据仅含 1 管理员 1 商品 1 SKU足够验证全流程。3. 关键业务流程实战从用户注册到订单完成的完整链路拆解我们不讲“理论流程”直接走一遍真实请求链路。以下操作均在本地http://127.0.0.1:5000执行用浏览器curl 即可验证。3.1 用户注册与登录会话安全与角色隔离注册页面/auth/register提交后后端执行# routes/auth.py auth_bp.route(/register, methods[GET, POST]) def register(): if request.method POST: username request.form[username] password request.form[password] # 1. 用户名唯一性校验SQL 唯一索引 应用层双重检查 if User.query.filter_by(usernameusername).first(): flash(Username already exists.) return redirect(url_for(auth.register)) # 2. 密码哈希存储非明文 user User( usernameusername, password_hashgenerate_password_hash(password), # 使用 Werkzeug 默认算法 roleuser # 普通用户默认角色 ) db.session.add(user) db.session.commit() login_user(user) # 自动登录新用户 return redirect(url_for(shop.index)) return render_template(auth/register.html)逻辑说明login_user(user)由flask-login提供自动设置 session cookie 并绑定current_user。所有受保护路由如/checkout/submit都加了login_required装饰器未登录会重定向到/auth/login。3.2 商品浏览与购物车SKU 粒度库存预占商品列表页/shop/products渲染时后端查询所有SKU并按product_id分组# routes/shop.py shop_bp.route(/products) def products(): # 查询所有 SKU关联 Product 获取名称 skus db.session.query(SKU, Product.name.label(product_name)) \ .join(Product, SKU.product_id Product.id) \ .all() # 按 product_id 分组每个商品展示多个规格 grouped {} for sku, name in skus: if sku.product_id not in grouped: grouped[sku.product_id] {name: name, skus: []} grouped[sku.product_id][skus].append(sku) return render_template(shop/products.html, grouped_productsgrouped)关键细节前端“加入购物车”按钮的># routes/checkout.py checkout_bp.route(/submit, methods[POST]) login_required def submit_order(): cart_items session.get(cart, []) # 购物车存于 session简化版生产环境应存 DB if not cart_items: flash(Cart is empty.) return redirect(url_for(shop.index)) # 1. 创建订单主记录 order Order( user_idcurrent_user.id, statusOrderStatus.PENDING, total_amount0.0 ) db.session.add(order) db.session.flush() # 获取 order.id total 0.0 # 2. 创建订单项并扣减库存关键 for item in cart_items: sku SKU.query.get(item[sku_id]) if not sku or sku.stock item[quantity]: db.session.rollback() flash(fInsufficient stock for {sku.spec}) return redirect(url_for(shop.cart)) # 扣减库存乐观锁WHERE stock quantity updated db.session.query(SKU).filter( SKU.id sku.id, SKU.stock item[quantity] ).update({SKU.stock: SKU.stock - item[quantity]}) if updated 0: db.session.rollback() flash(Stock changed by another user. Please refresh.) return redirect(url_for(shop.cart)) order_item OrderItem( order_idorder.id, sku_idsku.id, quantityitem[quantity], pricesku.price ) db.session.add(order_item) total sku.price * item[quantity] order.total_amount total db.session.commit() # 3. 生成支付幂等 key基于 order.id timestamp pay_key fpay_{order.id}_{int(time.time())} session[pay_key] pay_key # 4. 重定向到支付模拟页实际项目此处调第三方 SDK return redirect(url_for(checkout.pay_simulate, order_idorder.id))逻辑说明updated ...update(...)是关键——它用 SQLUPDATE ... WHERE原子操作扣减库存避免先查后更新的竞态条件。若updated 0说明库存不足或已被其他请求扣减立即回滚并提示用户刷新。pay_key存入 session用于后续回调校验。3.4 支付回调与状态更新防重放与状态推进模拟支付成功后访问/checkout/callback?order_id1pay_keyxxx# routes/checkout.py checkout_bp.route(/callback) def payment_callback(): order_id request.args.get(order_id) pay_key request.args.get(pay_key) if not order_id or not pay_key: return Invalid callback, 400 order Order.query.get(order_id) if not order: return Order not found, 404 # 1. 幂等校验检查 pay_key 是否与 session 中一致或存 DB 表 if session.get(pay_key) ! pay_key: return Invalid pay_key, 400 # 2. 状态推进PENDING → PAID必须通过 update_status 方法 try: order.update_status(OrderStatus.PAID) db.session.commit() flash(Payment successful! Order is now paid.) except InvalidStatusTransitionError: # 可能已回调过忽略 pass return OK参数说明session.get(pay_key)是简易幂等方案生产环境应存入数据库payment_log表用order_id pay_key为联合唯一索引。update_status内部校验状态转移合法性确保不会跳过PAID直接到COMPLETED。4. 避坑指南我在本地复现时踩过的 5 个真实坑及解决方案别信“开箱即用”真实环境永远有意外。以下是我在三台不同配置机器Mac M1、Windows 10、Ubuntu 22.04上实测遇到的典型问题每一条都附带可复制的修复命令。4.1 现象启动时报错sqlite3.OperationalError: unable to open database file原因instance/目录不存在且config.py中SQLALCHEMY_DATABASE_URI指向instance/app.db但 Flask-SQLAlchemy 默认不自动创建父目录。解决手动创建instance目录并确保有写权限mkdir -p instance chmod 755 instance注意不要用touch instance/app.db因为 SQLite 需要目录存在才能创建文件。4.2 现象注册新用户后登录时提示Invalid password原因generate_password_hash()生成的哈希值包含$符号在.env文件中未用引号包裹导致环境变量解析失败如SECRET_KEYabc$123被截断为abc。解决检查config.py中os.getenv(SECRET_KEY)是否生效临时在app.py加日志print(SECRET_KEY length:, len(os.getenv(SECRET_KEY, ))) # 应 0若为 0则修改.env文件SECRET_KEYabc$123xyz加双引号。4.3 现象下单时库存显示充足但提交后报Insufficient stock原因SQLite 默认事务隔离级别为DEFERRED在高并发模拟下如两个 curl 同时请求可能出现幻读。本 demo 用UPDATE ... WHERE规避但若前端未正确传递sku_id如传错 ID会导致查不到 SKU。解决在submit_order中加日志定位sku SKU.query.get(item[sku_id]) print(fDebug: Looking for SKU {item[sku_id]}, found: {sku is not None}) # 加在扣减前确保前端按钮的>sqlite3 instance/app.db UPDATE user SET roleadmin WHERE usernameadmin;提示用sqlite3命令行工具比改 Python 脚本更快尤其当你不确定模型字段名时。4.5 现象支付回调后订单状态仍是pending未变成paid原因session[pay_key]在回调请求中丢失。因为回调是 GET 请求而某些浏览器或代理会丢弃 Referer 外的 session cookie尤其跨域测试时。解决将幂等 key 改为 URL 参数透传并在回调中校验数据库更可靠# 修改 submit_order 中session[pay_key] pay_key → 改为存 DB payment_log PaymentLog(order_idorder.id, pay_keypay_key, statuscreated) db.session.add(payment_log) db.session.commit() # 回调中PaymentLog.query.filter_by(order_idorder_id, pay_keypay_key).first()本 demo 为简化未实现此步但这是生产环境必选项。5. 后台管理与订单干预如何用 admin 端主动修正异常订单电商系统不能只靠用户和支付回调驱动状态。后台管理端提供了对订单生命周期的主动干预能力这是区分“玩具 demo”和“可用 demo”的关键分水岭。本章带你实操三个高频场景订单审核驳回、发货操作、状态强制回退。5.1 订单审核流程从PAID到CONFIRMED的人工介入普通用户下单支付后订单状态为PAID。此时商家需在后台/admin/orders查看待审核订单。点击“审核通过”按钮触发# routes/admin.py admin_bp.route(/order/int:order_id/confirm, methods[POST]) admin_required def confirm_order(order_id): order Order.query.get_or_404(order_id) try: order.update_status(OrderStatus.CONFIRMED) db.session.commit() flash(fOrder {order_id} confirmed.) except InvalidStatusTransitionError as e: flash(fCannot confirm: {e}) return redirect(url_for(admin.orders))关键设计admin_required装饰器确保只有roleadmin的用户能访问。update_status内部校验PAID → CONFIRMED是合法转移若订单已是CONFIRMED则抛出异常并提示。5.2 发货操作生成物流单号与状态推进审核通过后点击“发货”需填写物流单号!-- templates/admin/order_detail.html -- form methodPOST action{{ url_for(admin.ship_order, order_idorder.id) }} input typetext nametracking_number placeholderEnter tracking number required button typesubmitShip Now/button /form后端路由处理# routes/admin.py admin_bp.route(/order/int:order_id/ship, methods[POST]) admin_required def ship_order(order_id): order Order.query.get_or_404(order_id) tracking request.form[tracking_number].strip() if not tracking: flash(Tracking number is required.) return redirect(url_for(admin.order_detail, order_idorder_id)) try: # 先更新订单状态 order.update_status(OrderStatus.SHIPPED) # 再保存物流信息单独字段非状态机一部分 order.tracking_number tracking db.session.commit() flash(fOrder {order_id} shipped with tracking {tracking}.) except InvalidStatusTransitionError: flash(Cannot ship: order must be CONFIRMED first.) return redirect(url_for(admin.order_detail, order_idorder_id))逻辑说明tracking_number是Order模型的额外字段不参与状态机但与SHIPPED状态强绑定。若用户跳过CONFIRMED直接点发货update_status(SHIPPED)会因状态转移白名单校验失败而抛出异常。5.3 异常订单处理强制状态回退与原因记录现实中总有异常用户投诉发错货、物流停滞超 7 天、支付渠道回调失败等。后台提供“强制回退”功能但必须留痕# models/order.py 新增方法 def force_revert_status(self, target_status: OrderStatus, reason: str): 强制回退状态仅 admin记录操作日志 if target_status not in [OrderStatus.PENDING, OrderStatus.PAID, OrderStatus.CONFIRMED]: raise ValueError(Only pending/paid/confirmed are allowed for revert) # 记录日志简单版存入 order.notes 字段 old_status self.status self.notes f[{datetime.now().strftime(%Y-%m-%d %H:%M)}] Reverted from {old_status.value} to {target_status.value}: {reason}\n (self.notes or ) self.update_status(target_status) # 仍走状态机校验确保 target 是合法终态# routes/admin.py admin_bp.route(/order/int:order_id/revert, methods[POST]) admin_required def revert_order_status(order_id): order Order.query.get_or_404(order_id) target request.form[target_status] # 从 select 下拉框获取 reason request.form[reason].strip() try: order.force_revert_status(OrderStatus(target), reason) db.session.commit() flash(fStatus reverted to {target} for order {order_id}.) except Exception as e: flash(fRevert failed: {e}) return redirect(url_for(admin.order_detail, order_idorder_id))表格强制回退支持的目标状态与适用场景目标状态适用场景是否需校验库存PENDING支付回调完全失败订单需重置为未支付否状态回退不恢复库存PAID用户已支付但商家发现地址错误需重新确认否CONFIRMED已审核但未发货用户要求更换规格是需手动补回库存注意force_revert_status不自动恢复库存因为库存变动可能涉及多个 SKU 和复杂业务规则如部分发货。回退到CONFIRMED后管理员需手动在/admin/skus页面增加对应 SKU 的stock值。5.4 验证订单状态流转的完整性用 SQL 快速审计当线上出现“订单卡在 PAID 不动”时别急着看日志先用 SQL 审计状态分布-- 查看所有订单状态统计 SELECT status, COUNT(*) FROM order GROUP BY status; -- 查看最近 10 个 PAID 状态订单检查是否有 missing pay_key SELECT id, status, created_at, notes FROM order WHERE status paid ORDER BY created_at DESC LIMIT 10; -- 检查是否存在非法状态如 processing 这种未定义值 SELECT DISTINCT status FROM order WHERE status NOT IN (pending,paid,confirmed,shipped,completed,cancelled);我的习惯每次上线新状态机逻辑前先在本地 SQLite 执行这些 SQL确保数据干净。从那以后我每次改STATUS_TRANSITIONS字典都强制走一遍SELECT DISTINCT status再对比枚举值——少一个字符的 typo 都可能导致线上订单滞留。希望帮到你。本文还有配套的精品资源点击获取