大实训 · 从零创建 Flask 项目（4 小节连堂）

# app.py —— 项目入口。第 1 小节所有代码都先挤在这一个文件里，第 2 小节再拆分 from flask import Flask, jsonify # Flask 是应用类；jsonify 把字典转成 JSON 响应（自动加 Content-Type 头） app = Flask(__name__) # 创建应用实例；__name__ 帮 Flask 定位项目根目录（找静态文件、模板用） app.json.ensure_ascii = False # 让 JSON 里的中文原样显示，而不是 "\u53e4\u5178" 这样的转义 @app.route("/") # 路由装饰器：把下面的函数绑定到根路径 / —— curl 或浏览器访问 / 就会执行它 def index(): # 返回统一结构 {code, msg, data}：这是第 7 章的规范，从第一个接口就养成习惯 return jsonify({ "code": 200, # 业务状态码：200 = 成功 "msg": "success", # 给人看的提示信息 "data": { # 给程序用的具体数据 "service": "古典文学网 API", # 服务名——本实训的贯穿项目 "version": "0.1", # 版本号，每次大改动手动 +0.1，方便确认「跑的是新代码」 }, }) if __name__ == "__main__": # 只有直接运行本文件才执行（被 import 时不执行） app.run(debug=True, port=5000) # 启动开发服务器；debug=True：改代码自动重启、出错给调试页

# config.py —— 配置中心：把写死的配置收拢到一个文件，换环境只改这里 import os # 标准库：读取环境变量，避免把数据库密码硬编码进代码 from dotenv import load_dotenv # python-dotenv：启动时把 .env 文件里的键值对装进环境变量 load_dotenv() # 执行加载：之后 os.getenv("XXX") 就能读到 .env 里的 XXX 了 class BaseConfig: # 基础配置：开发和生产共用的部分 SECRET_KEY = os.getenv("SECRET_KEY", "dev-secret-key") # Flask 签名用密钥；第二个参数是没有环境变量时的兜底值 SQLALCHEMY_DATABASE_URI = os.getenv( # SQLAlchemy 连库地址（URI）：驱动+用户名+密码+主机+库名 "DATABASE_URL", "mysql+pymysql://root:123456@127.0.0.1:3306/wenxue_zero?charset=utf8mb4", ) # 默认连本机 MySQL——把 123456 换成你的密码；没装 MySQL 的同学见下方「SQLite 退路」 SQLALCHEMY_TRACK_MODIFICATIONS = False # 关闭模型变更追踪通知：我们用不到，还费内存 JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY", "jwt-dev-secret") # 签发 JWT 的密钥：泄露 = 别人能伪造登录态 class DevConfig(BaseConfig): # 开发环境：继承基础配置，只写差异项 DEBUG = True # 开调试模式：出错给交互式调试页、改代码自动重启 class ProdConfig(BaseConfig): # 生产环境 DEBUG = False # 生产绝不能开调试页——它会把源码和变量暴露给任何访问者 config_map = { # 名字 → 配置类 的映射表：应用工厂按名字取用 "dev": DevConfig, "prod": ProdConfig, }

# extensions.py —— db 等扩展实例在这里创建 # 为什么单独建文件：db 要被 models 和 views 同时 import。如果它在 app.py 里创建， # 就会出现 app.py → views/user.py → app.py 的「循环导入」——Python 直接报 ImportError。 # 把扩展放在独立的第三方中立文件里，大家 import 它，它谁也不 import，环就断了。 from flask_sqlalchemy import SQLAlchemy # ORM 扩展：用 Python 类操作数据库表 from flask_jwt_extended import JWTManager # JWT 扩展：负责签发与校验登录令牌 db = SQLAlchemy() # 创建“裸”的 db 实例——此刻还没绑定任何应用，所以谁 import 都安全 jwt = JWTManager() # 同理，JWT 管理器也先创建，等 app 工厂里再 init_app 绑定

# utils/response.py —— 统一响应格式：所有接口长一个样，前端只写一套处理逻辑 from flask import jsonify # 字典 → JSON 响应 def ok(data=None, msg="success", code=200): """成功响应：调用方只管给数据，外壳自动补齐""" return jsonify({"code": code, "msg": msg, "data": data}), code # 返回元组：第二项是 HTTP 状态码 def fail(msg="error", code=400, data=None): """失败响应：错误信息也包进同样结构，前端只需判断 code 字段""" return jsonify({"code": code, "msg": msg, "data": data}), code

# views/home.py —— 首页模块蓝图 from flask import Blueprint # Blueprint：可整体注册到主应用的“路由包” from utils.response import ok # 复用统一响应——接口里不再手写 jsonify 外壳 home_bp = Blueprint("home", __name__) # 创建蓝图；第一个参数是蓝图名，供 url_for 反查用 @home_bp.get("/") # 等价于 @home_bp.route("/", methods=["GET"]) —— .get/.post 是快捷写法 def index(): """首页探活接口：服务是否活着，看它就行""" return ok({"service": "古典文学网 API", "version": "0.2"}) # 版本号提到 0.2：拆分完成的标记

# views/__init__.py —— 视图包的“总装配车间” def register_blueprints(app): """把所有蓝图集中注册到 app：以后新增模块，只需在这里多 import 一行""" from views.home import home_bp # 注意：在函数体内 import（延迟导入）——进一步避免循环导入 app.register_blueprint(home_bp) # 注册后，home_bp 里的路由才真正生效

# app.py —— 应用工厂：只负责“组装零件”，不再写任何具体接口 from flask import Flask # 应用类 from config import config_map # 配置映射表 from extensions import db, jwt # 扩展实例（第 3 小节会用 jwt，现在先绑上） from views import register_blueprints # 蓝图总注册函数 def create_app(config_name="dev"): """应用工厂：每次调用返回一个按指定配置组装好的 Flask 应用""" app = Flask(__name__) # 创建应用实例 app.json.ensure_ascii = False # 中文原样显示：第 1 小节配置过，拆分后要跟着搬过来，否则中文变 \uXXXX app.config.from_object(config_map[config_name]) # 按名字加载配置类，如 "dev" → DevConfig db.init_app(app) # 把 db 绑定到这个 app——工厂模式的关键一步：先有裸实例，此处再绑定 jwt.init_app(app) # 同样绑定 JWT 管理器 register_blueprints(app) # 注册全部蓝图——路由从这一刻起生效 return app # 把组装好的应用交出去 app = create_app() # 模块级调用一次：供 python app.py / flask --app app 使用 if __name__ == "__main__": # 直接运行本文件时才启动开发服务器 app.run(debug=True, port=5000) # debug 也可以从 config 读，这里显式写方便第 1 次理解

-- 建库：utf8mb4 是完整 UTF-8 编码（MySQL 课讲过：utf8 是残缺版，存不了 emoji） CREATE DATABASE wenxue_zero DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; USE wenxue_zero; # 切换到新库 -- 表不用手动建：下一节的 User 模型会由 db.create_all() 自动建表（ORM 的意义）

# models/user.py —— 一个类对应一张表，一个属性对应一列 from werkzeug.security import check_password_hash, generate_password_hash # ^ werkzeug 是 Flask 自带依赖，不用额外安装：一对密码「散列/比对」工具 from extensions import db # 从 extensions 导入 db——绝不从 app 导入，防止循环导入 class User(db.Model): # 继承 db.Model 后，SQLAlchemy 才把它当表模型管理 __tablename__ = "user" # 指定表名；不写则默认用类名小写 id = db.Column(db.Integer, primary_key=True, autoincrement=True) # 主键，自增编号 username = db.Column(db.String(32), unique=True, nullable=False) # 用户名：唯一 + 非空，重复插入直接报错 password_hash = db.Column(db.String(256), nullable=False) # 存哈希！数据库里永远不出现明文密码 nickname = db.Column(db.String(32), default="") # 昵称：可选字段，默认空字符串 def set_password(self, raw): """注册/改密时调用：明文 → 单向散列""" self.password_hash = generate_password_hash(raw) # 单向：能从明文算出哈希，不能反推 def check_password(self, raw): """登录时调用：把用户输入同样散列后与库里的哈希比对""" return check_password_hash(self.password_hash, raw) # 相同明文 → 相同哈希 → True def to_dict(self): """序列化：模型对象 → 字典。注意：绝不把 password_hash 放进来""" return {"id": self.id, "username": self.username, "nickname": self.nickname}

# views/user.py —— 用户模块：两公开一保护 from flask import Blueprint, request # ^ request 对象封装了本次 HTTP 请求的一切：JSON 体、查询串、请求头（第 3 章核心） from flask_jwt_extended import create_access_token, get_jwt_identity, jwt_required # ^ JWT 三件套：签发令牌 / 从令牌读身份 / 保护接口（第 8 章核心） from extensions import db # db.session：数据库会话（暂存区 + 提交） from models.user import User # 用户模型 from utils.response import fail, ok # 统一响应 user_bp = Blueprint("user", __name__, url_prefix="/api/user") # ^ url_prefix：此蓝图下所有路由自动带 /api/user 前缀——接口分组从蓝图名开始 @user_bp.post("/register") # POST /api/user/register（写数据必须用 POST） def register(): data = request.get_json(silent=True) or {} # ^ 解析 JSON 请求体；silent=True：body 不是合法 JSON 时不抛异常而是返回 None，再 or {} 兜底成空字典 username = (data.get("username") or "").strip() # 取用户名并去掉首尾空格 password = data.get("password") or "" # 取密码；字段缺失时 or "" 防止 None 引发异常 if not username or not password: # 服务端必须再校验一次——前端校验可以被绕过 return fail("用户名和密码不能为空", code=400) # 400 = 请求参数有问题 if len(password) < 6: # 最简单的强度规则 return fail("密码至少 6 位", code=400) if User.query.filter_by(username=username).first(): # 按用户名查库：first() 取第一条，没有则 None return fail("用户名已被占用", code=400) user = User(username=username) # 创建模型对象——此刻只存在于内存，尚未写库 user.set_password(password) # 明文 → 哈希后再保存 db.session.add(user) # 放进会话（相当于 git add 的暂存区） db.session.commit() # 提交事务，真正写入 MySQL return ok({"id": user.id, "username": user.username}, msg="注册成功") # 只回非敏感字段 @user_bp.post("/login") # POST /api/user/login def login(): data = request.get_json(silent=True) or {} # 同上：安全解析请求体 user = User.query.filter_by(username=(data.get("username") or "").strip()).first() if user is None or not user.check_password(data.get("password") or ""): # ^ 用户不存在 或 密码不符——故意合并成一句提示，不给撞库者「哪个字段错了」的线索 return fail("用户名或密码错误", code=401) # 401 = 未通过身份认证 token = create_access_token(identity=str(user.id)) # ^ 签发 JWT：把用户 id 塞进令牌（identity 要求字符串，所以要 str()） # ^ 令牌由 JWT_SECRET_KEY 签名，客户端改一个字节都会校验失败 return ok({"token": token, "nickname": user.nickname}, msg="登录成功") # ^ 前端把 token 存起来（localStorage），之后每个受保护请求都带上它 @user_bp.get("/me") # GET /api/user/me —— 受保护接口 @jwt_required() # 守门装饰器：请求头没有合法 token，直接 401，函数体根本不会执行 def me(): uid = get_jwt_identity() # 从 token 里取回签发时存的 identity（用户 id 字符串） user = db.session.get(User, int(uid)) # 按主键查询（get 走主键，最快） if user is None: # token 合法但用户已被删——边界情况也要兜住 return fail("用户不存在", code=404) return ok(user.to_dict()) # 返回个人信息（不含密码哈希）

# views/__init__.py —— 增加 user_bp def register_blueprints(app): from views.home import home_bp from views.user import user_bp # 新增：导入用户蓝图 app.register_blueprint(home_bp) app.register_blueprint(user_bp) # 新增：注册用户蓝图

# app.py —— create_app 里、register_blueprints(app) 之后，加上自动建表 with app.app_context(): # 需要应用上下文：离开它 db 不知道自己服务于哪个 app db.create_all() # 扫描全部模型，表不存在就建（教学用；生产环境用 Flask-Migrate 管理变更）

# app.py —— create_app 内新增（放在 jwt.init_app(app) 之后） from utils.response import fail # 文件顶部别忘了导入 @jwt.unauthorized_loader # 钩子①：请求头压根没带 token 时触发 def missing_token(reason): return fail("请先登录", code=401) # 把框架默认的英文报错，统一成我们的响应结构 @jwt.expired_token_loader # 钩子②：token 过期时触发（header/payload 是 JWT 两段内容） def expired_token(header, payload): return fail("登录已过期，请重新登录", code=401) # 前端据此跳转登录页

# views/user.py —— 顶部加： import logging # 标准库日志模块 logger = logging.getLogger("wenxue") # 获取名为 wenxue 的日志器 # 在 register() 的 db.session.commit() 之后加一行： logger.info("用户 %s 注册成功，id=%s", username, user.id) # ^ 占位符式写法：值与模板分离，防注入也更易读；控制台会输出带时间的日志行