简介面向Flask学习者的购物平台API完整项目基于Python语言开发定位为后端接口设计与二次开发示例适合快速入门电商服务端开发。项目完整覆盖用户、商品、购物车、订单、支付及分类管理等核心模块配套数据库迁移脚本与简单HTML前端页面可本地运行调试帮助读者理解API路由、模型映射、业务分层及认证鉴权等关键机制。压缩包内共53个文件以36个Python脚本为主体搭配XML/INI环境配置、PEM支付证书、MAKO模板及Markdown说明文档包体仅85KB结构清晰适合按模块逐步查阅。目前已有576人学习下载属于轻量但覆盖电商主流程的小型后端案例。通过研读源码可掌握统一响应封装、错误码管理、支付宝密钥接入、数据库迁移脚本编写等实践要点对后续独立开发同类API接口具有直接的借鉴价值。1. 这个购物平台 API 源码到底解决什么问题购物平台的业务边界比大多数个人项目要清晰商品要能分页查、能看详情用户要能注册登录购物车要能加减订单要能在事务里扣库存。这些接口如果散落在一个几百行的app.py里前期开发快后期加一个满减活动就要动老代码。基于 Python Flask 框架的购物平台 API 开发源码核心不是“用 Flask 写接口”这件事本身而是把一套可维护的项目结构、统一的响应格式、可验证的鉴权和可追踪的数据库操作沉淀成模板。它适合两类人一类是刚学完 Flask 基础、想看到“真实项目该有的目录长什么样”的开发者另一类是给公司做内部商城或二手交易平台原型需要从零搭 API 但不想重复踩认证、分页、事务坑的工程师。这篇文章按“工程结构 → 核心接口 → 鉴权与校验 → 性能与安全”这条线展开代码可以直接抄但更重要的是理解每个参数为什么这么设。2. 用应用工厂与蓝图把购物平台拆成可维护的 Flask 工程2.1 为什么单文件app.py不适合购物平台 API购物平台至少涉及商品、用户、购物车、订单四类资源外加通用的认证、错误处理、数据库初始化。如果全部写在一个文件里路由装饰器之间靠注释分隔app.config里混着数据库连接串和 SECRET_KEY迁移到生产环境时你会发现自己根本不敢动任何一行。常见做法是采用应用工厂模式把创建Flask实例的动作封装进一个create_app函数扩展对象不在模块导入时初始化而是在函数内部通过init_app绑定。这样同一个代码库既能flask run跑开发服务器也能用gunicorn app:create_app()启动生产进程测试时还能为每个用例创建独立的 app 实例。2.2 最小可运行的项目目录与create_app先看目录结构这是源码里最先要确认的东西shop_api/ ├── app/ │ ├── __init__.py # create_app 所在位置 │ ├── extensions.py # db, jwt 等扩展对象 │ ├── models/ # SQLAlchemy 模型 │ │ ├── __init__.py │ │ ├── user.py │ │ ├── product.py │ │ ├── cart.py │ │ └── order.py │ ├── api/ # 蓝图 │ │ ├── __init__.py │ │ ├── auth.py │ │ ├── products.py │ │ ├── cart.py │ │ └── orders.py │ ├── utils/ # 装饰器、序列化工具 │ │ ├── __init__.py │ │ ├── decorators.py │ │ ├── responses.py │ │ └── validators.py │ └── config.py # 配置类 ├── migrations/ # Flask-Migrate 迁移脚本 ├── requirements.txt └── run.pyextensions.py单独拆出来是为了避免app和models循环导入# app/extensions.py from flask_sqlalchemy import SQLAlchemy from flask_jwt_extended import JWTManager db SQLAlchemy() jwt JWTManager()create_app的骨架如下# app/__init__.py from flask import Flask from .config import Config from .extensions import db, jwt def create_app(config_classConfig): app Flask(__name__) app.config.from_object(config_class) db.init_app(app) jwt.init_app(app) from .api.auth import auth_bp from .api.products import products_bp from .api.cart import cart_bp from .api.orders import orders_bp app.register_blueprint(auth_bp, url_prefix/api/v1/auth) app.register_blueprint(products_bp, url_prefix/api/v1/products) app.register_blueprint(cart_bp, url_prefix/api/v1/cart) app.register_blueprint(orders_bp, url_prefix/api/v1/orders) return app这里重点解释三个设计决策。第一url_prefix统一了接口版本前缀后续如果购物平台要接第三方的商品同步、促销引擎可以新增/api/v2而不破坏现有调用方。第二蓝图内部的路由不要写绝对路径。比如cart_bp里写cart_bp.route(/items)最终映射到/api/v1/cart/items这样改动前缀时不需要翻每个视图函数。第三db和jwt放在模块顶层但只在init_app时绑定保证了测试中db.drop_all()后重新建表时不会因为扩展已经绑定了旧的 app 而报错。2.3 商品与订单模型金额为什么用 Numeric 不用 Float购物平台的模型设计决定了后边所有接口的写法。商品模型里最容易犯的错是用Float存价格二进制浮点数在比较0.1 0.2 0.3时会得到False金额计算累积误差会在订单结算时暴露。SQLAlchemy 里用Numeric(10, 2)数据库底层存的是定点数# app/models/product.py from ..extensions import db from decimal import Decimal class Product(db.Model): __tablename__ products id db.Column(db.Integer, primary_keyTrue) name db.Column(db.String(128), nullableFalse, indexTrue) category db.Column(db.String(32), indexTrue) price db.Column(db.Numeric(10, 2), nullableFalse, defaultDecimal(0.00)) stock db.Column(db.Integer, nullableFalse, default0) version db.Column(db.Integer, nullableFalse, default1) def to_dict(self): return { id: self.id, name: self.name, category: self.category, price: str(self.price), stock: self.stock, }注意to_dict里str(self.price)的写法。Decimal直接放进jsonify会报TypeError转成字符串后前端拿到的是199.00而不是199.0避免 JavaScript 浮点精度问题。订单模型需要同时包含订单主表和订单明细表主表记录总金额、状态、创建时间明细表记录每个商品的快照价格# app/models/order.py class Order(db.Model): __tablename__ orders id db.Column(db.Integer, primary_keyTrue) user_id db.Column(db.Integer, db.ForeignKey(users.id), nullableFalse, indexTrue) total_amount db.Column(db.Numeric(10, 2), nullableFalse) status db.Column(db.String(16), nullableFalse, defaultPENDING) created_at db.Column(db.DateTime, server_defaultdb.func.now()) class OrderItem(db.Model): __tablename__ order_items id db.Column(db.Integer, primary_keyTrue) order_id db.Column(db.Integer, db.ForeignKey(orders.id), nullableFalse, indexTrue) product_id db.Column(db.Integer, nullableFalse) product_name db.Column(db.String(128), nullableFalse) price db.Column(db.Numeric(10, 2), nullableFalse) quantity db.Column(db.Integer, nullableFalse)下单时商品改价不影响历史订单这是OrderItem存快照字段的根本原因。每次查询都要join商品表拿名字和价格的做法在运营改价后会直接改写历史订单的展示金额这是购物平台源码里最常见的隐藏 bug。3. 购物平台商品与订单接口Flask 路由层怎么落 SQL3.1 商品分页查询page 与 page_size 的参数边界商品列表是所有购物平台访问量最高的接口之一直接Product.query.all()返回全部记录属于典型反模式。分页参数需要做边界收敛否则恶意调用方传page_size100000会把数据库连接池打满# app/api/products.py from flask import Blueprint, request, jsonify from ..models.product import Product from ..utils.responses import ok from ..utils.validators import clamp products_bp Blueprint(products, __name__) products_bp.route() def list_products(): page clamp(request.args.get(page, 1, typeint), 1, 10000) page_size clamp(request.args.get(page_size, 20, typeint), 1, 100) pagination Product.query \ .order_by(Product.id.asc()) \ .paginate(pagepage, per_pagepage_size, error_outFalse) items [p.to_dict() for p in pagination.items] return ok({ items: items, total: pagination.total, page: page, page_size: page_size, has_next: pagination.has_next, })clamp是自定义工具函数作用是把输入限制在合法区间内# app/utils/validators.py def clamp(value, lower, upper): if value is None: return lower return max(lower, min(value, upper))参数说明request.args.get(page, 1, typeint)中的typeint会在参数无法转成 int 时返回默认值 1而不是抛 400。error_outFalse让超出总页数的请求返回空列表而不是 404对客户端更友好。分页接口的响应里必须带has_next和total这是移动端下拉加载的判定依据只返回 items 会导致前端无法判断是否还有下一页。3.2 购物车加购先查库存再写记录购物车接口需要登录态这里用jwt_required()装饰器这是 Flask-JWT-Extended 提供的标准能力# app/api/cart.py from flask import Blueprint, request from flask_jwt_extended import jwt_required, get_jwt_identity from ..extensions import db from ..models.product import Product from ..models.cart import CartItem from ..utils.responses import ok, fail cart_bp Blueprint(cart, __name__) cart_bp.route(/items, methods[POST]) jwt_required() def add_to_cart(): user_id get_jwt_identity() data request.get_json(silentTrue) or {} product_id data.get(product_id) quantity data.get(quantity, 1) if not product_id or not isinstance(quantity, int) or quantity 0: return fail(product_id 和 quantity 必填quantity 必须为正整数, code400) product db.session.get(Product, product_id) if product is None: return fail(商品不存在, code404) if product.stock quantity: return fail(f库存不足当前剩余 {product.stock}, code409) cart_item CartItem.query \ .filter_by(user_iduser_id, product_idproduct_id) \ .first() if cart_item: cart_item.quantity quantity else: cart_item CartItem(user_iduser_id, product_idproduct_id, quantityquantity) db.session.add(cart_item) db.session.commit() return ok({cart_item_id: cart_item.id, quantity: cart_item.quantity})逻辑说明里要强调两个容易被忽略的点。第一request.get_json(silentTrue)返回空字典而不是抛异常这样前端传了空 body 或者在Content-Type缺失时接口返回的是业务错误码 400而不是 Flask 默认的 HTML 错误页——这是给小程序或 App 端用的 API 必须处理的差异。第二db.session.get(Product, product_id)是 SQLAlchemy 2.0 推荐的按主键查询写法替代老版本Product.query.get(product_id)。前者不会触发DeprecationWarning而且在开启query_expression时行为更可预期。3.3 下单接口事务内用行级锁保证不超卖下单是购物平台 API 里最需要认真对待的一个接口。两个用户同时购买同一商品最后一件时如果先读库存再减库存两次读到的都是 1就会产生超卖。解决思路是在事务内对商品行加排他锁# app/api/orders.py from flask import Blueprint, request from flask_jwt_extended import jwt_required, get_jwt_identity from sqlalchemy import func from ..extensions import db from ..models.product import Product from ..models.cart import CartItem from ..models.order import Order, OrderItem from ..utils.responses import ok, fail from decimal import Decimal orders_bp Blueprint(orders, __name__) orders_bp.route(, methods[POST]) jwt_required() def create_order(): user_id get_jwt_identity() data request.get_json(silentTrue) or {} cart_ids data.get(cart_item_ids, []) if not cart_ids: return fail(请选择要结算的购物车项, code400) # 使用 with_for_update 对购物车项加行级排他锁 cart_items CartItem.query \ .filter(CartItem.id.in_(cart_ids), CartItem.user_id user_id) \ .with_for_update() \ .all() if len(cart_items) ! len(set(cart_ids)): return fail(部分购物车项不存在, code404) total Decimal(0.00) order_items [] for item in cart_items: # 对商品行加锁防止并发改库存 product Product.query \ .filter(Product.id item.product_id) \ .with_for_update() \ .first() if product is None: db.session.rollback() return fail(f商品 {item.product_id} 已下架, code410) if product.stock item.quantity: db.session.rollback() return fail(f商品 {product.name} 库存不足, code409) product.stock - item.quantity total product.price * item.quantity order_items.append(OrderItem( product_idproduct.id, product_nameproduct.name, priceproduct.price, quantityitem.quantity, )) order Order( user_iduser_id, total_amounttotal, statusPENDING, itemsorder_items, ) db.session.add(order) # 清除已结算的购物车项 for item in cart_items: db.session.delete(item) db.session.commit() return ok({order_id: order.id, total_amount: str(total)}, code201)with_for_update()对应 MySQL 的SELECT ... FOR UPDATE在事务提交前其他事务对这些行的更新会被阻塞。注意这个特性依赖数据库的 InnoDB 引擎和已开启的事务SQLite 不支持该语法源码里如果同时想支持 SQLite 测试需要加条件判断或改用乐观锁。db.session.rollback()的使用时机很关键。当库存校验失败时购物车行已经被锁住如果不回滚直接返回事务会一直持有锁直到请求结束严重时造成死锁。每个fail返回前先rollback是这类接口必须养成的习惯。4. 把令牌校验做进 Flask 钩子API 鉴权与参数边界4.1 JWT 登录流程token 里到底该放什么购物平台的登录接口一般返回两个东西access_token和refresh_token。access_token有效期短比如 30 分钟refresh_token有效期长7 天。权威做法是把user_id放进identity不要把手机号、邮箱、密码哈希放进去。JWT 是 base64 编码不是加密任何能拿到 token 的人都可以直接解码看到 payload 内容。下面是登录接口的标准写法# app/api/auth.py from flask import Blueprint, request from flask_jwt_extended import create_access_token, create_refresh_token from werkzeug.security import check_password_hash from ..models.user import User from ..utils.responses import ok, fail auth_bp Blueprint(auth, __name__) auth_bp.route(/login, methods[POST]) def login(): data request.get_json(silentTrue) or {} username data.get(username) password data.get(password) if not username or not password: return fail(用户名和密码不能为空, code400) user User.query.filter_by(usernameusername).first() if user is None or not check_password_hash(user.password_hash, password): return fail(用户名或密码错误, code401) access_token create_access_token(identitystr(user.id)) refresh_token create_refresh_token(identitystr(user.id)) return ok({ access_token: access_token, refresh_token: refresh_token, token_type: Bearer, expires_in: 1800, })identity转成字符串是必须的Flask-JWT-Extended 在旧版本中对 int 类型的 subject 支持不完整转字符串后可以规避序列化兼容问题。expires_in字段和ACCESS_EXPIRES配置对应前端根据这个值提前 5 分钟调刷新接口而不是等 token 真正过期。配置项在config.py里# app/config.py import os from datetime import timedelta class Config: SECRET_KEY os.environ.get(SECRET_KEY, dev-secret-key) SQLALCHEMY_DATABASE_URI os.environ.get( DATABASE_URL, mysqlpymysql://root:passwordlocalhost/shop_api?charsetutf8mb4 ) SQLALCHEMY_TRACK_MODIFICATIONS False SQLALCHEMY_ENGINE_OPTIONS { pool_size: 10, pool_recycle: 3600, pool_pre_ping: True, } JWT_ACCESS_TOKEN_EXPIRES timedelta(minutes30) JWT_REFRESH_TOKEN_EXPIRES timedelta(days7)pool_recycle3600是给 MySQL 用的关键参数。MySQL 默认wait_timeout是 8 小时但云数据库厂商经常改成 1 小时连接空闲过久会被服务端断开SQLAlchemy 连接池里的连接感知不到下次请求就会报Lost connection to MySQL server。pool_pre_pingTrue会在每次从池里取连接前执行SELECT 1探活失效连接自动丢弃重建这两个参数组合起来能解决绝大多数 Flask 线上偶发 500 问题。4.2 用装饰器统一参数校验避免接口间复制粘贴商品列表、加购、下单这三个真实接口每个都要校验参数类型和取值范围直接在视图函数里写if会越写越长。源码里常见的做法是把校验逻辑抽成装饰器或者在utils/validators.py里提供可复用的校验函数。推荐一种轻量做法定义expect装饰器声明参数的类型和是否必填。# app/utils/decorators.py from functools import wraps from flask import request from .responses import fail def expect(schema): def decorator(fn): wraps(fn) def wrapper(*args, **kwargs): data request.get_json(silentTrue) or {} for field, rules in schema.items(): value data.get(field) if rules.get(required) and value is None: return fail(f参数 {field} 不能为空, code400) if value is not None and type in rules: if not isinstance(value, rules[type]): return fail(f参数 {field} 类型必须为 {rules[type].__name__}, code400) if value is not None and max in rules and value rules[max]: return fail(f参数 {field} 不能大于 {rules[max]}, code400) return fn(*args, **kwargs) return wrapper return decorator用法是给加购接口挂上校验声明cart_bp.route(/items, methods[POST]) jwt_required() expect({ product_id: {required: True, type: int}, quantity: {required: False, type: int, max: 99}, }) def add_to_cart(): ...很多源码里直接用marshmallow做序列化和校验但小项目里引入全套marshmallow会让新手困惑。先用这种字典模式跑通接口多了以后自然能体会到为什么需要marshmallow的嵌套校验和Schema.load能力。4.3 统一响应与错误处理API 的“合同”要稳定购物平台 API 面向多端Web、小程序、App响应格式必须固定。统一的成功响应和错误响应结构如下# app/utils/responses.py from flask import jsonify def ok(dataNone, code0, messagesuccess, http_status200): return jsonify({ code: code, message: message, data: data, }), http_status def fail(messageerror, code-1, http_status200): return jsonify({ code: code, message: message, data: None, }), http_status注意fail的http_status默认是 200HTTP 状态码表示传输层状态业务状态码表示业务结果。很多后端会把参数错误直接返回 400把未认证返回 401这本身没错但要在团队内约定清楚前端拦截器是统一处理code ! 0还是处理response.status ! 200。两种风格各有拥趸最怕的是接口 A 用 HTTP 状态码表达业务错误接口 B 用 body 里的 code 表达前端拦截器就会写出大量特判。再注册全局异常处理器捕获未处理的异常并记录日志避免 Flask 默认返回 HTML 错误页# app/__init__.py 的 create_app 内追加 from .utils.responses import fail app.errorhandler(404) def handle_404(e): return fail(资源不存在, code404, http_status404) app.errorhandler(500) def handle_500(e): app.logger.exception(Unhandled error) return fail(服务器内部错误, code500, http_status500)真发生五类错误时app.logger.exception会把堆栈打到日志文件而返回给客户端的是不含内部细节的通用提示。绝对不要把堆栈字符串拼进响应体返回给前端调试信息一旦暴露在公网 API 里相当于给攻击者送了一张系统内部结构图。5. 上线前必调的性能与安全旋钮从能跑到扛得住5.1 开启 gzip 压缩商品列表响应体立减 70%商品列表接口返回的 JSON 里包含大量中文字段名和商品描述不压缩时一个 50 条数据的列表轻松超过 200KB移动端弱网场景下体验极差。flask-compress是 Flask 生态里最省事的方案pip install flask-compress# app/__init__.py from flask_compress import Compress compress Compress() def create_app(config_classConfig): app Flask(__name__) app.config.from_object(config_class) compress.init_app(app)它的默认行为是对所有大于 500 字节的响应做 gzip无需每个路由手动判断。app.config[COMPRESS_MIN_SIZE]可以调成 1024避免对小响应做无意义的压缩。生产环境前端如果配置了 CDN需要在 CDN 层关闭压缩否则会双重压缩浪费 CPU。这个旋钮在压力测试里通常能把商品列表接口的 P95 延迟从 200ms 降到 80ms 以下。5.2 登录接口限流挡住密码爆破购物平台的登录接口是攻击者最爱的目标。无限制的登录接口配合撞库字典能在几小时内发起数十万次请求。flask-limiter提供了开箱即用的限流能力pip install flask-limiter# app/extensions.py from flask_limiter import Limiter from flask_limiter.util import get_remote_address limiter Limiter(key_funcget_remote_address)# app/__init__.py 中初始化后在 auth_bp 上挂载限制 from .extensions import limiter def create_app(config_classConfig): ... limiter.init_app(app) # app/api/auth.py auth_bp.route(/login, methods[POST]) limiter.limit(5 per minute, methods[POST], key_funclambda: request.get_json(silentTrue).get(username, request.remote_addr)) def login(): ...这段代码的关键在key_func。默认按 IP 限流但在公司出口 NAT 下所有员工共享一个 IP限流会误伤。按username限流更合理同时加上登录失败的标记比如连续失败 5 次锁定 15 分钟。限流返回的 429 状态码要单独在错误处理器里转成fail格式。单实例部署时flask-limiter 默认把计数存在内存中重启即清零生产环境需要用 Redis 存储即storage_uriredis://localhost:6379/0否则多 worker 下计数各自独立限流形同虚设。5.3 慢查询日志与连接池监控最后一个可复现的排错技巧购物平台 API 上线后最常见的性能问题是 MySQL 慢查询。在 SQLAlchemy 层开启 echo 只适合开发环境生产环境正确的做法是让 MySQL 自己记录慢查询SET GLOBAL slow_query_log ON; SET GLOBAL long_query_time 1; SET GLOBAL slow_query_log_file /var/log/mysql/slow.log;long_query_time 1表示超过 1 秒的 SQL 都会被记录。用mysqldumpslow -s c -t 20 /var/log/mysql/slow.log按执行次数排序查看前 20 条能快速定位到没用上索引的查询。另外确保Product.category和Order.user_id这类经常出现在WHERE和ORDER BY里的字段已建索引——SQLAlchemy 模型里indexTrue已经声明了但老项目如果从别的源码迁移过来一定要用db.session.execute(text(SHOW INDEX FROM products))核对实际库里的索引模型定义和数据库不同步是常事。最后把SQLALCHEMY_ENGINE_OPTIONS里的pool_size从 10 调到max_connections - 10的合理值比如 30并观察SHOW STATUS LIKE Threads_connected的曲线这是判断连接池参数是否合理的直接依据。本文还有配套的精品资源点击获取