请求与响应

在 Web 应用中，HTTP 协议的核心就是请求与响应的交互。客户端发送请求，服务器处理后返回响应。Flask 提供了强大而灵活的工具来处理这一过程，本章将深入探讨如何正确、高效地处理 HTTP 请求和构建响应。

理解请求响应流程

当用户在浏览器中访问一个 URL 时，浏览器会向服务器发送一个 HTTP 请求。这个请求包含了大量信息：请求方法（GET、POST 等）、URL 路径、查询参数、请求头、Cookie、请求体等。服务器接收到请求后，需要解析这些信息，执行相应的业务逻辑，然后返回一个 HTTP 响应。

Flask 将这个过程封装得非常简洁：

from flask import Flask, request

app = Flask(__name__)

@app.route('/hello')
def hello():
    # request 对象自动包含了当前请求的所有信息
    name = request.args.get('name', 'World')
    # 返回值会自动转换为 HTTP 响应
    return f'Hello, {name}!'

在这个简单的例子中，request 对象是我们与请求数据交互的核心入口，而视图函数的返回值则自动转换为响应。理解这两个概念是掌握 Flask 的关键。

请求对象（Request）

请求对象的工作原理

request 对象是 Flask 中最常用的全局对象之一。你可能会疑惑：如果在多线程环境中多个请求同时到达，全局的 request 对象如何保证线程安全？

Flask 使用了**上下文本地变量（Context Locals）**技术。简单来说，request 实际上是一个代理对象，它会自动指向当前线程正在处理的请求。每个请求都有自己独立的上下文，互不干扰。

from flask import Flask, request

app = Flask(__name__)

@app.route('/info')
def info():
    # 在这个视图函数中，request 指向当前请求
    # 即使同时有其他请求在处理，这里也是安全的
    return f'请求路径: {request.path}'

# 在请求上下文之外访问 request 会出错
# print(request.path)  # RuntimeError: Working outside of request context

获取请求数据

Web 应用需要从请求中获取各种数据，不同类型的数据存储在不同的属性中。

URL 查询参数

URL 查询参数是 URL 中 ? 后面的部分，常用于 GET 请求传递参数：

from flask import request

@app.route('/search')
def search():
    # URL: /search?q=flask&page=2
    
    # 使用 get 方法，不存在时返回 None 或默认值（推荐）
    query = request.args.get('q', '')           # 'flask'
    page = request.args.get('page', 1, type=int) # 2
    
    # 使用字典索引，不存在时抛出 KeyError
    # query = request.args['q']  # 如果 q 不存在会报错
    
    return f'搜索: {query}, 第 {page} 页'

为什么推荐使用 get 方法？

用户可能修改 URL 中的参数，如果使用字典索引访问不存在的键，会抛出 KeyError，Flask 会返回 400 错误页面。这对于用户体验来说是不友好的。使用 get 方法可以优雅地处理缺失的参数。

表单数据

表单数据通常来自 POST 请求的请求体：

@app.route('/login', methods=['GET', 'POST'])
def login():
    if request.method == 'POST':
        # 获取表单数据
        username = request.form.get('username')
        password = request.form.get('password')
        remember = request.form.get('remember', 'off')  # 复选框未勾选时不发送
        
        # 处理登录逻辑...
        return f'用户 {username} 登录成功'
    
    # GET 请求显示登录表单
    return '''
        <form method="post">
            <p><input type="text" name="username" placeholder="用户名"></p>
            <p><input type="password" name="password" placeholder="密码"></p>
            <p><label><input type="checkbox" name="remember"> 记住我</label></p>
            <p><button type="submit">登录</button></p>
        </form>
    '''

JSON 数据

现代 API 通常使用 JSON 格式传递数据：

@app.route('/api/users', methods=['POST'])
def create_user():
    # 检查 Content-Type 是否为 application/json
    if not request.is_json:
        return {'error': 'Content-Type must be application/json'}, 415
    
    # 方式一：使用 get_json() 方法
    data = request.get_json()
    # 可以设置 silent=True 静默失败（返回 None 而不是报错）
    # data = request.get_json(silent=True)
    
    # 方式二：直接访问 json 属性
    # data = request.json
    
    username = data.get('username')
    email = data.get('email')
    
    # 创建用户...
    return {'id': 1, 'username': username, 'email': email}, 201

get_json() 和 json 属性的区别：

方式	说明
`request.get_json()`	方法调用，可以传递参数控制行为（如 `silent=True`、`force=True`）
`request.json`	属性访问，简洁但无法控制行为

文件上传

处理文件上传需要 HTML 表单设置 enctype="multipart/form-data"：

from werkzeug.utils import secure_filename
import os

app.config['UPLOAD_FOLDER'] = 'uploads'
app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024  # 限制 16MB

ALLOWED_EXTENSIONS = {'txt', 'pdf', 'png', 'jpg', 'jpeg', 'gif'}

def allowed_file(filename):
    """检查文件扩展名是否合法"""
    return '.' in filename and \
           filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS

@app.route('/upload', methods=['GET', 'POST'])
def upload_file():
    if request.method == 'POST':
        # 检查是否有文件
        if 'file' not in request.files:
            return '没有选择文件', 400
        
        file = request.files['file']
        
        # 检查是否选择了文件（浏览器可能发送空文件名）
        if file.filename == '':
            return '没有选择文件', 400
        
        # 检查文件类型
        if not allowed_file(file.filename):
            return '不支持的文件类型', 400
        
        # 安全处理文件名，防止路径遍历攻击
        filename = secure_filename(file.filename)
        
        # 保存文件
        os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
        file.save(os.path.join(app.config['UPLOAD_FOLDER'], filename))
        
        return f'文件 {filename} 上传成功'
    
    return '''
        <form method="post" enctype="multipart/form-data">
            <input type="file" name="file">
            <button type="submit">上传</button>
        </form>
    '''

为什么使用 secure_filename()？

用户上传的文件名可能包含恶意路径（如 ../../../etc/passwd），直接使用可能导致文件被保存到任意位置。secure_filename() 会移除路径信息，只保留安全的文件名：

from werkzeug.utils import secure_filename

secure_filename('../../../etc/passwd')  # 返回 'etc_passwd'
secure_filename('我的文档.pdf')         # 返回 '_.pdf'（非 ASCII 字符被替换）
secure_filename('hello world.txt')     # 返回 'hello_world.txt'

组合数据

有时表单同时包含普通字段和文件：

@app.route('/submit', methods=['POST'])
def submit():
    # 获取普通表单字段
    title = request.form.get('title')
    description = request.form.get('description')
    
    # 获取上传的文件
    if 'attachment' in request.files:
        file = request.files['attachment']
        # 处理文件...
    
    # 也可以遍历所有文件
    for name, file in request.files.items():
        print(f'文件字段: {name}, 文件名: {file.filename}')
    
    return '提交成功'

请求属性详解

request 对象包含大量有用的属性，下面分类介绍：

基本信息属性

@app.route('/info')
def request_info():
    info = {
        # HTTP 方法
        'method': request.method,           # 'GET', 'POST', 'PUT', 'DELETE' 等
        
        # URL 相关
        'url': request.url,                 # 完整 URL: http://localhost:5000/info?a=1
        'base_url': request.base_url,       # 不含查询参数: http://localhost:5000/info
        'url_root': request.url_root,       # 应用根 URL: http://localhost:5000/
        'path': request.path,               # 路径部分: /info
        'full_path': request.full_path,     # 路径加查询参数: /info?a=1
        'script_root': request.script_root, # 脚本根路径（用于子目录部署）
        'query_string': request.query_string, # 原始查询字符串（字节）: b'a=1'
        
        # 端点信息
        'endpoint': request.endpoint,       # 路由端点名: 'request_info'
        'blueprint': request.blueprint,     # 所属蓝图（如果有）: 'auth'
        
        # 视图参数
        'view_args': request.view_args,     # URL 中的变量: {'id': 123}
    }
    return info

请求头属性

@app.route('/headers')
def headers_info():
    # 方式一：字典式访问
    content_type = request.headers.get('Content-Type')
    user_agent = request.headers.get('User-Agent')
    authorization = request.headers.get('Authorization')
    
    # 方式二：遍历所有请求头
    all_headers = dict(request.headers)
    
    # 常用请求头的快捷属性
    info = {
        'user_agent': request.user_agent.string,  # User-Agent 字符串
        'content_type': request.content_type,      # Content-Type
        'content_length': request.content_length,  # Content-Length
        'mimetype': request.mimetype,              # MIME 类型
        'mimetype_params': request.mimetype_params, # MIME 类型参数
    }
    
    return info

客户端信息属性

@app.route('/client')
def client_info():
    info = {
        # 客户端 IP（可能受代理影响）
        'remote_addr': request.remote_addr,        # 直接连接的客户端 IP
        
        # 通过 X-Forwarded-For 头获取真实 IP（需要配置代理）
        # 'remote_addr' 可能是代理服务器的 IP
        # 需要使用 ProxyFix 中间件处理
    }
    return info

如果应用部署在反向代理后面，需要正确配置以获取真实 IP：

from werkzeug.middleware.proxy_fix import ProxyFix

# 信任一层代理
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_port=1)

@app.route('/cookies')
def cookies_info():
    # 获取所有 Cookie（字典）
    all_cookies = request.cookies
    
    # 获取单个 Cookie
    session_id = request.cookies.get('session_id')
    preferences = request.cookies.get('preferences', '{}')
    
    return {
        'session_id': session_id,
        'all_cookies': dict(all_cookies)
    }

其他有用属性

@app.route('/other')
def other_info():
    info = {
        # Scheme 和协议
        'scheme': request.scheme,              # 'http' 或 'https'
        'is_secure': request.is_secure,        # 是否 HTTPS
        
        # Host 信息
        'host': request.host,                  # Host 头，如 'localhost:5000'
        
        # 数据大小
        'content_length': request.content_length,  # 请求体长度
        'max_content_length': app.config.get('MAX_CONTENT_LENGTH'),
        
        # 数据流
        'data': request.data,                  # 原始请求体（字节）
        'stream': request.stream,              # 请求体流（用于大文件）
        
        # 其他
        'encoding': request.encoding,          # 请求编码
        'charset': request.charset,            # 字符集
    }
    return info

请求数据属性对比

属性	来源	Content-Type	用途
`request.args`	URL 查询参数	无关	GET 请求参数
`request.form`	请求体	`application/x-www-form-urlencoded` 或 `multipart/form-data`	表单数据
`request.files`	请求体	`multipart/form-data`	上传文件
`request.json`	请求体	`application/json`	JSON 数据
`request.data`	请求体	任意	原始字节
`request.values`	合并	无关	`args` + `form` 的合并（不推荐）
`request.get_json()`	请求体	`application/json`	解析 JSON（推荐）

注意：request.values 合并了 args 和 form，但可能导致参数覆盖问题，不推荐使用。应该明确知道数据来源。

响应对象（Response）

理解响应转换

Flask 视图函数的返回值会自动转换为响应对象。Flask 支持多种返回值类型：

from flask import jsonify, make_response

# 1. 字符串 - 自动转换为 HTML 响应
@app.route('/string')
def string_response():
    return 'Hello, World!'
    # 等价于：返回状态码 200，Content-Type: text/html

# 2. 字典/列表 - 自动转换为 JSON 响应（Flask 1.1+）
@app.route('/dict')
def dict_response():
    return {'message': 'success', 'data': [1, 2, 3]}
    # 自动调用 jsonify()，返回 Content-Type: application/json

@app.route('/list')
def list_response():
    return [1, 2, 3]
    # 返回 JSON 数组

# 3. 元组 - 包含额外信息
@app.route('/tuple1')
def tuple_response_1():
    return 'Not Found', 404
    # (内容, 状态码)

@app.route('/tuple2')
def tuple_response_2():
    return 'Created', 201, {'Location': '/resource/1'}
    # (内容, 状态码, 响应头)

@app.route('/tuple3')
def tuple_response_3():
    return {'error': 'Bad Request'}, {'X-Custom': 'value'}
    # (内容, 响应头)，状态码默认 200

# 4. Response 对象 - 完全控制
@app.route('/response')
def response_object():
    response = make_response('Custom Response')
    response.status_code = 200
    response.headers['X-Custom-Header'] = 'Custom Value'
    response.mimetype = 'text/plain'
    return response

响应转换规则

Flask 按以下顺序处理视图函数的返回值：

如果返回的是 Response 对象，直接使用
如果是字符串，创建一个包含该字符串的响应对象
如果是生成器或迭代器，作为流式响应处理
如果是字典或列表，调用 jsonify() 创建 JSON 响应
如果是元组，解析元组中的状态码和响应头
如果都不是，假设是 WSGI 应用并调用它

使用 make_response

当需要修改响应属性时，使用 make_response 创建响应对象：

from flask import make_response

@app.route('/custom')
def custom_response():
    # 方式一：包装现有返回值
    resp = make_response('Hello')
    resp.status_code = 200
    resp.headers['X-Custom'] = 'Value'
    resp.set_cookie('name', 'value')
    return resp
    
    # 方式二：直接创建
    # resp = make_response('Hello', 200, {'X-Custom': 'Value'})

@app.route('/template')
def template_response():
    html = render_template('page.html', title='My Page')
    resp = make_response(html)
    resp.headers['Cache-Control'] = 'public, max-age=300'
    return resp

JSON 响应

构建 API 时，JSON 是最常用的响应格式：

from flask import jsonify

@app.route('/api/users')
def list_users():
    users = [
        {'id': 1, 'name': 'Alice'},
        {'id': 2, 'name': 'Bob'},
    ]
    
    # 方式一：直接返回字典/列表（Flask 1.1+）
    return users
    
    # 方式二：使用 jsonify（更明确）
    # return jsonify(users)
    
    # 方式三：返回时指定状态码
    # return jsonify(users), 200

@app.route('/api/users/<int:id>')
def get_user(id):
    user = User.query.get_or_404(id)
    
    # 返回单个对象
    return jsonify({
        'id': user.id,
        'name': user.name,
        'email': user.email,
        'created_at': user.created_at.isoformat()
    })

@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.get_json()
    
    # 创建用户...
    user = User(**data)
    db.session.add(user)
    db.session.commit()
    
    # 返回 201 Created
    return jsonify({
        'id': user.id,
        'name': user.name
    }), 201

@app.route('/api/error')
def api_error():
    # 错误响应
    return jsonify({
        'error': 'ValidationError',
        'message': '用户名不能为空',
        'field': 'username'
    }), 400

流式响应

对于大量数据或实时数据，可以使用流式响应：

from flask import Response, stream_with_context
import csv
import io

@app.route('/stream')
def stream_response():
    def generate():
        """生成器函数，逐步产生数据"""
        yield '开始\n'
        for i in range(10):
            yield f'数据 {i}\n'
            time.sleep(0.5)  # 模拟耗时操作
        yield '结束\n'
    
    # 方式一：使用 Response
    return Response(generate(), mimetype='text/plain')
    
    # 方式二：使用 stream_template（模板流式渲染）
    # return stream_template('stream.html', data=generate())

@app.route('/download/large-csv')
def download_large_csv():
    """生成大型 CSV 文件下载"""
    def generate_csv():
        # 生成 CSV 头
        yield 'id,name,email\n'
        
        # 逐行生成数据
        for user in User.query.yield_per(1000):  # 批量加载
            yield f'{user.id},{user.name},{user.email}\n'
    
    response = Response(
        stream_with_context(generate_csv()),
        mimetype='text/csv'
    )
    response.headers['Content-Disposition'] = 'attachment; filename=users.csv'
    return response

@app.route('/stream/template')
def stream_template():
    """流式渲染模板"""
    def generate_data():
        for i in range(1000):
            yield {'id': i, 'value': f'Item {i}'}
    
    return stream_template('data.html', items=generate_data())

注意：使用流式响应时，after_request 钩子会在流开始时就执行，而不是在流结束后。

文件下载

提供文件下载有几种方式：

from flask import send_file, send_from_directory

@app.route('/download/<filename>')
def download_file(filename):
    # 方式一：send_file - 发送文件
    return send_file(
        f'/path/to/files/{filename}',
        as_attachment=True,           # 作为附件下载
        download_name='report.pdf'    # 下载时的文件名
    )

@app.route('/uploads/<filename>')
def uploaded_file(filename):
    # 方式二：send_from_directory - 安全地从目录发送
    # 会自动处理路径遍历攻击
    return send_from_directory(
        app.config['UPLOAD_FOLDER'],
        filename,
        as_attachment=True
    )

@app.route('/download/generated')
def download_generated():
    # 方式三：动态生成内容后下载
    from io import BytesIO
    
    # 生成内容
    buffer = BytesIO()
    buffer.write(b'Hello, World!')
    buffer.seek(0)
    
    return send_file(
        buffer,
        as_attachment=True,
        download_name='hello.txt',
        mimetype='text/plain'
    )

重定向

使用 redirect() 将用户重定向到其他 URL：

from flask import redirect, url_for

@app.route('/old-page')
def old_page():
    # 临时重定向（302）
    return redirect(url_for('new_page'))

@app.route('/permanent')
def permanent_redirect():
    # 永久重定向（301）
    return redirect(url_for('new_page'), code=301)

@app.route('/external')
def external_redirect():
    # 重定向到外部 URL
    return redirect('https://example.com')

@app.route('/login-required')
def login_required_page():
    # 重定向到登录页，并保存原 URL
    return redirect(url_for('login', next=request.url))

@app.route('/login')
def login():
    next_url = request.args.get('next')
    if next_url:
        return redirect(next_url)
    return redirect(url_for('index'))

中止请求

使用 abort() 立即中止请求并返回错误：

from flask import abort

@app.route('/user/<int:id>')
def get_user(id):
    user = User.query.get(id)
    if user is None:
        abort(404)  # 返回 404 错误
    return render_template('user.html', user=user)

@app.route('/admin')
def admin():
    if not current_user.is_admin:
        abort(403)  # 返回 403 Forbidden
    return 'Admin Panel'

@app.route('/api/user/<int:id>')
def api_get_user(id):
    user = User.query.get(id)
    if user is None:
        # API 返回 JSON 错误
        abort(404, description='用户不存在')

# 自定义错误响应
@app.errorhandler(404)
def not_found(error):
    if request.path.startswith('/api/'):
        return jsonify({'error': 'Not found', 'message': error.description}), 404
    return render_template('404.html'), 404

Cookie 是存储在客户端的小型数据片段，每次请求都会自动发送到服务器。

from flask import make_response

@app.route('/set-cookie')
def set_cookie():
    resp = make_response('Cookie 已设置')
    
    # 基本设置
    resp.set_cookie('username', 'john')
    
    # 完整参数
    resp.set_cookie(
        'session_id',                  # Cookie 名称
        'abc123',                      # Cookie 值
        max_age=3600,                  # 存活时间（秒）
        expires=None,                  # 过期时间（datetime）
        path='/',                      # 有效路径
        domain=None,                   # 有效域名
        secure=True,                   # 仅 HTTPS 传输
        httponly=True,                 # 禁止 JavaScript 访问
        samesite='Lax'                 # 同站策略：Strict, Lax, None
    )
    
    return resp

@app.route('/get-cookie')
def get_cookie():
    # 获取单个 Cookie
    username = request.cookies.get('username', 'Guest')
    
    # 获取所有 Cookie
    all_cookies = dict(request.cookies)
    
    return f'用户名: {username}'

@app.route('/delete-cookie')
def delete_cookie():
    resp = make_response('Cookie 已删除')
    resp.delete_cookie('username')
    return resp

# 生产环境 Cookie 配置
app.config.update(
    SESSION_COOKIE_SECURE=True,      # 仅 HTTPS
    SESSION_COOKIE_HTTPONLY=True,    # 防止 XSS
    SESSION_COOKIE_SAMESITE='Lax',   # 防止 CSRF
)

@app.route('/safe-cookie')
def safe_cookie():
    resp = make_response('安全 Cookie')
    resp.set_cookie(
        'auth_token',
        generate_secure_token(),
        secure=True,      # 生产环境必须
        httponly=True,    # 防止 JavaScript 窃取
        samesite='Strict' # 严格同站策略
    )
    return resp

安全最佳实践：

敏感 Cookie 必须设置 secure=True（仅 HTTPS）
设置 httponly=True 防止 XSS 攻击窃取 Cookie
设置 samesite 防止 CSRF 攻击
不要在 Cookie 中存储敏感信息（可被用户查看）
对 Cookie 值进行签名或加密

会话管理（Session）

Session 是 Flask 提供的高级会话管理机制，它基于 Cookie 实现，但更加安全和方便。

工作原理

Flask 的 Session 使用签名 Cookie方式存储：

服务端将数据序列化后存入 Cookie
使用密钥对 Cookie 进行签名（防止篡改）
客户端每次请求自动发送 Cookie
服务端验证签名后反序列化数据

这意味着用户可以看到 Session 内容，但无法修改（除非知道密钥）。因此不要在 Session 中存储敏感信息。

基本使用

from flask import session

# 必须设置密钥
app.secret_key = 'your-secret-key-here'

@app.route('/')
def index():
    # 检查是否登录
    if 'user_id' in session:
        return f'已登录，用户 ID: {session["user_id"]}'
    return '未登录'

@app.route('/login', methods=['POST'])
def login():
    username = request.form.get('username')
    password = request.form.get('password')
    
    # 验证用户...
    if verify_user(username, password):
        # 设置 Session
        session['user_id'] = get_user_id(username)
        session['username'] = username
        
        # 持久化 Session
        session.permanent = True
        
        return '登录成功'
    return '用户名或密码错误', 401

@app.route('/logout')
def logout():
    # 清除特定项
    session.pop('user_id', None)
    
    # 清除所有
    session.clear()
    
    return '已退出登录'

Session 配置

from datetime import timedelta

app.config.update(
    # 密钥（必须）
    SECRET_KEY='your-secret-key',
    
    # Session Cookie 名称
    SESSION_COOKIE_NAME='session',
    
    # Session 有效期（仅当 session.permanent = True 时生效）
    PERMANENT_SESSION_LIFETIME=timedelta(days=7),
    
    # Cookie 安全设置
    SESSION_COOKIE_SECURE=True,      # 仅 HTTPS
    SESSION_COOKIE_HTTPONLY=True,    # 禁止 JS 访问
    SESSION_COOKIE_SAMESITE='Lax',   # 同站策略
    
    # Cookie 路径和域名
    SESSION_COOKIE_PATH='/',
    SESSION_COOKIE_DOMAIN=None,
)

持久化 Session

from datetime import timedelta
from flask import session

# 设置持久 Session
@app.route('/remember-login')
def remember_login():
    session['user_id'] = 123
    session.permanent = True  # 使用 PERMANENT_SESSION_LIFETIME
    return '登录成功，记住登录状态'

# 临时 Session（关闭浏览器后失效）
@app.route('/temp-login')
def temp_login():
    session['user_id'] = 123
    # session.permanent 默认为 False
    return '登录成功，关闭浏览器后失效'

Session 注意事项

# 1. 不要存储敏感信息
session['credit_card'] = '1234-5678-9012-3456'  # 危险！用户可以看到

# 2. 不要存储大数据
session['large_data'] = huge_json_data  # Cookie 大小有限制（约 4KB）

# 3. 使用密钥保护
import secrets
app.secret_key = secrets.token_hex(32)  # 使用强随机密钥

# 4. 服务器端 Session（推荐用于敏感数据）
# 使用 Flask-Session 扩展将 Session 存储在服务器
from flask_session import Session
app.config['SESSION_TYPE'] = 'redis'
Session(app)

消息闪现

消息闪现是一种跨请求传递消息的机制，常用于显示操作结果：

from flask import flash, get_flashed_messages

@app.route('/settings', methods=['GET', 'POST'])
def settings():
    if request.method == 'POST':
        # 保存设置...
        flash('设置已保存', 'success')
        return redirect(url_for('settings'))
    
    return render_template('settings.html')

# 在模板中使用

模板中显示闪现消息：

<!-- 获取所有消息 -->
{% with messages = get_flashed_messages(with_categories=true) %}
  {% if messages %}
    <div class="flash-messages">
      {% for category, message in messages %}
        <div class="alert alert-{{ category }}">
          {{ message }}
        </div>
      {% endfor %}
    </div>
  {% endif %}
{% endwith %}

<!-- 只获取特定类别 -->
{% with errors = get_flashed_messages(category_filter=['error']) %}
  {% for error in errors %}
    <div class="alert alert-danger">{{ error }}</div>
  {% endfor %}
{% endwith %}

请求钩子

请求钩子允许你在请求处理的不同阶段执行代码，这是 Flask 中非常强大的功能。

钩子类型

钩子	执行时机	用途
`before_request`	每个请求前	身份验证、加载用户
`before_first_request`	第一个请求前	初始化操作
`after_request`	每个请求后（无异常）	添加响应头、处理响应
`teardown_request`	每个请求结束	清理资源
`teardown_appcontext`	应用上下文结束	清理应用级资源

before_request

在请求处理前执行，如果返回非 None 值，请求会被中断：

from flask import g, redirect, url_for

@app.before_request
def load_logged_in_user():
    """在每个请求前加载当前用户"""
    user_id = session.get('user_id')
    
    if user_id is None:
        g.user = None
    else:
        g.user = User.query.get(user_id)

@app.before_request
def check_login():
    """检查登录状态，保护需要登录的路由"""
    # 允许访问登录相关路由
    if request.endpoint in ('login', 'register', 'static'):
        return
    
    if g.user is None:
        return redirect(url_for('login'))

# 在蓝图中使用
@api_bp.before_request
def api_auth():
    """API 蓝图的前置检查"""
    if not request.headers.get('X-API-Key'):
        return {'error': 'API Key required'}, 401

after_request

在请求处理后执行，必须返回响应对象：

@app.after_request
def add_security_headers(response):
    """添加安全响应头"""
    response.headers['X-Content-Type-Options'] = 'nosniff'
    response.headers['X-Frame-Options'] = 'SAMEORIGIN'
    response.headers['X-XSS-Protection'] = '1; mode=block'
    return response

@app.after_request
def add_cors_headers(response):
    """添加 CORS 头"""
    response.headers['Access-Control-Allow-Origin'] = '*'
    response.headers['Access-Control-Allow-Methods'] = 'GET, POST, PUT, DELETE'
    response.headers['Access-Control-Allow-Headers'] = 'Content-Type'
    return response

# 注意：after_request 不处理异常，如果请求过程中抛出异常，after_request 不会执行

teardown_request

在请求结束时执行，无论是否发生异常：

@app.teardown_request
def teardown_request(exception=None):
    """请求结束时的清理工作"""
    if exception:
        app.logger.error(f'请求异常: {exception}')
    
    # 关闭数据库连接等清理工作
    db = getattr(g, 'db', None)
    if db is not None:
        db.close()

# teardown_request 函数必须接受一个参数（异常对象，可能为 None）
# 不应该有返回值（返回值会被忽略）

钩子执行顺序

# 应用级钩子执行顺序：
# 1. before_first_request（仅第一次请求）
# 2. before_request（应用级）
# 3. before_request（蓝图级）
# 4. 视图函数
# 5. after_request（蓝图级）
# 6. after_request（应用级）
# 7. teardown_request（蓝图级）
# 8. teardown_request（应用级）

app_bp = Blueprint('app', __name__)

@app_bp.before_request
def bp_before():
    print('蓝图 before_request')

@app_bp.after_request
def bp_after(response):
    print('蓝图 after_request')
    return response

@app_bp.teardown_request
def bp_teardown(exception):
    print('蓝图 teardown_request')

使用 g 对象

g 对象用于在请求期间存储和传递数据：

from flask import g

@app.before_request
def before_request():
    # 存储请求级数据
    g.start_time = time.time()
    g.db = connect_db()

@app.after_request
def after_request(response):
    # 访问请求数据
    elapsed = time.time() - g.start_time
    response.headers['X-Response-Time'] = f'{elapsed:.3f}s'
    return response

@app.teardown_request
def teardown_request(exception):
    # 清理资源
    db = getattr(g, 'db', None)
    if db is not None:
        db.close()

@app.route('/data')
def get_data():
    # 在视图函数中访问
    db = g.db
    return {'data': db.query()}

g 对象与 session 的区别：

特性	g	session
生命周期	单个请求	跨请求持久化
存储位置	服务端内存	客户端 Cookie
数据可见性	仅当前请求	跨请求可见
典型用途	数据库连接、请求上下文	用户登录状态

小结

本章深入讲解了 Flask 的请求与响应处理：

请求对象：理解上下文本地变量、获取不同类型的数据、请求属性详解
响应对象：响应转换规则、JSON 响应、流式响应、文件下载
Cookie 操作：设置、读取、删除和安全配置
会话管理：Session 的工作原理、配置和使用
消息闪现：跨请求传递消息
请求钩子：在不同阶段拦截和处理请求

掌握这些知识，你就能构建出功能完善、安全可靠的 Web 应用。

理解请求响应流程​

请求对象（Request）​

请求对象的工作原理​

获取请求数据​

URL 查询参数​

表单数据​

JSON 数据​

文件上传​

组合数据​

请求属性详解​

基本信息属性​

请求头属性​

客户端信息属性​

Cookie 属性​

其他有用属性​

请求数据属性对比​

响应对象（Response）​

理解响应转换​

响应转换规则​

使用 make_response​

JSON 响应​

流式响应​

文件下载​

重定向​

中止请求​

Cookie 操作​

设置 Cookie​

读取 Cookie​

删除 Cookie​

Cookie 安全注意事项​

会话管理（Session）​

工作原理​

基本使用​

Session 配置​

持久化 Session​

Session 注意事项​

消息闪现​

请求钩子​

钩子类型​

before_request​

after_request​

teardown_request​

钩子执行顺序​

使用 g 对象​

小结​

参考资料​