Flask 教程

会话保持

# 登录后测试受保护接口
def test_profile(client):
    # 先登录（Session 自动保持）
    client.post("/login", json={"username": "a", "password": "b"})

    resp = client.get("/profile")
    assert resp.status_code == 200

完整测试示例

def test_user_flow(client):
    # 创建
    resp = client.post("/api/users", json={"name": "张三"})
    assert resp.status_code == 201
    uid = resp.get_json()["id"]

    # 查询
    resp = client.get(f"/api/users/{uid}")
    assert resp.get_json()["name"] == "张三"

    # 更新
    resp = client.put(f"/api/users/{uid}", json={"name": "李四"})
    assert resp.get_json()["name"] == "李四"

    # 删除
    resp = client.delete(f"/api/users/{uid}")
    assert resp.status_code == 200

    # 404
    resp = client.get(f"/api/users/{uid}")
    assert resp.status_code == 404

Flask Test CLI Runner API

掌握 Flask Test CLI Runner API：命令行测试。

flask 命令入口

# 设置应用
export FLASK_APP=app.py
flask --help           # 查看所有命令

# 常用命令
flask run              # 启动开发服务器
flask routes           # 列出所有路由
flask shell            # 交互式 shell（带上下文）

自定义 CLI 命令

import click
from flask import Flask

app = Flask(__name__)

@app.cli.command("init-db")
def init_db_command():
    # 初始化数据库
    db.create_all()
    click.echo("数据库已初始化")

@app.cli.command("create-admin")
@click.argument("username")
@click.option("--password", prompt=True, hide_input=True)
def create_admin(username, password):
    # 创建管理员
    admin = User(username=username, password=password, is_admin=True)
    db.session.add(admin)
    db.session.commit()
    click.echo(f"管理员 {username} 已创建")

flask init-db
flask create-admin admin_user
flask create-admin --password 123456 bob

测试 CLI 命令

from click.testing import CliRunner

def test_cli():
    runner = CliRunner()
    result = runner.invoke(app.cli, ["init-db"])
    assert result.exit_code == 0
    assert "初始化" in result.output

cli 注册方式

# 方式一：装饰器（上面示例）
# 方式二：单独函数
import click

@click.command("init-db")
def init_db():
    pass

app.cli.add_command(init_db)

# 方式三：分组
@app.cli.group()
def db():
    # 数据库命令组
    pass

@db.command("init")
def db_init():
    click.echo("db init")

# flask db init

Flask Application Globals

掌握 Flask Application Globals：g 与 current_app。

g 对象（请求级全局）

from flask import g

# g 在同一个请求内跨函数共享
@app.before_request
def load_user():
    # 每次请求前加载用户
    g.user = User.query.first()

@app.route("/me")
def me():
    # 视图内直接使用
    return {"user": g.user.name if g.user else None}

# 多函数共享
def get_current_user():
    # 缓存到 g，避免重复查询
    if "user" not in g:
        g.user = User.query.first()
    return g.user

g 的应用场景

# 1. 数据库连接
@app.before_request
def open_db():
    g.db = connect_db()

@app.teardown_request
def close_db(exc):
    g.db.close()

# 2. 请求日志/耗时
@app.before_request
def start_timer():
    g.start = time.time()

# 3. 配置缓存
def get_settings():
    if "settings" not in g:
        g.settings = load_settings()
    return g.settings

current_app

from flask import current_app

# 在请求/应用上下文中访问应用配置
current_app.config["SECRET_KEY"]
current_app.name          # 应用名
current_app.logger        # 应用日志

# 在扩展初始化中常用
def init_db(app):
    with app.app_context():
        db.init_app(app)

g 的注意事项

• g 只在一个请求内有效，请求结束清空
• 不能跨请求/跨线程使用
• 适合缓存请求内临时数据
• 不要在 g 存大对象

Flask 实用函数与类

掌握 Flask 实用函数与类：常用内置工具。

URL 工具

from flask import url_for, request

# 反向生成 URL
url_for("index")
url_for("user", uid=42)
url_for("static", filename="css/style.css")
url_for("api.users", page=2)    # /api/users?page=2

# 当前 URL
request.url
request.base_url
request.host_url

安全工具

from werkzeug.security import generate_password_hash, check_password_hash

# 密码哈希（绝不存明文）
hashed = generate_password_hash("mypassword")
check_password_hash(hashed, "mypassword")   # True

from werkzeug.utils import secure_filename

# 安全文件名（防路径穿越）
secure_filename("../../etc/passwd")   # "etc_passwd"

JSON 工具

from flask import jsonify
import json

# Flask 的 jsonify（带 MIME）
jsonify({"ok": True})

# 标准 json
json.dumps({"a": 1}, ensure_ascii=False)
json.loads('{"a": 1}')

# 序列化对象
json.dumps(obj, default=str)   # 处理 datetime 等

请求/响应工具

from flask import abort, redirect, make_response

abort(404)                        # 终止并返回错误
abort(403, "无权访问")
redirect("/")                     # 重定向
make_response("内容")             # 构造响应

from flask import send_file, send_from_directory
send_file("file.pdf", as_attachment=True)
send_from_directory("uploads", "a.pdf")

模板全局工具

# 模板可用的内置对象
config        # 应用配置
request       # 请求
session       # 会话
g             # 全局
url_for()     # URL 生成
get_flashed_messages()  # 消息

Flask Message Flashing API

掌握 Flask Message Flashing API：闪现消息。

基本用法

from flask import flash, redirect, render_template

# 视图里发送消息
@app.route("/submit", methods=["POST"])
def submit():
    # 业务处理...
    flash("保存成功！", "success")    # 消息 + 类别
    flash("记得备份", "warning")
    return redirect("/")

# 一次性：下次请求读取后消失

模板中显示

<!-- base.html -->
{% with messages = get_flashed_messages(with_categories=true) %}
  {% if messages %}
    <div class="messages">
      {% for category, message in messages %}
        <div class="alert alert-{{ category }}">
          {{ message }}
        </div>
      {% endfor %}
    </div>
  {% endif %}
{% endwith %}

按类别过滤

# 只取某类别
get_flashed_messages(category_filter=["error"])

# 模板中
{% for msg in get_flashed_messages(category_filter=["error"]) %}
  <p class="error">{{ msg }}</p>
{% endfor %}

API 场景替代

# 前后端分离项目，flash 不太适用
# 直接返回结构化信息
return jsonify({
    "success": True,
    "message": "保存成功",
    "code": 0,
}), 200

# flash 适合服务端渲染（传统 Web）场景

完整示例

@app.route("/register", methods=["POST"])
def register():
    form = RegisterForm()
    if form.validate_on_submit():
        user = User(username=form.username.data)
        db.session.add(user)
        db.session.commit()
        flash(f"欢迎加入，{user.username}！", "success")
        return redirect(url_for("index"))

    flash("表单校验失败，请检查", "error")
    return render_template("register.html", form=form)

Flask JSON

掌握 Flask JSON：JSON 的序列化与解析。

返回 JSON

from flask import jsonify

# 字典自动转 JSON
@app.route("/api/data")
def data():
    return jsonify({
        "name": "CodeNest",
        "version": 1.0,
        "tags": ["教程", "编程"],
        "active": True
    })

# 数组
return jsonify([1, 2, 3])

# 指定状态码
return jsonify({"error": "未找到"}), 404

序列化对象

# 模型转字典
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))
    created_at = db.Column(db.DateTime)

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "created_at": self.created_at.isoformat() if self.created_at else None,
        }

# 列表
users = User.query.all()
return jsonify([u.to_dict() for u in users])

解析 JSON 请求

from flask import request

@app.post("/api/users")
def create():
    data = request.get_json()          # 解析失败返回 None
    if data is None:
        return {"error": "无效的 JSON"}, 400

    # silent=True：解析失败不抛异常
    data = request.get_json(silent=True)

    # force=True：非 JSON Content-Type 也尝试解析
    data = request.get_json(force=True)

    name = data.get("name")
    return {"created": name}, 201

JSON 编码细节

# 中文不转义
import json
json.dumps({"name": "张三"}, ensure_ascii=False)
# '{"name": "张三"}'

# 处理不可序列化对象
json.dumps({"date": datetime.now()}, default=str)

# Flask 配置
app.config["JSON_AS_ASCII"] = False    # 中文原样（Flask 2.2 前）
# Flask 3.x 默认 ensure_ascii=False

统一响应包装

def api_response(data=None, message="ok", code=0, status=200):
    return jsonify({
        "code": code,
        "message": message,
        "data": data,
    }), status

@app.get("/api/users")
def users():
    return api_response([{"id": 1}])

@app.post("/api/users")
def create_user():
    data = request.get_json()
    if not data or not data.get("name"):
        return api_response(message="参数错误", code=400, status=400)
    return api_response({"id": 1}, status=201)

Flask 模板渲染 API

掌握 Flask 模板渲染 API：Jinja2 高级用法。

render_template 参数

from flask import render_template

# 基本
render_template("index.html", title="首页", users=users)

# 模板字符串（不常用）
from flask import render_template_string
render_template_string("<h1>{{ title }}</h1>", title="Hello")

模板上下文处理器

# 给所有模板注入变量
@app.context_processor
def inject_globals():
    return {
        "site_name": "CodeNest",
        "current_year": datetime.now().year,
    }

# 所有模板可直接使用 {{ site_name }}
# <p>{{ site_name }} © {{ current_year }}</p>

模板过滤器

# 自定义过滤器
@app.template_filter("money")
def money_filter(value):
    return f"¥{value:,.2f}"

# 模板中
# {{ price | money }}  → ¥1,234.56

# 注册函数到模板
@app.template_global()
def format_time(dt):
    return dt.strftime("%Y-%m-%d %H:%M")

Jinja2 高级语法

<!-- 宏（可复用片段）-->
{% macro input(name, type="text", placeholder="") %}
  <input type="{{ type }}" name="{{ name }}" placeholder="{{ placeholder }}">
{% endmacro %}

{{ input("username") }}
{{ input("password", type="password") }}

<!-- 循环高级 -->
{% for item in items %}
  {% if loop.first %}首个{% endif %}
  {{ loop.index }}. {{ item.name }}   <!-- 索引从 1 开始 -->
  {% if loop.last %}末个{% endif %}
{% endfor %}

<!-- set 定义变量 -->
{% set base_price = 100 %}
<p>{{ base_price * 0.9 }}</p>

转义与安全

<!-- 自动转义（防 XSS）-->
{{ user_input }}        <!-- <script> 会被转义 -->

<!-- 确需输出 HTML（仅可信内容）-->
{{ content | safe }}

<!-- 禁用全局自动转义（不推荐）-->
{% autoescape false %}
  {{ content }}
{% endautoescape %}

Flask Config API

掌握 Flask Config API：配置加载与访问。

配置属性访问

from flask import Flask, current_app

app = Flask(__name__)

# 设置
app.config["DEBUG"] = True
app.config["SECRET_KEY"] = "secret"

# 读取（在上下文中）
current_app.config["SECRET_KEY"]
current_app.config.get("NOT_EXIST", "default")

# 视图外（无上下文）
with app.app_context():
    val = current_app.config["DEBUG"]

常用配置项

app.config.update(
    DEBUG=False,                     # 调试
    SECRET_KEY="complex",            # 密钥
    JSON_AS_ASCII=False,             # JSON 中文
    SESSION_COOKIE_HTTPONLY=True,    # Session 安全
    SQLALCHEMY_DATABASE_URI="sqlite:///app.db",
    MAX_CONTENT_LENGTH=16 * 1024 * 1024,  # 上传限制 16MB
    PERMANENT_SESSION_LIFETIME=86400,
)

配置加载方式

# from_object
app.config.from_object("config.ProdConfig")

# from_pyfile
app.config.from_pyfile("config.py")

# from_envvar（环境变量指向配置文件）
app.config.from_envvar("APP_CONFIG")

# from_mapping
app.config.from_mapping(
    DEBUG=True,
    SECRET_KEY="x",
)

环境变量配置

import os

app.config.update(
    SECRET_KEY=os.environ.get("SECRET_KEY", "dev-secret"),
    DB_URL=os.environ.get("DB_URL", "sqlite:///dev.db"),
)

读取配置最佳实践

# 在扩展初始化中使用
db.init_app(app)   # 内部读取 SQLALCHEMY_DATABASE_URI

# 自定义配置封装
def get_config(key, default=None):
    return current_app.config.get(key, default)

Flask Stream Helpers API

掌握 Flask Stream Helpers API：流式响应。

流式响应基础

from flask import Response

@app.route("/stream")
def stream():
    def generate():
        for i in range(10):
            yield f"数据块 {i}
"

    return Response(generate(), mimetype="text/plain")

# 客户端逐步接收，而非一次性

实时日志流

import time

@app.route("/logs")
def stream_logs():
    def generate():
        while True:
            yield f"日志时间: {time.time()}
"
            time.sleep(1)
    return Response(generate(), mimetype="text/plain")

大文件下载

# 大文件分块发送（省内存）
@app.route("/download")
def download():
    def generate():
        with open("big_file.zip", "rb") as f:
            while chunk := f.read(8192):
                yield chunk
    return Response(
        generate(),
        headers={"Content-Disposition": "attachment; filename=big.zip"},
        mimetype="application/zip",
    )

SSE（Server-Sent Events）

import json

@app.route("/events")
def sse():
    def generate():
        yield "retry: 3000

"   # 3 秒重连
        for i in range(5):
            data = json.dumps({"count": i})
            yield f"data: {data}

"   # SSE 格式
            time.sleep(1)
    return Response(generate(), mimetype="text/event-stream")

// 前端接收 SSE
const es = new EventSource("/events");
es.onmessage = (e) => console.log(JSON.parse(e.data));

注意

• 流式响应用生成器（yield）
• 长时间流需考虑代理超时设置
• SSE 是单向推送（服务器→客户端）；双向用 WebSocket

Flask 内部对象 API

掌握 Flask 内部对象 API：请求/应用上下文内部机制。

本地代理（Local Proxy）

# request/session/current_app/g 都是"本地代理"
# 它们根据当前上下文自动指向正确的对象

from flask import request, current_app, g

# 使用（无需关心上下文切换）
request.method
current_app.name
g.user

上下文对象

from flask import Flask, request, session

app = Flask(__name__)

# 请求上下文：request + session
# 应用上下文：current_app + g

# 手动管理（一般用 with）
with app.test_request_context("/?page=2"):
    print(request.args.get("page"))   # 2

# 内部实现
from flask.ctx import RequestContext, AppContext

# 上下文栈（ctx stack）
# flask._app_ctx_stack
# flask._request_ctx_stack

current_app 工作原理

# current_app 是 LocalProxy
# 它从应用上下文栈取栈顶的 app

# 为什么要代理？
# - 一个进程可能有多个 Flask 应用（测试时常见）
# - 代理保证总是取"当前"的应用

# 直接使用 app 对象也可以（简单场景）
app.config["DEBUG"]

g 的生命周期

# g 绑定在应用上下文
# 应用上下文 push → g 可用
# 应用上下文 pop → g 清空

with app.app_context():
    g.x = 1
# 离开 with 后 g 不可用

# 每个请求：请求上下文 push → 视图 → pop
# 所以 g 是"请求级"的

何时需要了解内部机制

• 调试"working outside of application context"错误
• 编写扩展（需要访问上下文）
• 理解测试中的 context
• 多应用/多线程场景

Flask 信号 API

掌握 Flask 信号 API：事件通知机制（Blinker）。

信号基础

# Flask 内置信号（需要 blinker）
from flask.signals import (
    request_started, request_finished,
    template_rendered, got_request_exception
)

# 订阅信号
def log_request(sender, **extra):
    print("请求开始")

request_started.connect(log_request)

# 可取消订阅
request_started.disconnect(log_request)

内置信号

自定义信号

from blinker import Namespace

# 创建命名空间
my_signals = Namespace()

# 定义信号
user_registered = my_signals.signal("user-registered")
user_logged_in = my_signals.signal("user-logged-in")

# 发送信号
@app.route("/register", methods=["POST"])
def register():
    user = create_user(...)
    user_registered.send(app, user=user)   # 发送（携带数据）
    return {"ok": True}

订阅自定义信号

def send_welcome_email(sender, user, **extra):
    print(f"给 {user.username} 发欢迎邮件")

user_registered.connect(send_welcome_email)

# 装饰器方式
@user_registered.connect
def log_register(sender, user, **extra):
    app.logger.info(f"新用户注册: {user.username}")

信号 vs 钩子

Flask 类视图 API

掌握 Flask 类视图 API：用类组织视图（Flask 3.0+）。

method_view（Flask 3.0+）

from flask import request, jsonify
from flask.views import MethodView

# 类视图：按 HTTP 方法组织
class UserView(MethodView):
    def get(self):
        users = User.query.all()
        return jsonify([u.to_dict() for u in users])

    def post(self):
        data = request.get_json()
        user = User(name=data["name"])
        db.session.add(user)
        db.session.commit()
        return jsonify(user.to_dict()), 201

class UserDetailView(MethodView):
    def get(self, uid):
        user = db.session.get(User, uid)
        if not user:
            return {"error": "不存在"}, 404
        return jsonify(user.to_dict())

    def put(self, uid):
        user = db.session.get(User, uid)
        if not user:
            return {"error": "不存在"}, 404
        data = request.get_json()
        user.name = data.get("name", user.name)
        db.session.commit()
        return jsonify(user.to_dict())

    def delete(self, uid):
        user = db.session.get(User, uid)
        if not user:
            return {"error": "不存在"}, 404
        db.session.delete(user)
        db.session.commit()
        return {"ok": True}

# 注册
app.add_url_rule("/api/users", view_func=UserView.as_view("users"))
app.add_url_rule("/api/users/",
                 view_func=UserDetailView.as_view("user_detail"))

类视图装饰器

class AdminView(MethodView):
    decorators = [login_required]   # 所有方法加装饰器

    def get(self):
        return {"admin": True}

    def post(self):
        return {"created": True}

类视图 vs 函数视图

Flask 定义路由 API

掌握 Flask 定义路由 API：路由注册的多种方式。

方式一：装饰器（最常用）

@app.route("/")
def index():
    return "首页"

# 指定方法
@app.route("/data", methods=["GET", "POST"])
def data():
    return {"ok": True}

# 简化写法（Flask 2.0+）
@app.get("/data")
def get_data(): ...
@app.post("/data")
def post_data(): ...
@app.put("/data")
def put_data(): ...
@app.delete("/data")
def delete_data(): ...

方式二：add_url_rule

def index():
    return "首页"

app.add_url_rule("/", "index", index)

# 带方法
app.add_url_rule("/api/users", "users", users_view,
                 methods=["GET", "POST"])

# 等价于 @app.route

方式三：蓝图注册

bp = Blueprint("api", __name__, url_prefix="/api")

@bp.route("/users")
def users():
    return {"ok": True}

app.register_blueprint(bp)

方式四：类视图

from flask.views import MethodView

class UsersView(MethodView):
    def get(self): return {"method": "get"}
    def post(self): return {"method": "post"}

app.add_url_rule("/api/users", view_func=UsersView.as_view("users"))

路由命名与 url_for

# 默认端点名 = 函数名
@app.route("/user/")
def user_detail(uid): ...

url_for("user_detail", uid=42)   # /user/42

# 自定义端点名
@app.route("/x", endpoint="my_x")
def x(): ...
url_for("my_x")

查看所有路由

flask routes
# 或
# python -c "from app import app; [print(r) for r in app.url_map]"
# Endpoint  Methods  Rule
# index    GET      /
# users    GET,POST /api/users

Flask 视图函数 API

掌握 Flask 视图函数 API：返回值与请求处理。

视图返回类型

from flask import jsonify, render_template, redirect, Response

# 1. 字符串（默认 text/html）
return "hello"

# 2. 字典/列表（自动 JSON）
return {"ok": True}
return [1, 2, 3]

# 3. 元组
return "内容", 201
return "内容", 200, {"X-Custom": "v"}

# 4. jsonify
return jsonify(ok=True)

# 5. 模板
return render_template("index.html")

# 6. 重定向
return redirect("/other")

# 7. Response
return Response("内容", status=200)

获取请求数据

from flask import request

@app.post("/api/users")
def create():
    # 参数优先级：args(URL) > form > json > files
    data = request.get_json() or request.form

    name = data.get("name")
    page = request.args.get("page", 1, type=int)

    # 校验
    if not name:
        return {"error": "缺少 name"}, 400

    return {"created": name, "page": page}, 201

错误返回

from flask import abort

# 快速错误
abort(404)
abort(400, "参数错误")   # 自定义描述

# 自定义错误处理
@app.errorhandler(400)
def bad_request(e):
    return jsonify({
        "error": "Bad Request",
        "detail": getattr(e, "description", "")
    }), 400

请求钩子与视图

@app.before_request
def before():
    # 视图前的统一处理
    pass

@app.after_request
def after(response):
    # 视图后的统一处理
    return response

视图最佳实践

• 参数解析（request）与校验分离
• 业务逻辑抽到 service 层
• 统一返回格式（code/message/data）
• 视图保持"薄"（薄视图，厚服务）

Flask 命令行 API

掌握 Flask 命令行 API：CLI 命令与交互式 shell。

flask 命令基础

export FLASK_APP=app.py
export FLASK_ENV=development    # 开发模式（2.3 前）
# Flask 2.3+ 用 --debug
flask --app app.py run --debug

# 常用命令
flask run          # 启动开发服务器
flask routes       # 列出路由
flask shell        # 交互式 shell
flask --help       # 帮助

自定义命令

import click
from flask import Flask

app = Flask(__name__)

@app.cli.command("init-db")
def init_db():
    # 初始化数据库
    db.create_all()
    click.echo("数据库初始化完成")

@app.cli.command("seed")
@click.option("--count", default=10, help="数量")
def seed(count):
    # 填充测试数据
    for i in range(count):
        db.session.add(User(name=f"用户{i}"))
    db.session.commit()
    click.echo(f"已创建 {count} 条数据")

flask init-db
flask seed --count 20

命令分组

@app.cli.group("db")
def db_group():
    # 数据库命令组

@db_group.command("init")
def db_init():
    click.echo("db init")

@db_group.command("drop")
def db_drop():
    click.echo("db drop")

# flask db init / flask db drop

flask shell 增强

# 自动导入常用对象到 shell
@app.shell_context_processor
def make_shell_context():
    return {
        "db": db,
        "User": User,
        "app": app,
    }

# flask shell 后可直接用 db、User

CLI 测试

from click.testing import CliRunner

def test_cli():
    runner = CliRunner()
    result = runner.invoke(app.cli, ["init-db"])
    assert result.exit_code == 0

Flask Web 应用

综合运用 Flask 知识，构建完整 Web 应用。

项目规划

功能：
1. 文章列表（分页 + 搜索）
2. 文章详情
3. 分类筛选
4. 用户注册/登录
5. 后台管理（增删改）
6. 部署上线

项目结构

blog/
├── run.py              # 入口
├── config.py           # 配置
├── app/
│   ├── __init__.py     # 工厂
│   ├── models.py       # 模型
│   ├── forms.py        # 表单
│   ├── views/
│   │   ├── __init__.py
│   │   ├── main.py     # 前台
│   │   ├── auth.py     # 认证
│   │   └── admin.py    # 后台
│   ├── templates/
│   └── static/
├── requirements.txt
└── .env

工厂函数

# app/__init__.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

def create_app():
    app = Flask(__name__)
    app.config.from_object("config.Config")

    db.init_app(app)

    # 注册蓝图
    from .views.main import main_bp
    from .views.auth import auth_bp
    from .views.admin import admin_bp

    app.register_blueprint(main_bp)
    app.register_blueprint(auth_bp, url_prefix="/auth")
    app.register_blueprint(admin_bp, url_prefix="/admin")

    return app

模型

# app/models.py
from . import db
from datetime import datetime

class Category(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True)
    articles = db.relationship("Article", backref="category")

class Article(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text)
    category_id = db.Column(db.Integer,
                            db.ForeignKey("category.id"))
    views = db.Column(db.Integer, default=0)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

前台视图

# app/views/main.py
from flask import Blueprint, render_template
from ..models import Article

main_bp = Blueprint("main", __name__)

@main_bp.route("/")
def index():
    page = request.args.get("page", 1, type=int)
    articles = Article.query.order_by(
        Article.created_at.desc()).paginate(
            page=page, per_page=10)
    return render_template("index.html", articles=articles)

后台管理

# app/views/admin.py
from flask import Blueprint, request, redirect, url_for
from ..models import Article, db

admin_bp = Blueprint("admin", __name__)

@admin_bp.route("/articles/create", methods=["GET", "POST"])
def create():
    if request.method == "POST":
        article = Article(
            title=request.form["title"],
            content=request.form["content"],
        )
        db.session.add(article)
        db.session.commit()
        return redirect(url_for("main.index"))
    return render_template("admin/form.html")

Jinja2 模板

掌握 Jinja2 模板：Flask 内置模板引擎详解。

模板基础

<!-- 变量与过滤器 -->
<p>{{ user.name | upper }}</p>
<p>{{ content | truncate(30) }}</p>
<p>{{ price | round(2) }}</p>

<!-- 控制结构 -->
{% if user.is_admin %}
  <a href="/admin">后台</a>
{% endif %}

{% for item in items %}
  <li>{{ loop.index }}. {{ item }}</li>
{% endfor %}

继承与块

<!-- base.html -->
<html>
<head>
  <title>{% block title %}CodeNest{% endblock %}</title>
  {% block head %}{% endblock %}
</head>
<body>
  {% include "nav.html" %}
  {% block content %}{% endblock %}
  {% block footer %}{% endblock %}
</body>
</html>

自定义过滤器

@app.template_filter("timeago")
def timeago(dt):
    seconds = (datetime.now() - dt).total_seconds()
    if seconds < 60: return "刚刚"
    if seconds < 3600: return f"{int(seconds//60)} 分钟前"
    if seconds < 86400: return f"{int(seconds//3600)} 小时前"
    return f"{int(seconds//86400)} 天前"

# 模板：{{ article.created_at | timeago }}

模板测试

{% if user is defined %}已定义{% endif %}
{% if items is iterable %}可迭代{% endif %}
{% if 'a' in items %}包含{% endif %}

性能与安全

• 默认自动转义（防 XSS）
• 复杂逻辑放视图/过滤器，模板保持简单
• 频繁渲染可开模板缓存（app.config["TEMPLATES_AUTO_RELOAD"]）

Flask-SQLAlchemy

掌握 Flask-SQLAlchemy：Flask 与 SQLAlchemy 的集成。

初始化

pip install flask-sqlalchemy

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///blog.db"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False

db = SQLAlchemy(app)

定义模型

class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True, nullable=False)
    email = db.Column(db.String(100), unique=True)
    password_hash = db.Column(db.String(128))
    is_admin = db.Column(db.Boolean, default=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    # 关系
    articles = db.relationship("Article", backref="author",
                               lazy="dynamic")

    def to_dict(self):
        return {
            "id": self.id,
            "username": self.username,
            "email": self.email,
            "created_at": self.created_at.isoformat() if self.created_at else None,
        }

class Article(db.Model):
    __tablename__ = "articles"

    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    content = db.Column(db.Text)
    user_id = db.Column(db.Integer, db.ForeignKey("users.id"))
    category_id = db.Column(db.Integer, db.ForeignKey("categories.id"))
    views = db.Column(db.Integer, default=0)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

class Category(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50), unique=True)
    articles = db.relationship("Article", backref="category")

建表与增删改查

# 建表
with app.app_context():
    db.create_all()

# 增
user = User(username="张三", email="z@example.com")
db.session.add(user)
db.session.commit()

# 查
User.query.all()                       # 全部
User.query.get(1)                      # 按主键
User.query.filter_by(username="张三").first()
User.query.filter(User.id > 5).all()
User.query.order_by(User.created_at.desc()).limit(10).all()

# 改
user = User.query.get(1)
user.username = "李四"
db.session.commit()

# 删
db.session.delete(user)
db.session.commit()

分页

page_obj = Article.query.paginate(page=1, per_page=10)
page_obj.items      # 当前页数据
page_obj.total      # 总数
page_obj.pages      # 总页数
page_obj.has_next   # 是否有下一页
page_obj.next_num   # 下一页页码

视图函数

掌握 Flask 视图函数：视图的组织与模式。

视图模式

# 1. 简单视图
@app.route("/")
def index():
    return render_template("index.html")

# 2. 带参数
@app.route("/article/")
def article(aid):
    a = Article.query.get_or_404(aid)
    return render_template("detail.html", article=a)

# 3. 处理表单
@app.route("/create", methods=["GET", "POST"])
def create():
    if request.method == "POST":
        title = request.form.get("title")
        if not title:
            flash("标题必填", "error")
        else:
            db.session.add(Article(title=title))
            db.session.commit()
            flash("创建成功", "success")
            return redirect(url_for("index"))
    return render_template("create.html")

装饰器封装

from functools import wraps
from flask import redirect, url_for, session

def login_required(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        if not session.get("user_id"):
            flash("请先登录", "warning")
            return redirect(url_for("auth.login"))
        return f(*args, **kwargs)
    return wrapper

def admin_required(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        if not session.get("is_admin"):
            abort(403)
        return f(*args, **kwargs)
    return wrapper

@app.route("/dashboard")
@login_required
def dashboard():
    return render_template("dashboard.html")

@app.route("/admin")
@login_required
@admin_required
def admin():
    return render_template("admin.html")

错误视图

@app.errorhandler(404)
def not_found(e):
    return render_template("404.html"), 404

@app.errorhandler(403)
def forbidden(e):
    return render_template("403.html"), 403

视图组织建议

• 按蓝图分模块（main/auth/admin）
• 权限用装饰器统一管理
• 业务逻辑抽 service 层
• 错误页统一处理

Flask-Migrate

掌握 Flask-Migrate：数据库迁移（版本控制表结构）。

安装与初始化

pip install flask-migrate

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"
db = SQLAlchemy(app)
migrate = Migrate(app, db)

# 模型...
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(50))

export FLASK_APP=app.py

# 初始化迁移目录
flask db init

# 生成迁移脚本（对比模型与数据库）
flask db migrate -m "create users table"

# 应用迁移
flask db upgrade

# 回滚一个版本
flask db downgrade

# 查看状态
flask db current
flask db history

模型变更流程

1. 修改 models.py（如加字段）
2. flask db migrate -m "add email to users"
3. 检查生成的迁移文件
4. flask db upgrade
5. 测试

迁移文件

# migrations/versions/xxx_create_users.py
def upgrade():
    op.add_column("users",
                  sa.Column("email", sa.String(100), nullable=True))

def downgrade():
    op.drop_column("users", "email")

为什么需要迁移

• 表结构变更可追踪、可回滚
• 团队协作同步表结构
• 生产环境安全升级
• 替代手动 ALTER TABLE

关键词搜索与分页

掌握 Flask 关键词搜索与分页：内容站核心功能。

搜索实现

@app.route("/search")
def search():
    keyword = request.args.get("q", "").strip()

    if not keyword:
        return render_template("search.html", articles=[])

    # LIKE 模糊搜索
    articles = Article.query.filter(
        Article.title.contains(keyword) |
        Article.content.contains(keyword)
    ).order_by(Article.created_at.desc()).all()

    return render_template("search.html",
                           articles=articles, keyword=keyword)

分页实现

@app.route("/")
def index():
    page = request.args.get("page", 1, type=int)
    per_page = 10

    # 分页查询
    pagination = Article.query.order_by(
        Article.created_at.desc()
    ).paginate(page=page, per_page=per_page)

    return render_template("index.html", pagination=pagination)

分页模板

<!-- 分页导航 -->
{% if pagination.pages > 1 %}
<nav class="pagination">
  {% if pagination.has_prev %}
    <a href="?page={{ pagination.prev_num }}">上一页</a>
  {% endif %}

  <span>{{ pagination.page }} / {{ pagination.pages }}</span>

  {% if pagination.has_next %}
    <a href="?page={{ pagination.next_num }}">下一页</a>
  {% endif %}
</nav>
{% endif %}

搜索 + 分页组合

@app.route("/search")
def search():
    keyword = request.args.get("q", "").strip()
    page = request.args.get("page", 1, type=int)

    query = Article.query
    if keyword:
        query = query.filter(
            Article.title.contains(keyword))

    pagination = query.order_by(
        Article.created_at.desc()
    ).paginate(page=page, per_page=10)

    return render_template("search.html",
                           pagination=pagination,
                           keyword=keyword)

<!-- 分页时保留搜索词 -->
<a href="?q={{ keyword }}&page={{ pagination.next_num }}">下一页</a>

优化建议

• 数据量大时 LIKE 慢，用全文索引（MySQL FULLTEXT）
• 更大规模用 Elasticsearch
• 分页加 ORDER BY（稳定顺序）
• 关键词防注入：SQLAlchemy 自动参数化

Flask-Login

掌握 Flask-Login：用户认证扩展。

安装与初始化

pip install flask-login

from flask_login import LoginManager, UserMixin, login_user,     logout_user, login_required, current_user

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = "auth.login"      # 未登录跳转
login_manager.login_message = "请先登录"       # 提示

# User 模型继承 UserMixin
class User(UserMixin, db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(50), unique=True)
    password_hash = db.Column(db.String(128))

# 用户加载器
@login_manager.user_loader
def load_user(user_id):
    return db.session.get(User, int(user_id))

登录/登出

@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        user = User.query.filter_by(
            username=request.form["username"]).first()

        if user and check_password(user.password_hash,
                                   request.form["password"]):
            login_user(user, remember=True)   # 记住登录
            return redirect(request.args.get("next") or url_for("index"))

        flash("用户名或密码错误", "error")
    return render_template("login.html")

@app.route("/logout")
@login_required
def logout():
    logout_user()
    return redirect(url_for("index"))

保护视图

from flask_login import login_required, current_user

@app.route("/profile")
@login_required
def profile():
    return render_template("profile.html", user=current_user)

# 模板中判断
# {% if current_user.is_authenticated %}已登录{% endif %}
# {% if current_user.is_authenticated %}
#   {{ current_user.username }}
# {% else %}
#   <a href="{{ url_for('login') }}">登录</a>
# {% endif %}

权限控制

from functools import wraps

def admin_required(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        if not current_user.is_admin:
            abort(403)
        return f(*args, **kwargs)
    return wrapper

@app.route("/admin")
@login_required
@admin_required
def admin():
    return "管理后台"

其他功能

# 未认证访问 → 跳转登录页（login_view）
# remember=True → 关闭浏览器保持登录
# current_user → 模板/视图中的当前用户
# is_authenticated → 是否登录
# is_anonymous → 是否匿名

表单处理

掌握 Flask 表单处理：Flask-WTF 完整实践。

安装

pip install flask-wtf

定义表单

from flask_wtf import FlaskForm
from wtforms import (StringField, PasswordField, TextAreaField,
                     BooleanField, SelectField, DateField)
from wtforms.validators import (DataRequired, Email, Length,
                                EqualTo, Optional)

class ArticleForm(FlaskForm):
    title = StringField("标题", validators=[
        DataRequired(message="标题必填"),
        Length(min=3, max=100, message="3-100 字符"),
    ])
    category = SelectField("分类", coerce=int, validators=[
        DataRequired()])
    content = TextAreaField("内容", validators=[
        DataRequired()])
    is_published = BooleanField("发布")

class RegisterForm(FlaskForm):
    username = StringField("用户名", validators=[
        DataRequired(), Length(3, 20)])
    email = StringField("邮箱", validators=[
        DataRequired(), Email()])
    password = PasswordField("密码", validators=[
        DataRequired(), Length(6, 50)])
    confirm = PasswordField("确认密码", validators=[
        EqualTo("password", message="两次密码不一致")])

视图处理

@app.route("/articles/new", methods=["GET", "POST"])
def new_article():
    form = ArticleForm()
    form.category.choices = [
        (c.id, c.name) for c in Category.query.all()]

    if form.validate_on_submit():      # POST + 校验通过
        article = Article(
            title=form.title.data,
            content=form.content.data,
            category_id=form.category.data,
            is_published=form.is_published.data,
        )
        db.session.add(article)
        db.session.commit()
        flash("文章已发布", "success")
        return redirect(url_for("index"))

    return render_template("form.html", form=form)

模板渲染

<form method="post">
  {{ form.hidden_tag() }}    <!-- CSRF 令牌（必须） -->

  <div>
    {{ form.title.label }}
    {{ form.title(class="form-control") }}
    {% for err in form.title.errors %}
      <span class="error">{{ err }}</span>
    {% endfor %}
  </div>

  <div>{{ form.category.label }} {{ form.category() }}</div>
  <div>{{ form.content.label }} {{ form.content(rows=8) }}</div>
  <div>{{ form.is_published() }} {{ form.is_published.label }}</div>

  <button type="submit">保存</button>
</form>

错误处理

@app.route("/register", methods=["GET", "POST"])
def register():
    form = RegisterForm()
    if form.validate_on_submit():
        # 处理注册...
        flash("注册成功", "success")
        return redirect(url_for("login"))

    # 校验失败：form.errors 含错误信息
    # 模板中自动显示（errors 循环）
    return render_template("register.html", form=form)

CSRF 说明

# Flask-WTF 默认开启 CSRF
# 必须配置 SECRET_KEY
# 模板必须 {{ form.hidden_tag() }}
# 或单独 token：<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">

# API 场景可关闭
# app.config["WTF_CSRF_ENABLED"] = False

收藏功能与后台管理

掌握 Flask 收藏功能与后台管理：完整业务功能。

收藏模型

# models.py
class Favorite(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey("users.id"))
    article_id = db.Column(db.Integer, db.ForeignKey("articles.id"))
    created_at = db.Column(db.DateTime, default=datetime.utcnow)

    # 防止重复收藏
    __table_args__ = (
        db.UniqueConstraint("user_id", "article_id",
                            name="unique_favorite"),
    )

收藏接口

from flask_login import login_required, current_user

@app.route("/article//favorite", methods=["POST"])
@login_required
def toggle_favorite(aid):
    article = Article.query.get_or_404(aid)

    fav = Favorite.query.filter_by(
        user_id=current_user.id,
        article_id=aid).first()

    if fav:
        db.session.delete(fav)
        db.session.commit()
        return {"favorited": False}
    else:
        db.session.add(Favorite(
            user_id=current_user.id,
            article_id=aid))
        db.session.commit()
        return {"favorited": True}

@app.route("/favorites")
@login_required
def my_favorites():
    articles = Article.query.join(Favorite).filter(
        Favorite.user_id == current_user.id).all()
    return render_template("favorites.html", articles=articles)

后台管理（admin 蓝图）

# views/admin.py
from flask import Blueprint, render_template, request, redirect
from flask_login import login_required, current_user
from functools import wraps

admin_bp = Blueprint("admin", __name__, url_prefix="/admin")

def admin_required(f):
    @wraps(f)
    def wrapper(*args, **kwargs):
        if not current_user.is_admin:
            return "无权访问", 403
        return f(*args, **kwargs)
    return wrapper

@admin_bp.route("/")
@login_required
@admin_required
def dashboard():
    stats = {
        "articles": Article.query.count(),
        "users": User.query.count(),
        "favorites": Favorite.query.count(),
    }
    return render_template("admin/dashboard.html", stats=stats)

@admin_bp.route("/articles")
@login_required
@admin_required
def article_list():
    articles = Article.query.order_by(
        Article.created_at.desc()).all()
    return render_template("admin/articles.html",
                           articles=articles)

@admin_bp.route("/articles//delete", methods=["POST"])
@login_required
@admin_required
def delete_article(aid):
    article = Article.query.get_or_404(aid)
    db.session.delete(article)
    db.session.commit()
    flash("文章已删除", "success")
    return redirect(url_for("admin.article_list"))

打包上线

掌握 Flask 打包上线：完整部署流程。

上线前检查

• DEBUG=False
• SECRET_KEY 从环境变量读取
• 生产数据库配置
• 依赖锁定（requirements.txt）
• 数据库迁移已应用

准备部署文件

project/
├── app.py / run.py
├── config.py
├── requirements.txt
├── .env              # 生产环境变量（不入库）
└── migrations/       # 迁移文件

pip freeze > requirements.txt

Gunicorn 启动

pip install gunicorn

# 启动（4 worker，2 线程）
gunicorn -w 4 --threads 2 -b 127.0.0.1:8000 "app:create_app()"

# 或应用文件方式
gunicorn -w 4 -b 127.0.0.1:8000 wsgi:app

wsgi.py

# wsgi.py（Gunicorn 入口）
from app import create_app

app = create_app()

Nginx 配置

server {
    listen 80;
    server_name www.laowang.click;

    location /static/ {
        alias /www/wwwroot/app/static/;
        expires 7d;
    }

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}

宝塔面板部署

1. 网站 → Python 项目 → 创建项目
2. 选择 Python 版本，上传代码
3. 安装依赖：pip install -r requirements.txt
4. 执行迁移：flask db upgrade
5. 启动命令：gunicorn -w 4 -b 127.0.0.1:8000 wsgi:app
6. 打开外网映射（自动 Nginx 反代）
7. 申请 SSL 证书

上线验证

curl -I https://www.laowang.click/       # 200
curl http://127.0.0.1:8000/api/health    # 健康检查
tail -f /var/log/nginx/error.log          # 查错误

https://palletsprojects.com/projects/flask/

Flask 官方资源：Flask 官方文档与社区。

官方文档

项目托管

• GitHub：pallets/flask
• PyPI：pypi.org/project/flask
• 中文社区：flask 相关博客与教程

核心扩展（Pallets 官方）

学习资源

• Flask Mega-Tutorial（Miguel Grinberg）
• 《Flask Web 开发》（Miguel Grinberg 著）
• 官方文档的 Tutorial 部分
• CodeNest 本站 FastAPI 教程对照学习