「每个请求都要记一条日志」「每个响应都要加跨域头」「每次请求都要先看有没有登录」——这些事如果写在每个视图里，就是无休止的复制粘贴。本章教你用请求钩子把公共逻辑插到请求生命周期的固定点上，一次写好、全局生效；最后再补上一个服务端渲染时代的经典知识点——flash 消息闪现。
钩子让代码在请求生命周期的固定节点自动运行：请求进门先过 before_request（鉴权/日志/预置），出门前经 after_request（加头/算耗时），最后 teardown_request 收尾（无论成败都执行）。
import time
from flask import Flask, g, request, jsonify
app = Flask(__name__)
@app.before_request
def before_every_request():
# ① 给每个请求打上起始时间戳，配合 after_request 算耗时
g.start_time = time.perf_counter()
# ② 记录一条请求日志（真实项目会写进日志文件）
app.logger.info(">> %s %s", request.method, request.path)
# ③ 统一给 g 预置一个默认用户，视图里可以放心判断 g.user 是否存在
g.user = None
# ④ 返回 None = 放行，继续走视图函数
return None@app.after_request
def after_every_request(response):
# ① 计算本次请求耗时，写进自定义响应头（方便前端与运维排查）
if hasattr(g, "start_time"):
cost_ms = (time.perf_counter() - g.start_time) * 1000
response.headers["X-Response-Time"] = f"{cost_ms:.2f}ms"
# ② 统一加安全相关的响应头
response.headers["X-Content-Type-Options"] = "nosniff"
# ③ 必须把 response 返回出去，否则请求会报错
return responsefrom flask import Flask, g
from .models import db
app = Flask(__name__)
@app.teardown_request
def teardown(exception):
# ① 请求出错时把没提交的事务回滚，避免脏数据残留在连接里
if exception is not None:
db.session.rollback()
app.logger.error("请求异常：%s", exception)
# ② 关闭请求期间打开的临时资源（示意）
tmp = g.pop("tmp_file", None)
if tmp:
tmp.close()
# 注意：此时响应已经准备完毕，改不了状态码和响应体了
return None用 before_request 记开始时间、after_request 算耗时，就能给所有接口统一打日志（方法、路径、状态码、耗时），不必在每个视图里重复写计时代码。
import time
from flask import g, request, current_app
def register_middleware(app):
"""把请求日志钩子注册到应用上，在 create_app() 里调用一次即可。"""
@app.before_request
def _log_start():
g.start_time = time.perf_counter()
@app.after_request
def _log_finish(response):
cost_ms = (time.perf_counter() - g.get("start_time", time.perf_counter())) * 1000
current_app.logger.info(
"%s %s -> %s %.2fms",
request.method, # 请求方法
request.full_path, # 路径（含查询参数）
response.status_code, # 状态码
cost_ms, # 耗时
)
return responseCORS 头、响应包装这类「每个接口都要做的事」适合放在 after_request 钩子里统一处理。新增接口时无需重复配置，全站自动生效。
from flask import request
# 允许跨域访问的前端来源（精确写，不要用 *）
ALLOW_ORIGIN = "http://localhost:5173"
def register_middleware(app):
# ① 浏览器在发「非简单请求」前会先发一个 OPTIONS 预检请求，
# 直接放行并返回 204，不必进视图
@app.before_request
def _handle_preflight():
if request.method == "OPTIONS":
return "", 204
return None
# ② 在每一个响应上补上 CORS 头
@app.after_request
def _add_cors_headers(response):
response.headers["Access-Control-Allow-Origin"] = ALLOW_ORIGIN
response.headers["Access-Control-Allow-Methods"] = "GET, POST, PUT, DELETE, OPTIONS"
response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
response.headers["Access-Control-Max-Age"] = "3600" # 预检结果缓存 1 小时
return responseimport json
def register_wrapper(app):
@app.after_request
def _wrap_json(response):
# 只处理 JSON 响应，其它（图片、文件下载）原样放行
if not response.is_json:
return response
data = response.get_json()
# 已经是 {code, msg, data} 就不要再包一层
if isinstance(data, dict) and "code" in data and "msg" in data:
return response
response.set_data(json.dumps(
{"code": 200, "msg": "success", "data": data},
ensure_ascii=False,
))
response.content_type = "application/json; charset=utf-8"
return responsefrom flask import Blueprint, flash, redirect, render_template, request, url_for
from werkzeug.security import check_password_hash
from ..models import User
web_bp = Blueprint("web", __name__)
@web_bp.post("/login")
def login():
username = request.form.get("username", "")
password = request.form.get("password", "")
user = User.query.filter_by(username=username).first()
if user is None or not check_password_hash(user.password, password):
# 存一条「错误」类别的提示，随后重定向回登录页
flash("用户名或密码错误", "error")
return redirect(url_for("web.login_page"))
flash("登录成功，欢迎回来", "success")
return redirect(url_for("web.user_center"))
@web_bp.get("/login")
def login_page():
# 渲染模板时，页面里会自动取出并显示上面 flash 的消息
return render_template("login.html"){% with messages = get_flashed_messages(with_categories=true) %}
{% if messages %}
<ul class="flash-list">
{% for category, message in messages %}
<li class="flash flash-{{ category }}">{{ message }}</li>
{% endfor %}
</ul>
{% endif %}
{% endwith %}
<!-- 渲染出来的 HTML（category 是 success，所以多了 flash-success 类）：
<ul class="flash-list">
<li class="flash flash-success">登录成功，欢迎回来</li>
</ul> -->注册多个钩子时：before_request 按注册先后依次执行（FIFO），after_request 则是倒序执行（LIFO，像压栈弹栈）。记住「进门排队、出门倒序」就不会乱。
请求进来
│
├─ before_request #1 （注册顺序 →）
├─ before_request #2
│ 若其中某个返回了响应，就直接跳到 after_request，请求不进入视图
│
├─ 视图函数 view()
│ 若视图抛异常 → 交给 errorhandler → 再进 after_request
│
├─ after_request #2 （← 注册逆序）
├─ after_request #1
│
└─ teardown_request （无论成功 / 失败都会执行）# 第 1 题：简易限流
import time
from collections import defaultdict
from flask import g, request, jsonify
_hits = defaultdict(list) # {ip: [时间戳, ...]}
@app.before_request
def rate_limit():
now = time.time()
ip = request.remote_addr
# 只保留最近 60 秒内的记录
_hits[ip] = [t for t in _hits[ip] if now - t < 60]
if len(_hits[ip]) >= 60:
return jsonify(code=429, msg="请求过于频繁，请稍后再试", data=None)
_hits[ip].append(now)
return None
# 第 2 题：加自定义响应头
@app.after_request
def add_powered_by(response):
response.headers["X-Powered-By"] = "hiwen"
return response
# 为什么下载接口不会被包装钩子改坏：
# 因为包装钩子里有 if not response.is_json: return response
# 图片/CSV 的响应 is_json 为 False，会原样放行
# 第 3 题：红影赏析
ALLOW_ORIGIN = "http://localhost:5173"
def register_middleware(app):
@app.before_request
def _start():
g.start_time = time.perf_counter()
@app.after_request
def _cors_and_log(response):
# (a) 统一 CORS
response.headers["Access-Control-Allow-Origin"] = ALLOW_ORIGIN
response.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
# (b) 耗时 + 慢请求告警
cost_ms = (time.perf_counter() - g.start_time) * 1000
if cost_ms > 500:
app.logger.warning("慢请求 %s %s %.2fms",
request.method, request.path, cost_ms)
return response三题都指向同一个判断标准：「这件事是不是每个请求都要做？」 是——放钩子；不是——放视图。抓住这条，你就不会再纠结该写在哪了。