教程 / Python3 / Flask 请求生命周期

Flask 请求生命周期

Flask 请求生命周期 理解 Flask 请求的完整生命周期,是写好 Flask 应用的关键。 本章从源码层面拆解一个 HTTP 请求从进入服务器到返回响应的全过程,帮你建立清晰的执行模型。 什么是请求生命周期 请求生

理解 Flask 请求的完整生命周期,是写好 Flask 应用的关键。

本章从源码层面拆解一个 HTTP 请求从进入服务器到返回响应的全过程,帮你建立清晰的执行模型。


什么是请求生命周期

请求生命周期(Request Lifecycle)指的是一个 HTTP 请求在 Flask 内部经历的完整处理过程。

它包含以下阶段:接收请求 → 创建上下文 → 预处理 → 路由分发 → 视图执行 → 响应构建 → 清理上下文 → 返回响应。

了解这个过程,你就能正确回答以下问题:

  • g 对象的数据什么时候可用?什么时候被清理?
  • before_request 和 after_request 的执行顺序是什么?
  • 异常发生时会依次经过哪些处理环节?
  • 为什么在视图函数外使用 request 会报错?

完整流程图

下图展示了 Flask 从接收请求到返回响应的完整调用链:


分步详解

第一阶段:WSGI 入口

当 HTTP 请求到达服务器时,WSGI 服务器(如 Gunicorn、Werkzeug 开发服务器)会调用 Flask 应用实例。

Flask 实现了 __call__ 方法,使其成为一个合法的 WSGI 应用:

实例

# Flask 源码简化版(实际位置:flask/app.py)
# WSGI 服务器会这样调用你的应用:
# response = app(environ, start_response)

class Flask(App):
    def __call__(self, environ, start_response):
        """WSGI 服务器调用的入口点"""
        return self.wsgi_app(environ, start_response)

    def wsgi_app(self, environ, start_response):
        """真正的 WSGI 处理逻辑"""
        # 第一步:根据 environ 创建请求上下文
        ctx = self.request_context(environ)
        try:
            ctx.push()                    # 推入上下文栈
            response = self.full_dispatch_request(ctx)  # 完整调度
        except Exception as e:
            response = self.handle_exception(ctx, e)    # 异常兜底
        finally:
            ctx.pop(error)               # 弹出上下文,执行清理
        return response(environ, start_response)

Flask 把入口点设计为调用 wsgi_app 方法,而不是直接在 __call__ 里写逻辑。这样你可以通过替换 app.wsgi_app 来插入中间件,而不影响对 app 实例本身的调用。

第二阶段:创建上下文

request_context(environ) 根据 WSGI 环境变量创建一个 AppContext 对象。

这个对象包含了当前请求的所有信息,当它被推入栈中后,request、session、g、current_app 这些全局代理才开始指向当前请求的数据。

实例

# 理解上下文的简单实验
from flask import Flask, request, current_app

app = Flask(__name__)

with app.app_context():
    # 只有应用上下文——current_app 可用,request 不可用
    print(current_app.name)  # ✓ 正常
    # print(request.method)  # ✗ 报错:Working outside of request context

with app.test_request_context("/hello?name=runoob"):
    # 请求上下文也包含了应用上下文——两者都可用
    print(current_app.name)     # ✓ 输出 app 名称
    print(request.path)         # ✓ 输出 /hello
    print(request.args["name"]) # ✓ 输出 runoob
上下文类型 何时创建 何时销毁 可用的全局变量
应用上下文 请求到达时 / 手动 with app.app_context() 请求结束 / with 块退出 current_app, g
请求上下文 请求到达时 / 手动 with app.test_request_context() 请求结束 / with 块退出 request, session(含应用上下文的所有变量)

初学者最容易犯的错误就是在模块级别直接访问 request。记住:request 只有在请求上下文中才存在。你只能在视图函数内部(或它调用的函数里)访问它。

第三阶段:preprocess_request(预处理)

上下文就绪后,full_dispatch_request 首先调用 preprocess_request。

这个阶段会执行所有用 @app.before_request 注册的函数。

如果任意一个 before_request 函数返回了非 None 的值,Flask 会把它当作最终响应,跳过后续的视图函数执行。

实例

from flask import Flask, request, session, redirect, url_for

app = Flask(__name__)
app.secret_key = "dev-secret"

@app.before_request
def check_authentication():
    """在每次请求前检查用户是否已登录"""
    # 跳过登录和注册页面的认证检查
    if request.endpoint in ("login", "register", "static"):
        return None  # None 表示继续正常流程

    # 如果未登录,直接返回重定向响应(跳过视图函数)
    if "username" not in session:
        return redirect(url_for("login"))

    # 用户已登录,正常继续
    return None

@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        session["username"] = request.form["username"]
        return redirect(url_for("dashboard"))
    return '<form method="post"><input name="username"><input type="submit"></form>'

@app.route("/register")
def register():
    return "注册页面(无需登录)"

@app.route("/dashboard")
def dashboard():
    return f"欢迎 {session['username']} 来到 RUNOOB 控制台"

第四阶段:dispatch_request(路由分发)

预处理完成后,Flask 进入核心的分发阶段:

  1. 匹配 URL 到对应的路由规则(Rule)
  2. 提取 URL 中的变量参数(view_args)
  3. 调用对应的视图函数,传入提取的参数

实例

# 源码简化版:dispatch_request 的工作流程
def dispatch_request(self, ctx):
    req = ctx.request
    # 获取匹配到的路由规则(Rule 对象)
    rule = req.url_rule
    # 获取 URL 中的变量参数。例如 /post/42 → {"post_id": 42}
    view_args = req.view_args
    # 从 view_functions 字典中找到对应的函数并调用
    return self.view_functions[rule.endpoint](**view_args)

匹配失败时会抛出 RoutingException,最终由错误处理器捕获并返回 404。

第五阶段:视图函数执行

这是你编写业务逻辑的地方。Flask 调用你的视图函数并获取返回值。

返回值可以是多种类型——字符串、字典、元组、Response 对象——这些都会在下一阶段被统一处理。

第六阶段:finalize_request(响应构建)

视图函数执行完毕后,finalize_request 负责将返回值转换为标准的 Response 对象:

实例

# 源码简化版:finalize_request 的工作流程
def finalize_request(self, ctx, rv):
    # 1. 调用 make_response 将各种类型的返回值转为 Response 对象
    response = self.make_response(rv)

    # 2. 通过 process_response 执行 after_request 钩子
    response = self.process_response(ctx, response)

    # 3. 发送信号:request_finished
    request_finished.send(self, response=response)

    return response

make_response 的转换规则:

返回值类型 转换方式
str / bytes 作为响应体,Content-Type: text/html
dict / list 调用 jsonify(),Content-Type: application/json
tuple 按 (body, status) 或 (body, headers) 解析
Response 对象 直接使用
iterator / generator 作为流式响应

第七阶段:after_request 钩子

process_response 在执行 make_response 之后,依次调用所有 @app.after_request 注册的函数。

每个 after_request 函数接收并返回 Response 对象,可以修改响应头、添加 Cookie 等。

实例

# after_request 典型用法:为每个响应添加安全头
@app.after_request
def add_security_headers(response):
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    return response  # 必须返回 response 对象

# 也常用于添加 CORS 头
@app.after_request
def add_cors_headers(response):
    response.headers["Access-Control-Allow-Origin"] = "*"
    response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE"
    return response

第八阶段:上下文清理

响应发送后,finally 块中的 ctx.pop() 执行上下文清理:

  1. 调用所有 teardown_request 注册的函数(无论请求成功或失败都会执行)
  2. 调用所有 teardown_appcontext 注册的函数
  3. 发送 request_tearing_down 和 appcontext_tearing_down 信号
  4. 释放 g 对象中存储的所有数据

实例

# teardown 典型用法:关闭数据库连接
@app.teardown_appcontext
def close_database_connection(error):
    """即使请求出错,也确保数据库连接被关闭"""
    db = g.pop("db", None)
    if db is not None:
        db.close()
    # error 参数:如果请求过程中有异常,这里会接收到
    if error:
        app.logger.warning(f"请求异常,已关闭数据库连接: {error}")

# teardown 与 after_request 的区别
@app.after_request
def after(response):
    # after_request 在正常流程中执行,异常时不执行
    # 用于修改成功的响应
    return response

@app.teardown_request
def teardown(error):
    # teardown 始终执行——无论请求成功还是失败
    # 用于清理资源(关闭文件、释放锁等)
    pass

异常处理流程

当视图函数抛出异常时,Flask 有一条专门的异常处理路径:

  1. full_dispatch_request 捕获异常
  2. 调用 handle_user_exception 判断异常类型
  3. HTTP 异常(如 abort(404))→ 查找对应的 errorhandler
  4. 普通异常 → 查找异常类的 errorhandler
  5. 找不到处理器 → 调用 handle_exception,返回 500 错误
  6. 无论异常与否,最终都会经过 finalize_request 构建响应

异常路径中的关键点:after_request 钩子对错误响应也会执行(从 Flask 1.1 开始),但 before_request 返回的非 None 值会短路整个流程,后面的视图函数和钩子都不会执行。


钩子函数速查表

钩子 注册方式 执行时机 典型用途
before_request @app.before_request 路由分发之前 登录检查、请求日志、权限验证
before_first_request @app.before_request(已废弃) 第一个请求前(仅一次) 初始化操作(建议改用工厂模式)
after_request @app.after_request 响应构建之后,发送之前 添加响应头、CORS、日志
teardown_request @app.teardown_request 请求上下文弹出时(始终执行) 释放资源、关闭文件
teardown_appcontext @app.teardown_appcontext 应用上下文弹出时(始终执行) 关闭数据库连接、清理缓存
errorhandler @app.errorhandler(code) 对应错误码/异常发生时 自定义错误页面

信号事件

Flask 使用 blinker 库在关键节点发送信号:

实例

from flask import Flask, signals

app = Flask(__name__)

# 连接信号——在请求到达时记录
@signals.request_started.connect_via(app)
def log_request_start(sender, **extra):
    print("请求开始")

# 连接信号——在请求完成时记录
@signals.request_finished.connect_via(app)
def log_request_end(sender, response, **extra):
    print(f"请求结束,响应状态码: {response.status_code}")

# 连接信号——捕获异常
@signals.got_request_exception.connect_via(app)
def log_exception(sender, exception, **extra):
    print(f"请求中发生异常: {exception}")
信号 触发时机
request_started 请求上下文建立后,预处理之前
request_finished 响应构建完成后,发送之前
got_request_exception 请求处理过程中发生异常时
request_tearing_down 请求上下文销毁时
appcontext_pushed 应用上下文推入栈时
appcontext_popped 应用上下文弹出栈时
message_flashed 调用 flash() 时
template_rendered 模板渲染成功时

关键要点总结

核心规则:一个完整的请求生命周期中,before_request → 视图函数 → after_request → teardown 这四个节点一定会按顺序执行(teardown 始终执行,无论前三个是否出错)。

理解请求生命周期后,你在写 Flask 应用时就知道:认证逻辑放 before_request,业务逻辑放视图函数,响应头修改放 after_request,资源清理放 teardown。