简介这是一套基于Flask与MySQL实现的完整电商购物商城项目源码面向Python Web开发初学者及全栈入门者帮助掌握轻量级框架下的用户认证、商品管理、订单流程与后台运维等核心电商功能。资源共246个文件包含40个HTML模板页、39个PNG/JPG商品与界面图片、10个核心Python业务逻辑文件、3个CSS样式表及2个JavaScript交互脚本辅以SQL数据库脚本与字体图标资源整体压缩包仅12.27MB结构清晰、即开即用。已有478人学习下载项目覆盖注册登录、商品分类与搜索、购物车、订单状态跟踪、用户中心、收藏夹、后台多模块管理含轮播图、评价、账户权限等10大功能模块并集成WTF-Forms表单验证、SQLAlchemy ORM、AJAX异步交互与RESTful接口设计代码注释充分适合作为Flask实战教学案例或二次开发基础模板。1. Flask购物商城PythonMySQL不是Demo是能跑通支付、库存和用户会话的真实最小闭环很多刚学完Flask路由和SQLAlchemy基础的人一写“商城”就卡在登录态失效、商品库存扣减超卖、订单状态无法原子更新这三关。这不是代码没写完而是没意识到一个能上线的购物商城核心不在页面多炫而在HTTP无状态协议下如何用MySQL持久化业务一致性——用户加购时库存要锁住、下单时要校验余额与库存、支付回调后要更新多张表且不可回滚。本项目用纯Flask不依赖Django或FastAPI原生MySQL连接非ORM全包把用户认证、商品浏览、购物车、订单生成、库存扣减这五个模块串成一条可验证的数据流。适合已掌握Python函数定义、MySQL建表语法、Linux命令行操作的开发者目标不是教你怎么写HTML而是让你亲手把flask run启动的服务变成能承受并发下单、支持管理员后台查询、日志可追溯的生产级最小原型。2. 用FlaskMySQL构建购物商城的底层数据契约从ER图到可执行建表语句2.1 为什么不用SQLAlchemy ORM而选原生MySQL连接常见误区是认为ORM能简化开发但商城场景中大量涉及库存预占、订单状态机跳转、支付结果幂等校验这些操作需要精确控制事务边界和锁粒度。SQLAlchemy的session.commit()默认开启隐式事务而SELECT ... FOR UPDATE必须显式在同一个连接内执行否则锁会立即释放。原生mysql-connector-python或PyMySQL能直接暴露connection对象便于手动管理start_transaction()、commit()、rollback()且SQL可读性高排查死锁时能直接看到执行计划。我一般会用PyMySQL——它纯Python实现无需编译pip install PyMySQL即可比mysql-connector-python在CentOS 7上兼容性更稳。提示不要用sqlite3替代MySQL做本项目。SQLite不支持行级锁高并发下单时库存扣减必然超卖且无服务端权限管理无法模拟真实生产环境的数据库账号隔离策略。2.2 商城核心四张表的建表逻辑与字段设计依据建表不是照搬电商文档而是按业务动作反推字段。例如“购物车”表不能只存user_id, product_id, quantity必须加created_at和updated_at——因为前端加购接口需判断是否超过30分钟未操作过期自动清理“订单”表必须有status ENUM(pending,paid,shipped,completed,cancelled)而非TINYINT避免状态含义模糊导致后续扩展困难。以下为可直接执行的建表SQLMySQL 5.7-- 用户表密码用bcrypt哈希禁用明文存储 CREATE TABLE users ( id INT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(50) UNIQUE NOT NULL, password_hash VARCHAR(128) NOT NULL, email VARCHAR(100) UNIQUE, balance DECIMAL(10,2) DEFAULT 0.00, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ); -- 商品表status控制上下架stock为剩余库存非总销量 CREATE TABLE products ( id INT PRIMARY KEY AUTO_INCREMENT, name VARCHAR(100) NOT NULL, price DECIMAL(10,2) NOT NULL, stock INT NOT NULL DEFAULT 0, status ENUM(on_sale,off_sale,deleted) DEFAULT on_sale, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ); -- 购物车表user_idproduct_id联合唯一避免重复添加 CREATE TABLE carts ( id INT PRIMARY KEY AUTO_INCREMENT, user_id INT NOT NULL, product_id INT NOT NULL, quantity INT NOT NULL DEFAULT 1, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE, FOREIGN KEY (product_id) REFERENCES products(id) ON DELETE CASCADE, UNIQUE KEY user_product_unique (user_id, product_id) ); -- 订单表关键字段order_no业务单号、total_amount下单时快照价格、pay_status支付网关回调更新 CREATE TABLE orders ( id INT PRIMARY KEY AUTO_INCREMENT, order_no VARCHAR(32) UNIQUE NOT NULL, user_id INT NOT NULL, total_amount DECIMAL(10,2) NOT NULL, status ENUM(pending,paid,shipped,completed,cancelled) DEFAULT pending, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, paid_at DATETIME NULL, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE RESTRICT );2.2.1 字段设计背后的业务约束说明字段为什么这样设不这样设的风险products.stock类型为INT而非BIGINT因单个SKU日销量极少超21亿若用BIGINTMySQL在InnoDB引擎下会额外占用8字节空间百万级商品表将多占数GB磁盘orders.order_no用UUID4或时间戳随机数生成绝不用自增ID自增ID暴露订单量被爬虫批量请求/order/123可推测日单量且分布式部署时无法保证全局唯一carts.updated_at设为ON UPDATE CURRENT_TIMESTAMP前端定时轮询购物车时可用该字段判断是否需强制刷新减少无效请求2.3 初始化数据库连接池避免Flask多线程下的连接泄漏Flask默认使用单线程开发服务器但生产环境如gunicorn会启多个worker进程。若每个请求都新建MySQL连接连接数会迅速耗尽。必须用连接池管理# db.py import pymysql from pymysql.connections import Connection from contextlib import contextmanager class DatabasePool: def __init__(self, hostlocalhost, userroot, password123456, databaseshop): self.config { host: host, user: user, password: password, database: database, charset: utf8mb4, autocommit: False, # 关键关闭自动提交由业务代码控制事务 cursorclass: pymysql.cursors.DictCursor } self._pool None def get_connection(self) - Connection: if self._pool is None: from DBUtils.PooledDB import PooledDB self._pool PooledDB( creatorpymysql, maxconnections20, # 最大连接数 mincached5, # 初始化时创建5个空闲连接 maxcached10, # 最多缓存10个空闲连接 blockingTrue, # 连接池满时阻塞等待 **self.config ) return self._pool.connection() # 全局实例 db_pool DatabasePool()注意autocommitFalse是硬性要求。所有涉及库存扣减、订单创建的操作必须显式调用conn.commit()否则事务不会生效。blockingTrue防止高并发时连接池返回None导致500错误。3. 实现购物车与订单生成的核心事务链从加购到支付成功的完整SQL路径3.1 加购接口的原子性保障先查库存再插入还是先插入再校验错误做法先INSERT INTO carts再用UPDATE products SET stockstock-1扣库存。这会导致两个问题1购物车数量可超库存用户反复点击加购2库存扣减与购物车操作不在同一事务网络中断时状态不一致。正确路径是在同一个MySQL事务内完成三步①SELECT stock FROM products WHERE id%s FOR UPDATE—— 锁住该商品行② 判断stock quantity不满足则ROLLBACK并返回错误③INSERT INTO carts (...) VALUES (...)或UPDATE carts SET quantityquantity%s WHERE user_id%s AND product_id%s# cart.py from flask import request, jsonify from db import db_pool def add_to_cart(): data request.get_json() user_id data.get(user_id) product_id data.get(product_id) quantity data.get(quantity, 1) conn db_pool.get_connection() try: with conn.cursor() as cursor: # 步骤1锁定商品行防止并发修改 cursor.execute( SELECT stock FROM products WHERE id %s FOR UPDATE, (product_id,) ) product cursor.fetchone() if not product or product[stock] quantity: return jsonify({error: 库存不足}), 400 # 步骤2检查购物车是否已存在该商品 cursor.execute( SELECT id, quantity FROM carts WHERE user_id %s AND product_id %s, (user_id, product_id) ) cart_item cursor.fetchone() if cart_item: # 存在则更新数量 new_quantity cart_item[quantity] quantity cursor.execute( UPDATE carts SET quantity %s, updated_at NOW() WHERE id %s, (new_quantity, cart_item[id]) ) else: # 不存在则插入新记录 cursor.execute( INSERT INTO carts (user_id, product_id, quantity) VALUES (%s, %s, %s), (user_id, product_id, quantity) ) conn.commit() return jsonify({message: 加入购物车成功}), 200 except Exception as e: conn.rollback() return jsonify({error: str(e)}), 500 finally: conn.close()3.1.1 关键参数说明与调试技巧FOR UPDATE必须在BEGIN事务内执行否则锁无效。Flask中conn.cursor()默认开启事务无需显式START TRANSACTION。NOW()用MySQL函数而非Pythondatetime.now()避免应用服务器与数据库服务器时钟偏差导致updated_at异常。conn.rollback()捕获任何异常包括SQL语法错、唯一键冲突后必须回滚否则连接会处于未结束事务状态下次复用时抛出Lost connection to MySQL server during query。3.2 下单接口的跨表一致性如何保证库存扣减与订单创建不脱节下单不是简单地把购物车商品挪到订单表而是四步原子操作① 查询购物车所有商品及实时库存② 扣减对应商品库存UPDATE products SET stockstock-? WHERE id? AND stock?③ 插入订单主表与订单明细表本例为简化未建明细表实际应拆分④ 清空用户购物车其中第②步的AND stock?是关键——利用MySQL的影响行数判断来规避超卖若返回cursor.rowcount 0说明库存已被其他事务扣完整笔订单失败。# order.py def create_order(): data request.get_json() user_id data.get(user_id) conn db_pool.get_connection() try: with conn.cursor() as cursor: # 步骤1获取购物车商品含价格快照 cursor.execute( SELECT c.product_id, c.quantity, p.price FROM carts c JOIN products p ON c.product_id p.id WHERE c.user_id %s AND p.status on_sale , (user_id,)) cart_items cursor.fetchall() if not cart_items: return jsonify({error: 购物车为空}), 400 total_amount sum(item[price] * item[quantity] for item in cart_items) # 步骤2扣减库存带条件更新失败则rowcount为0 for item in cart_items: cursor.execute( UPDATE products SET stock stock - %s WHERE id %s AND stock %s, (item[quantity], item[product_id], item[quantity]) ) if cursor.rowcount 0: raise Exception(f商品 {item[product_id]} 库存不足) # 步骤3生成订单号时间戳6位随机数 import time, random order_no f{int(time.time())}{random.randint(100000, 999999)} # 步骤4插入订单 cursor.execute( INSERT INTO orders (order_no, user_id, total_amount) VALUES (%s, %s, %s), (order_no, user_id, total_amount) ) # 步骤5清空购物车 cursor.execute(DELETE FROM carts WHERE user_id %s, (user_id,)) conn.commit() return jsonify({ order_no: order_no, total_amount: float(total_amount) }), 201 except Exception as e: conn.rollback() return jsonify({error: str(e)}), 500 finally: conn.close()提示订单金额必须用products.price快照值计算而非下单时实时查价。否则用户加购后商家调价订单金额与购物车显示不符引发客诉。4. Flask路由与请求处理的实战配置解决Session失效、CSRF防护与JSON参数解析4.1 用户登录态管理用Flask-Login还是原生SessionFlask-Login封装了用户对象加载逻辑但商城需精细控制Session生命周期如用户修改密码后强制所有设备登出。原生session更可控# auth.py from flask import session, request, jsonify from werkzeug.security import check_password_hash import secrets def login(): data request.get_json() username data.get(username) password data.get(password) conn db_pool.get_connection() try: with conn.cursor() as cursor: cursor.execute( SELECT id, password_hash FROM users WHERE username %s, (username,) ) user cursor.fetchone() if user and check_password_hash(user[password_hash], password): # 生成随机session_id避免固定值被预测 session[user_id] user[id] session.permanent True # 启用永久会话 return jsonify({message: 登录成功}), 200 else: return jsonify({error: 用户名或密码错误}), 401 finally: conn.close() def logout(): session.clear() # 彻底清除所有session数据 return jsonify({message: 已退出登录}), 2004.1.1 Session安全配置要点在app.py中必须设置app.secret_key secrets.token_hex(16) # 随机密钥重启后旧session失效 app.config[SESSION_COOKIE_SECURE] True # 仅HTTPS传输生产环境 app.config[SESSION_COOKIE_HTTPONLY] True # 禁止JS读取防XSS窃取 app.config[PERMANENT_SESSION_LIFETIME] timedelta(hours24) # 24小时过期注意session.permanent True必须配合PERMANENT_SESSION_LIFETIME否则默认有效期为浏览器会话周期关闭浏览器即失效不符合商城“7天免登录”需求。4.2 防御CSRF攻击为什么GET接口不需要CSRF token而POST需要CSRF跨站请求伪造本质是利用浏览器自动携带Cookie的特性诱导用户在不知情下提交恶意请求。GET请求是幂等的如/product/123只读即使被伪造也无危害但POST /order会创建订单必须防护。Flask内置WTF-CSRF可自动注入token但商城通常用前后端分离架构Vue/React需手动传递# 在登录成功后返回CSRF token app.route(/api/csrf-token, methods[GET]) def get_csrf_token(): if user_id not in session: return jsonify({error: 未登录}), 401 token secrets.token_urlsafe(32) session[csrf_token] token return jsonify({csrf_token: token})前端在每次POST请求头中带上X-CSRF-Token: token后端校验app.before_request def validate_csrf(): if request.method in [POST, PUT, DELETE]: token request.headers.get(X-CSRF-Token) if not token or token ! session.get(csrf_token): return jsonify({error: CSRF token无效}), 4034.3 JSON参数解析的健壮性处理避免request.get_json()返回Nonerequest.get_json()在Content-Type非application/json时返回None但前端可能误发text/plain。必须加fallbackdef get_json_data(): # 优先尝试JSON解析 data request.get_json() if data is not None: return data # 兼容application/x-www-form-urlencoded if request.form: return {k: v for k, v in request.form.items()} # 兼容query string如GET请求带参数 if request.args: return {k: v for k, v in request.args.items()} return {}然后在所有接口开头调用data get_json_data() if not data: return jsonify({error: 请求体为空或格式错误}), 4005. 生产环境部署与性能验证用ab压测库存扣减接口并定位慢查询5.1 用Apache Bench模拟高并发下单验证库存扣减是否线程安全本地测试不能只跑单请求必须用abApache Bench制造并发压力。假设你的Flask服务运行在http://localhost:5000对加购接口发起100个并发、总共1000次请求ab -n 1000 -c 100 -p cart_data.json -T application/json http://localhost:5000/api/cart/add其中cart_data.json内容为{user_id: 1, product_id: 1, quantity: 1}观察输出关键指标Requests per second若低于50说明数据库连接池或SQL效率有问题Failed requests非零值表示超卖发生库存扣减失败Time per request (mean)若超过500ms需检查索引提示-c 100代表100个并发连接必须小于MySQL的max_connections默认151否则ab会报socket: Too many open files。5.2 定位慢查询开启MySQL慢日志并分析执行计划在MySQL配置文件my.cnf中启用慢日志[mysqld] slow_query_log ON slow_query_log_file /var/log/mysql/mysql-slow.log long_query_time 0.1 # 超过100ms的查询记入日志 log_queries_not_using_indexes ON重启MySQL后执行压测再用mysqldumpslow分析mysqldumpslow -s t -t 10 /var/log/mysql/mysql-slow.log重点关注SELECT ... FOR UPDATE未命中索引的查询。例如EXPLAIN SELECT stock FROM products WHERE id 123 FOR UPDATE;若type列为ALL全表扫描说明缺少主键索引——但id已是主键此时问题可能是FOR UPDATE在非事务块中执行导致MySQL无法使用索引优化锁范围。5.3 三个必加的MySQL索引提升商城查询性能表名字段索引类型作用carts(user_id, product_id)联合唯一索引加购时快速判断是否已存在避免全表扫描productsstatus普通索引WHERE statuson_sale过滤上架商品避免查全部商品orders(user_id, created_at)联合索引用户查看历史订单时按时间倒序覆盖索引避免回表创建语句ALTER TABLE carts ADD UNIQUE INDEX idx_user_product (user_id, product_id); ALTER TABLE products ADD INDEX idx_status (status); ALTER TABLE orders ADD INDEX idx_user_time (user_id, created_at);验证索引是否生效EXPLAIN SELECT * FROM orders WHERE user_id 1 ORDER BY created_at DESC LIMIT 10;若Extra列出现Using index说明走覆盖索引若为Using filesort则索引未被充分利用。5.4 用Python脚本验证库存扣减的最终一致性写一个独立脚本模拟100个用户同时抢购10件商品验证最终库存是否为0# test_stock_consistency.py import threading import requests import time def buy_item(user_id): url http://localhost:5000/api/cart/add data {user_id: user_id, product_id: 1, quantity: 1} try: requests.post(url, jsondata, timeout5) except: pass threads [] for i in range(100): t threading.Thread(targetbuy_item, args(i,)) threads.append(t) t.start() for t in threads: t.join() # 查看最终库存 conn db_pool.get_connection() with conn.cursor() as cursor: cursor.execute(SELECT stock FROM products WHERE id 1) stock cursor.fetchone()[stock] print(f最终库存: {stock}) # 应为0 conn.close()若输出最终库存: -5说明存在超卖需检查FOR UPDATE是否被正确使用。本文还有配套的精品资源点击获取