简介基于Python Flask框架实现的购物平台API源码面向正在学习Flask后端开发、接口设计或电商系统搭建的开发者可帮助理解从商品、用户、购物车到订单支付的一条完整业务链路。压缩包共53个文件以36个Python脚本和5个XML配置为主体另含PEM证书、txt说明、HTML/Mako页面以及数据库迁移相关文件整体仅85KB便于快速下载和本地部署。项目结构清晰model层定义商品、用户、购物车、订单等数据模型router层提供对应接口路由libs封装统一响应与错误码auth实现登录鉴权alipay相关文件展示支付对接方式templates目录作为前端页面入口alembic迁移脚本可辅助初始化数据库。目前已有576人学习下载。借助该源码可上手Flask蓝图、SQLAlchemy模型映射、鉴权中间件、支付宝公钥私钥配置及API返回格式设计适合作为课程设计或中小型电商项目的参考基础。1. 拿到Flask购物平台API源码先看什么从网上下载一个标着“Flask购物平台API”的源码包直接扔进编辑器里改代码大概率会在数据库迁移和支付宝密钥这两步上卡住。我最近拆了一个非常典型的版本压缩包里一共54个文件36个Python脚本、5个XML配置、2个PEM证书外加一套alembic迁移脚本。定位很清楚给线上购物App或小程序提供JSON接口业务链路就是“注册登录—浏览商品—加购物车—创建订单—支付宝支付”。这个源码包适合两类人一是刚把Flask基础语法过完想找一个能跑通的完整项目当作SQLAlchemy和蓝图参考的初学者二是需要在内部快速搭建电商后端Demo或者学习如何把支付宝支付、JWT鉴权、数据库迁移这些东西整合进Flask开发流程的工程师。源码里没有复杂的前端只有一个index.html和零散模板核心价值在API设计本身而不是页面渲染。2. 工程结构与数据模型从目录拆解一个Flask业务系统拿到源码以后我先看文件树。这个包的目录结构在同类Flask项目里很有代表性model层放SQLAlchemy模型router层放蓝图libs层放工具函数和响应封装migrations是alembic迁移目录config和key放配置与支付证书。下面是一份我整理过的核心文件职责表。2.1 文件职责速览路径类型职责app.py入口Flask应用初始化、配置加载、蓝图注册manage.py脚本命令行入口用于启动服务或执行命令model/user.py模型用户表model/commodity.py模型商品表model/category.py模型商品分类表model/shopcar.py模型购物车表model/order.py / orderitem.py模型订单主表与订单明细表router/user.py蓝图注册、登录、用户信息接口router/commodity.py蓝图商品列表、详情、搜索接口router/shopcar.py蓝图购物车增删改查接口router/order.py蓝图订单创建、支付、取消接口router/alipay.py蓝图支付宝支付回调处理libs/response.py工具统一JSON响应libs/auth.py装饰器JWT登录态校验migrations/脚本alembic数据库版本管理config/key/证书支付宝公钥与商户私钥从表里能看到这套源码把“路由—模型—业务”拆得非常清楚。实际项目里最怕的是把所有接口写在app.py一个文件里而这里router目录用一个个py文件隔开每个模块对应一组资源后续扩展、多人协作都不容易冲突。grub.py和grub2.py这种以“数据装载”命名的辅助脚本通常负责把外部商品文本灌入数据库后面会专门说。2.2 用户、商品与订单模型设计model/user.py采用的写法很标准是基于Flask-SQLAlchemy的。下面是我抽取出来的核心结构# model/user.py from datetime import datetime from werkzeug.security import generate_password_hash, check_password_hash from model import db # db在model/__init__.py中统一实例化 class User(db.Model): __tablename__ user id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(64), uniqueTrue, nullableFalse, indexTrue) password_hash db.Column(db.String(128), nullableFalse) phone db.Column(db.String(20), uniqueTrue) created_at db.Column(db.DateTime, defaultdatetime.utcnow) def set_password(self, password): self.password_hash generate_password_hash(password) def check_password(self, password): return check_password_hash(self.password_hash, password) def to_json(self): return { id: self.id, username: self.username, phone: self.phone, created_at: self.created_at.strftime(%Y-%m-%d %H:%M:%S) }这里有两个细节值得关注。一是密码字段不存明文用werkzeug.security做哈希登录时再通过check_password校验二是created_at用datetime.utcnow而不是datetime.now避免服务器时区影响时间记录。to_json方法是API项目里非常常见的做法ORM对象不直接序列化而是先转成字典再交给jsonify这样可以控制暴露字段比如password_hash永远不出现在响应体里。商品和分类模型之间是外键关系订单与用户是一对多订单与商品通过orderitem表形成多对多。源码中orderitem.py应该包含order_id、commodity_id、price、quantity这些字段。price单独存在订单明细里是电商必须的设计因为商品价格会变下单那一刻的价格不能被商品表后续改动影响否则对账时会出现金额对不上的问题。2.3 数据库迁移与初始化流程项目里带了一套migrations目录和alembic.ini这是用Flask-Migrate管理表结构变更的。拿到源码后只要Python依赖装好按下面三步就能把库建出来pip install -r requirements.txt flask db upgrade python manage.py runserver迁移脚本里已经有了一堆versions下的文件所以不需要再执行flask db init。那些b87a0779521b_.py之类的文件就是已经生成的迁移版本数据库会按顺序执行。如果没有执行flask db upgrade就直接跑应用会出现“No such table”的SQLAlchemy报错。如果要把默认的SQLite换成MySQL需要改config/settings.py里的连接串。常见做法是保留一个默认值再通过环境变量覆盖# config/settings.py import os BASE_DIR os.path.abspath(os.path.dirname(__file__)) class Config: SECRET_KEY os.environ.get(SECRET_KEY) or dev-secret SQLALCHEMY_TRACK_MODIFICATIONS False SQLALCHEMY_DATABASE_URI os.environ.get( DATABASE_URI, sqlite:/// os.path.join(BASE_DIR, shopping.db) )参数说明SECRET_KEY用于加密session和JWT生产环境一定不要用默认值DATABASE_URI通过环境变量注入例如mysqlpymysql://root:passlocalhost/shopping就能从SQLite平滑切到MySQL。大部分刚接触Flask的开发者会在这一步被坑——改了连接串但没装pymysql驱动启动时直接抛ModuleNotFoundError。3. 路由蓝图与响应规范API接口层的实现套路Flask路由层是用户直接接触的部分这套API的router目录下按资源划分了蓝图libs里封装了统一响应和错误码。下面从三个角度拆接口层。3.1 蓝图划分与路由注册蓝图的好处是同一个应用里可以按业务域拆分路由文件避免app.py越来越重。从router/下的文件名看接口前缀大概是这样划分的蓝图文件URL前缀主要接口router/user.py/api/userregister、login、inforouter/commodity.py/api/commoditylist、detail、searchrouter/shopcar.py/api/cartadd、update、delete、listrouter/order.py/api/ordercreate、pay、cancel、statusrouter/alipay.py/api/alipaynotify、return在入口文件里注册蓝图时应该像下面这样设置url_prefix让蓝图内部路由只写相对路径可读性更强# app.py from flask import Flask from router.user import user_bp from router.commodity import commodity_bp app Flask(__name__) app.config.from_object(config.settings.Config) def register_blueprints(): app.register_blueprint(user_bp, url_prefix/api/user) app.register_blueprint(commodity_bp, url_prefix/api/commodity) register_blueprints()url_prefix为/api/user代表蓝图里的user_bp.route(/login)最终对外暴露的路径是/api/user/login。这种挂载方式让接口文档天然好生成也方便统一加版本前缀。如果后续要升级到v2只需要再挂一次url_prefix原路由不破坏。项目里router/alipay.py单独一个蓝图也是为了让支付回调的路径不会被业务路由挤占。3.2 统一响应体与错误码从libs/response.py的命名看这套API采用了“业务状态码数据”的响应规范。接口返回的基本形态是# libs/response.py from flask import jsonify def success(dataNone, code0, messageok): return jsonify({ code: code, message: message, data: data }) def fail(code400, messageerror, dataNone): return jsonify({ code: code, message: message, data: data })使用统一响应后前端只需要解析code字段就能判断请求结果。我见过很多Flask项目有人返回字符串有人返回dict还有人直接render_template前端对接非常痛苦。这个源码的做法值得借鉴每个路由都返回success或fail包装过的JSON再配合libs/error_code.py里的错误码常量排查问题直接看code就可以定位到模块。下面是一份常用错误码分布表code含义典型场景0请求成功正常数据返回400参数错误缺少必填参数、参数格式错误401未登录或登录失效token缺失、token过期403无权限普通用户操作管理员接口500服务端异常数据库连接失败或未知异常设计错误码时要注意不要把HTTP状态码和业务code混为一谈。HTTP 200可以带业务code 400HTTP 401也可以带业务code 401。很多新手习惯把失败响应的HTTP状态码也改成400或500导致Nginx或网关侧日志无法按真实状态码统计后期排查会比较麻烦。3.3 登录态鉴权与token校验router/user.py里的登录接口通常会签一个token返回给前端后续请求把token放进Authorization头。libs/auth.py里封装的login_required装饰器是Flask API项目的标准做法# libs/auth.py from functools import wraps from flask import request, g, current_app import jwt def login_required(f): wraps(f) def wrapper(*args, **kwargs): token request.headers.get(Authorization, ) if token.startswith(Bearer ): token token[7:] if not token: return fail(code401, message未登录) try: payload jwt.decode(token, current_app.config[SECRET_KEY], algorithms[HS256]) g.user_id payload[uid] except jwt.ExpiredSignatureError: return fail(code401, message登录过期) except jwt.InvalidTokenError: return fail(code401, message无效token) return f(*args, **kwargs) return wrapper参数说明Authorization头的标准格式是“Bearer 空格 token”token在jwt.decode时使用的密钥必须和登录时一致算法一般选HS256。g.user_id存到Flask的g对象里视图函数里可以直接读取。这里最关键的坑是不要自己去base64解码token再判断用户jwt是带签名校验的只解不验容易被伪造。4. 从加购到支付购物车、订单和支付宝对接链路这一章把业务主链路串起来看商品列表、购物车操作、订单创建、支付对接和数据导入。4.1 商品查询与购物车数据操作先看商品列表router/commodity.py里最常见的写法是分页加过滤参数commodity_bp.route() def list_commodity(): page request.args.get(page, 1, typeint) per_page request.args.get(per_page, 20, typeint) category_id request.args.get(category_id, typeint) query Commodity.query if category_id: query query.filter_by(category_idcategory_id) pagination query.paginate(pagepage, per_pageper_page, error_outFalse) data { items: [c.to_json() for c in pagination.items], total: pagination.total, page: page, per_page: per_page } return success(datadata)page和per_page从URL的query string读取typeint做强制类型转换。error_outFalse非常关键它让页码超出范围时返回空列表而不是抛404前端拿到空数组后可以自己判断是否显示“没有更多数据”。如果希望接口更健壮建议把per_page限制到1到100之间避免有人传入超大数字拖垮数据库。购物车表一般是user_id、commodity_id、quantity的组合加购操作要判断商品是否存在、数量是否合法。下面是加购接口的常用逻辑cart_bp.route(/add, methods[POST]) login_required def add_to_cart(): commodity_id request.json.get(commodity_id) quantity request.json.get(quantity, 1) commodity Commodity.query.get(commodity_id) if not commodity: return fail(code400, message商品不存在) if quantity 0 or quantity 99: return fail(code400, message数量必须在1-99之间) cart ShopCar.query.filter_by( user_idg.user_id, commodity_idcommodity_id ).first() if cart: cart.quantity quantity else: cart ShopCar(user_idg.user_id, commodity_idcommodity_id, quantityquantity) db.session.add(cart) db.session.commit() return success(message已加入购物车)加上login_required之后g.user_id就是当前登录用户。同一商品重复加购时这里没有新增记录而是累加数量避免购物车出现重复行。如果你在复现时希望前端能直接改数量应该用update接口而不是把quantity直接覆盖为目标值否则并发操作容易互相覆盖。4.2 创建订单时的事务控制创建订单是这套API里最容易写出脏数据的地方因为要同时操作订单表、订单明细表、商品库存和购物车记录。源码里order.py的流程大致是order_bp.route(/create, methods[POST]) login_required def create_order(): user User.query.get(g.user_id) cart_items ShopCar.query.filter_by(user_iduser.id, checkedTrue).all() if not cart_items: return fail(code400, message没有选中的商品) order Order(user_iduser.id, statusunpaid, total_amount0) db.session.add(order) db.session.flush() # 只有flush之后 order.id 才会生成 total 0 order_items [] for cart in cart_items: commodity Commodity.query.get(cart.commodity_id) if commodity.stock cart.quantity: db.session.rollback() return fail(code400, messagef{commodity.title} 库存不足) commodity.stock - cart.quantity total commodity.price * cart.quantity order_items.append(OrderItem( order_idorder.id, commodity_idcommodity.id, pricecommodity.price, quantitycart.quantity )) db.session.delete(cart) order.total_amount total db.session.add_all(order_items) db.session.commit() return success(data{order_id: order.id, total_amount: total})这里最容易忽略的是db.session.flush()没有这一行order.id还是None后面订单明细的外键会直接报错。另一个关键点是循环里发现库存不足时调用db.session.rollback()把整个事务回滚否则可能出现“订单没创建成功但前面几件商品库存已经扣减”的脏数据。事务边界就是一次请求里要么全部成功提交要么一件事都不发生。订单创建后要维护status字段这套源码的状态流转一般如下status含义可跳转状态unpaid待支付paid / cancelledpaid已支付shipped / refundingshipped已发货completed / refundingcompleted已完成-cancelled已取消-接口层通常只允许有限的状态迁移比如只有unpaid状态的订单才能被取消paid之后必须走退款流程。如果直接把status改成任意值后续对账和库存回滚都会失控。4.3 支付宝支付接入与密钥配置源码包config/key下放了两个PEM文件app_private_key.pem是商户自己的应用私钥用于生成签名alipay_public_key.pem是支付宝公钥用于验证支付宝异步通知。很多人会把这两个文件放反导致pay接口能发起但回调验签始终失败。对接支付时的常见做法是生成一个支付链接让前端跳转核心步骤大致如下from alipay import AliPay alipay AliPay( appidcurrent_app.config[ALIPAY_APP_ID], app_notify_urlcurrent_app.config[ALIPAY_NOTIFY_URL], app_private_key_stringopen(current_app.config[APP_PRIVATE_KEY_PATH]).read(), alipay_public_key_stringopen(current_app.config[ALIPAY_PUBLIC_KEY_PATH]).read(), sign_typeRSA2, ) order_string alipay.api_alipay_trade_page_pay( out_trade_noorder_no, total_amounttotal_amount, subject购物平台订单, return_urlcurrent_app.config[ALIPAY_RETURN_URL], ) pay_url https://openapi.alipay.com/gateway.do? order_string生产环境网关一定不能写成沙箱地址沙箱环境要用openapi.alipaydev.com。app_notify_url是服务端异步通知地址必须公网可访问否则支付成功后支付宝回调进不来订单会一直停在等待支付状态。生产环境这里不能用localhost至少需要一个内网穿透或真实域名。支付成功以后回调里要重新校验金额、订单号、签名全部通过才把订单状态从unpaid改成paid。只依赖前端跳转返回是不可靠的因为用户可能支付成功后直接关掉了浏览器异步通知才是最终对账依据。4.4 用京东商品.txt批量导入初始化数据源码里京东商品.txt通常是从电商页面采集下来的商品信息配合grub.py或grub2.py用来初始化数据库。如果手动录入商品几十个分类会让人崩溃我一般会写一个不到100行的导入脚本# grub2.py from model import db, Commodity, Category def import_from_txt(path): with open(path, encodingutf-8) as f: for line in f: line line.strip() if not line: continue parts line.split(\t) # 按实际文件分隔符调整 if len(parts) 2: continue title, price parts[0], float(parts[1]) category_name parts[2] if len(parts) 2 else 默认分类 category Category.query.filter_by(namecategory_name).first() if not category: category Category(namecategory_name) db.session.add(category) db.session.flush() db.session.add(Commodity(titletitle, priceprice, category_idcategory.id)) db.session.commit()这个脚本有几个容易踩的细节。文件编码不一定都是UTF-8很多中文文本是GBK如果读出来乱码把encoding改成gbk。分隔符也不一定是\t打开文件看第一行再确认split()的参数。如果已经用alembic建好了表运行一次脚本再调商品列表接口就能看到数据。5. 启动部署避坑让Flask购物API真正能被调用前面把模型、路由和业务链路都拆完了最后聊一下我从源码到跑通服务时遇到的实际问题和验证方法。这套源码的入口是app.py或manage.py依赖装好后不要直接双击app.py很多问题出在环境变量和密钥路径上。第一坑密钥文件路径。config/settings.py里如果只用相对路径定位PEM证书在项目根目录启动没问题但用gunicorn启动且工作目录不对时open()证书会抛FileNotFoundError。建议在配置里基于BASE_DIR生成绝对路径启动前先打印确认文件存在。第二坑数据库表没建出来。必须执行flask db upgrade只有迁移执行过后续import model才不会出现NoSuchTableError。第三坑跨域访问。如果前端是独立域名的Vue项目需要支持OPTIONS预检和跨域响应头Flask里可以用flask-cors扩展解决别自己拼JSON返回。启动方式上我推荐先用开发服务器验证业务再用生产服务器接管python manage.py runserver --host 0.0.0.0 --port 5000 flask --app app.py routesflask routes命令会把所有已注册的路由完整输出包括HTTP方法和endpoint。第一次拿到这套源码时我会用这个命令确认哪些接口真正挂载成功再逐一用curl测试curl -X POST http://127.0.0.1:5000/api/user/login \ -H Content-Type: application/json \ -d {username:demo,password:123456}拿到token以后后续请求加上Authorization头再测购物车和订单接口。我习惯把每个接口的响应体都统一成success/fail结构前端只有在code0时才解析data其他情况直接弹message这样对账和排查都会非常省力。如果要上线把服务切到Gunicorn比如gunicorn -w 4 -b 0.0.0.0:5000 manage:app配合Nginx反代静态文件和API请求。数据库层面给commodity表的category_id和title加索引商品列表页的查询会快不少。购物车和订单表的数据量大以后记得按用户维度做分表先把开发环境的接口全部跑通再考虑这些优化不迟。本文还有配套的精品资源点击获取