Skip to content

02-路由与请求处理

Python 3.11+

本章讲解 Flask 路由系统、动态 URL、请求对象和响应构建。


第一部分:@app.route 装饰器

1.1 实际场景

你正在开发一个博客网站,需要为首页、文章列表、文章详情等不同页面设置不同的 URL 路径。

问题:如何将 URL 路径映射到对应的处理函数?

1.2 概念说明

路由(Route)将 URL 映射到视图函数,@app.route() 装饰器用于注册路由。

python
from flask import Flask

app: Flask = Flask(__name__)


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


@app.route("/about")
def about() -> str:
    return "关于我们"


@app.route("/contact")
def contact() -> str:
    return "联系方式"

访问 URL:


第二部分:动态路由规则

2.1 实际场景

博客文章有不同 ID,如 /post/1/post/2,你不想为每个 ID 都写一个路由函数。

问题:如何用一个路由处理不同的 URL 参数?

2.2 URL 变量类型转换器

使用尖括号 <variable_name> 可以在 URL 中捕获动态值并传递给视图函数。

python
from flask import Flask

app: Flask = Flask(__name__)


@app.route("/user/<username>")
def show_user(username: str) -> str:
    """字符串类型(默认)"""
    return f"用户:{username}"


@app.route("/post/<int:post_id>")
def show_post(post_id: int) -> str:
    """整数类型"""
    return f"文章 ID: {post_id}"


@app.route("/price/<float:price>")
def show_price(price: float) -> str:
    """浮点数类型"""
    return f"价格:{price}"


@app.route("/path/<path:subpath>")
def show_path(subpath: str) -> str:
    """路径类型(可包含斜杠)"""
    return f"子路径:{subpath}"

示例:博客文章 URL

python
# /article/2027/03/09/my-first-post
@app.route("/article/<int:year>/<int:month>/<int:day>/<slug:title>")
def article(year: int, month: int, day: int, title: str) -> str:
    return f"{year}{month}{day}日 - {title}"

第三部分:URL 构建:url_for()

3.1 实际场景

你在代码中写了 /user/profile 这个 URL,后来路由规则改为 /account/profile,所有引用都要修改。

问题:如何避免硬编码 URL,让代码易于重构?

3.2 基本用法

url_for() 函数根据视图函数名生成 URL,避免硬编码 URL。

python
from flask import Flask, url_for

app: Flask = Flask(__name__)


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


@app.route("/user/<username>")
def user_profile(username: str) -> str:
    return f"{username}的主页"


# 生成 URL
print(url_for("index"))           # 输出:/
print(url_for("user_profile", username="张三"))  # 输出:/user/张三

url_for() 的优势:

┌─────────────────────────────────────────────────────────────┐
│              使用 url_for() 的优势                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   1. 易于重构                                               │
│      修改路由规则后,不需要修改所有引用该 URL 的地方         │
│                                                             │
│   2. 自动处理特殊字符                                        │
│      url_for('/user/张三') → /user/%E5%BC%A0%E4%B8%89       │
│                                                             │
│   3. 支持测试环境                                           │
│      测试时可以重定向到测试服务器                           │
│                                                             │
└─────────────────────────────────────────────────────────────┘

第四部分:重定向与 abort

4.1 实际场景

用户登录成功后需要跳转到首页,未登录用户访问管理页面需要返回 403 错误。

问题:如何实现页面跳转和错误响应?

4.2 示例代码

redirect() 用于跳转到其他 URL,abort() 用于立即返回错误状态码。

python
from flask import Flask, redirect, abort, url_for, request
from typing import Optional

app: Flask = Flask(__name__)


# 永久重定向
@app.route("/old-page")
def old_page() -> str:
    return redirect(url_for("new_page"), code=301)


# 临时重定向
@app.route("/new-page")
def new_page() -> str:
    return "这是新页面"


# 根据条件重定向
@app.route("/admin")
def admin() -> str:
    user: Optional[dict] = get_current_user()  # 假设有这个函数
    if not user:
        return redirect(url_for("login"))
    if not user.get("is_admin"):
        abort(403)  # 返回 403 Forbidden
    return "管理员页面"


# 自定义错误页面
@app.errorhandler(403)
def forbidden(error) -> tuple[str, int]:
    return "您没有权限访问此页面", 403


@app.errorhandler(404)
def not_found(error) -> tuple[str, int]:
    return "页面未找到", 404


def get_current_user() -> Optional[dict]:
    """模拟获取当前用户"""
    return None

第五部分:request 对象

5.1 实际场景

你需要知道访问者使用什么浏览器、从哪个 IP 地址访问、请求了什么 URL。

问题:如何获取 HTTP 请求的详细信息?

5.2 request 对象常用属性

Flask 的 request 对象封装了 HTTP 请求的所有信息。

python
from flask import Flask, request

app: Flask = Flask(__name__)


@app.route("/request-info")
def request_info() -> str:
    """展示 request 对象的常用属性"""

    # 请求方法
    method: str = request.method  # GET, POST, PUT, DELETE, etc.

    # URL 信息
    url: str = request.url                # 完整 URL
    path: str = request.path              # 路径部分

    # 请求头
    user_agent: str = str(request.user_agent)  # 浏览器信息

    # 客户端信息
    remote_addr: str = request.remote_addr  # IP 地址

    return f"""
    <h3>请求信息</h3>
    <p>方法:{method}</p>
    <p>URL: {url}</p>
    <p>路径:{path}</p>
    <p>用户代理:{user_agent}</p>
    <p>IP 地址:{remote_addr}</p>
    """

第六部分:获取请求数据

6.1 实际场景

用户在搜索框输入关键词,或在登录表单提交用户名密码,你需要获取这些数据。

问题:如何获取不同类型的请求数据?

6.2 获取查询参数

python
from flask import Flask, request

app: Flask = Flask(__name__)


@app.route("/search")
def search() -> str:
    # URL: /search?q=python&page=2
    keyword: str = request.args.get("q", "")      # 获取单个参数
    page: int = request.args.get("page", 1, type=int)   # 带默认值和类型转换

    return f"搜索:{keyword}, 页码:{page}"

6.3 获取表单数据

python
@app.route("/login", methods=["POST"])
def login() -> str:
    # Content-Type: application/x-www-form-urlencoded
    username: str = request.form.get("username", "")
    password: str = request.form.get("password", "")

    return f"登录用户:{username}"

6.4 获取 JSON 数据

python
from flask import Flask, request, abort, jsonify
from typing import Any

app: Flask = Flask(__name__)


@app.route("/api/users", methods=["POST"])
def create_user() -> dict[str, Any]:
    # Content-Type: application/json
    data: dict[str, Any] = request.get_json()

    # 如果 JSON 格式无效,返回 400
    if not data:
        abort(400, description="无效的 JSON 数据")

    # 获取单个字段
    username: str = data.get("username", "")
    email: str = data.get("email", "")

    return {"message": f"创建用户:{username}"}

第七部分:请求钩子

7.1 实际场景

每个请求都需要检查用户是否登录,每个请求结束后都要记录日志。

问题:如何在请求处理的特定阶段自动执行代码?

7.2 四种请求钩子

python
import time
from flask import Flask, request, g, Response
from typing import Optional

app: Flask = Flask(__name__)


@app.before_request
def before_request() -> None:
    """在每个请求处理之前执行"""
    # 可用于:用户认证、日志记录、数据库连接
    print(f"处理请求:{request.method} {request.path}")
    g.start_time: float = time.time()  # g 对象用于存储请求级别的数据


@app.after_request
def after_request(response: Response) -> Response:
    """在每个请求处理之后执行(无异常时)"""
    # 可用于:添加响应头、记录日志
    response.headers["X-Custom-Header"] = "MyApp"
    return response  # 必须返回 response


@app.teardown_request
def teardown_request(exception: Optional[Exception] = None) -> None:
    """在请求结束时执行(无论是否有异常)"""
    # 可用于:关闭数据库连接、清理资源
    if exception:
        print(f"请求处理异常:{exception}")


@app.errorhandler(404)
def handle_404(error) -> tuple[str, int]:
    """错误处理器"""
    return "页面未找到", 404

第八部分:响应构建

8.1 实际场景

API 需要返回 JSON 格式的响应,某些页面需要设置特定的响应头。

问题:如何自定义响应格式和响应头?

8.2 Response 对象

视图函数返回的内容会被 Flask 包装成 Response 对象返回给客户端。

python
from flask import Flask, Response, make_response, jsonify
from typing import Any

app: Flask = Flask(__name__)


# 1. 返回字符串(最常见)
@app.route("/hello")
def hello() -> str:
    return "Hello, World!"


# 2. 返回元组 (响应体,状态码)
@app.route("/not-found")
def not_found() -> tuple[str, int]:
    return "页面不存在", 404


# 3. 返回元组 (响应体,状态码,响应头)
@app.route("/custom")
def custom() -> tuple[str, int, dict[str, str]]:
    return "自定义响应", 201, {"X-Custom": "Value"}


# 4. 使用 make_response 创建 Response 对象
@app.route("/make-response")
def make_resp() -> Response:
    response: Response = make_response("响应内容", 200)
    response.headers["X-Header"] = "Value"
    response.set_cookie("my_cookie", "value")
    return response

8.3 JSON 响应

python
# 返回单个对象
@app.route("/api/user/<int:user_id>")
def get_user(user_id: int) -> dict[str, Any]:
    user: dict[str, Any] = {
        "id": user_id,
        "name": "张三",
        "email": "zhangsan@example.com"
    }
    return jsonify(user)


# 返回错误响应
@app.route("/api/error")
def api_error() -> tuple[dict[str, str], int]:
    return jsonify({
        "error": "NOT_FOUND",
        "message": "资源不存在"
    }), 404

第九部分:会话管理

9.1 实际场景

用户登录后,下次访问网站时仍然需要显示登录状态。

问题:如何在多个请求之间保存用户状态?

9.2 session 对象

Flask 的 session 对象用于在请求之间保存用户状态。

python
from flask import Flask, session, redirect, url_for, request
from datetime import timedelta

app: Flask = Flask(__name__)
app.secret_key: str = "your-secret-key"  # 必须设置密钥


@app.route("/login", methods=["GET", "POST"])
def login() -> str:
    if request.method == "POST":
        username: str = request.form.get("username", "")
        # 存储到 session
        session["username"] = username
        return redirect(url_for("index"))

    return """
    <form method="post">
        <input name="username" placeholder="用户名">
        <button type="submit">登录</button>
    </form>
    """


@app.route("/")
def index() -> str:
    # 读取 session
    if "username" in session:
        return f"欢迎回来,{session['username']}!"
    return "您还未登录"


@app.route("/logout")
def logout() -> str:
    # 清除 session
    session.clear()
    return redirect(url_for("index"))


# 配置 session 过期时间
app.permanent_session_lifetime: timedelta = timedelta(days=7)

第九部分:响应对象与静态文件

9.1 关于响应

Flask 视图函数的返回值会自动转换为响应对象。

响应转换规则:
┌─────────────────────────────────────────────────────────────────┐
│  返回值类型          转换结果                                    │
├─────────────────────────────────────────────────────────────────┤
│  Response 对象      → 直接返回                                  │
│  str                → 创建 Response(body, 200, text/html)       │
│  bytes              → 创建 Response(body, 200, application/octet)│
│  dict / list        → jsonify() → Response(JSON, 200)          │
│  tuple (body, code) → Response(body, code)                     │
│  tuple (body, hdrs) → Response(body, 200, headers=hdrs)        │
│  tuple (body, c, h) → Response(body, c, headers=h)             │
│  generator          → 流式响应 (StreamingResponse)              │
└─────────────────────────────────────────────────────────────────┘
python
# response_demo.py
from flask import Flask, make_response, jsonify

app = Flask(__name__)

# 返回 dict → 自动转为 JSON
@app.route("/api/user")
def user_api() -> dict:
    return {"id": 1, "name": "Alice"}

# 使用 make_response 修改响应头
@app.route("/api/custom")
def custom_response():
    resp = make_response("自定义响应", 200)
    resp.headers["X-Custom-Header"] = "my-value"
    return resp

# 返回 tuple(body, status_code)
@app.route("/api/created")
def created() -> tuple[dict, int]:
    return {"id": 42}, 201

9.2 静态文件

动态 Web 应用需要静态文件(CSS、JS、图片)。Flask 默认从 static/ 目录提供。

静态文件目录结构:
┌─────────────────────────────────────────────────────────────────┐
│  my_project/                                                    │
│  ├── app.py                                                     │
│  ├── static/                         ← 静态文件根目录            │
│  │   ├── css/                          style.css               │
│  │   ├── js/                           main.js                 │
│  │   └── images/                       logo.png                │
│  └── templates/                      ← 模板目录                 │
│       └── index.html                                           │
│                                                                 │
│  URL 映射:                                                      │
│  /static/css/style.css   → static/css/style.css                │
│  /static/js/main.js      → static/js/main.js                   │
│  /static/images/logo.png → static/images/logo.png              │
│                                                                 │
│  生成 URL:                                                      │
│  url_for('static', filename='css/style.css')                    │
│  → '/static/css/style.css'                                     │
└─────────────────────────────────────────────────────────────────┘
python
# static_files_demo.py
from flask import Flask, url_for, send_from_directory
from pathlib import Path

app = Flask(__name__)

# 自定义静态文件目录
app2 = Flask(__name__, static_folder="assets", static_url_path="/assets")

# 自定义静态文件端点
UPLOAD_DIR = Path("/var/www/uploads")

@app.route("/uploads/<path:filename>")
def serve_upload(filename: str) -> tuple:
    """安全地提供上传文件下载。"""
    return send_from_directory(UPLOAD_DIR, filename)

生产环境建议: 静态文件应由 Nginx/Apache 直接提供,而非 Flask 开发服务器。


第十部分:L3 专家层 — 底层原理

10.1 Werkzeug 路由匹配算法(Trie 树)

Werkzeug 的 MapRule 类使用基于前缀树(Trie)变体的路由匹配机制。

┌─────────────────────────────────────────────────────────────────┐
│                    路由匹配流程                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   注册路由时:                                                   │
│       @app.route("/user/<int:id>")                              │
│       @app.route("/user/<username>")                            │
│       @app.route("/user/profile")                               │
│                                                                 │
│   编译后的内部结构(简化):                                      │
│                                                                 │
│       /user/                                                    │
│         ├── profile        → view: user_profile (静态)          │
│         ├── <int:id>       → view: show_user   (转换器)        │
│         └── <username>     → view: show_user   (转换器)        │
│                                                                 │
│   匹配请求 /user/42:                                            │
│       1. 前缀匹配 /user/ ✓                                      │
│       2. 尝试静态规则 profile → 不匹配                          │
│       3. 尝试转换器 <int:42> → 匹配!返回 {'id': 42}            │
│       4. 不再尝试后续规则                                        │
│                                                                 │
│   Werkzeug 转换器注册表:                                        │
│       default_converters = {                                    │
│           'string': UnicodeConverter,    # 匹配不含 / 的字符串   │
│           'int': IntegerConverter,       # 匹配整数              │
│           'float': FloatConverter,       # 匹配浮点数            │
│           'path': PathConverter,         # 匹配含 / 的路径       │
│           'any': AnyConverter,           # 匹配枚举值            │
│           'uuid': UUIDConverter,         # 匹配 UUID 格式        │
│       }                                                         │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘
python
# 自定义转换器示例
from werkzeug.routing import BaseConverter

class ListConverter(BaseConverter):
    """匹配逗号分隔的列表:/tags/python,flask,werkzeug"""
    regex = r"[^/]+(?:,[^/]+)*"

    def to_python(self, value: str) -> list[str]:
        return value.split(",")

    def to_url(self, values: list[str]) -> str:
        return ",".join(values)

app.url_map.converters["list"] = ListConverter

@app.route("/tags/<list:tag_names>")
def show_tags(tag_names: list[str]) -> str:
    return f"标签:{', '.join(tag_names)}"

10.2 请求上下文与应用上下文的底层实现

Flask 使用 Context Locals(上下文局部变量)实现线程安全的请求级状态。

┌─────────────────────────────────────────────────────────────────┐
│                    上下文堆栈模型                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   _app_ctx_stack:  [AppContext(app1)] ─→ LIFO 栈                │
│   _req_ctx_stack:  [RequestContext(app1, request1)]             │
│                                                                 │
│   上下文生命周期:                                                │
│                                                                 │
│   请求到达                                                       │
│       │                                                         │
│       ▼                                                         │
│   1. RequestContext.push()                                      │
│       ├── 创建应用上下文(若不存在)                              │
│       ├── 绑定 request 到 current request                       │
│       └── 绑定 session 到 current session                       │
│       │                                                         │
│       ▼                                                         │
│   2. 执行视图函数                                                │
│       │                                                         │
│       ▼                                                         │
│   3. RequestContext.pop()                                       │
│       ├── 执行 teardown_request 钩子                            │
│       ├── 保存 session 到 cookie                                │
│       └── 清理局部变量                                          │
│                                                                 │
│   上下文对象关系:                                                │
│   ┌───────────────────────────────────────────────────────┐     │
│   │  ApplicationContext                                    │     │
│   │  ├── app: Flask 实例                                   │     │
│   │  ├── g: 请求级全局变量 (globals)                       │     │
│   │  └── url_adapter: URL 匹配器                           │     │
│   │                                                        │     │
│   │  RequestContext(extends ApplicationContext)            │     │
│   │  ├── request: Request 对象                             │     │
│   │  ├── session: SecureCookieSession                      │     │
│   │  └── _cv_request: ContextVar 令牌                      │     │
│   └───────────────────────────────────────────────────────┘     │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘
python
# 手动管理上下文(适用于 CLI 脚本、定时任务)
from flask import Flask, current_app

app: Flask = Flask(__name__)

# 方式一:应用上下文
with app.app_context():
    print(current_app.name)  # 可以访问 app 级别的对象

# 方式二:测试客户端自动管理上下文
with app.test_client() as client:
    response = client.get("/api/users")
    # request、session 在 with 块内可用

10.3 session 的签名机制

Flask 的 session 默认使用客户端 cookie 存储,通过 ItsDangerous 库进行签名防止篡改。

┌─────────────────────────────────────────────────────────────────┐
│                Session 签名与验证流程                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   存储流程(响应阶段):                                          │
│                                                                 │
│   session_data = {"username": "alice", "role": "user"}          │
│         │                                                       │
│         ▼                                                       │
│   1. JSON 序列化 → {"username":"alice","role":"user"}           │
│         │                                                       │
│         ▼                                                       │
│   2. URL-safe Base64 编码 → eyJ1c2VybmFtZSI6ImFsaWNlIn0        │
│         │                                                       │
│         ▼                                                       │
│   3. HMAC-SHA256 签名                                           │
│       输入:payload + secret_key + timestamp                    │
│       输出:.tJ9xK3mP...(签名后缀)                            │
│         │                                                       │
│         ▼                                                       │
│   4. 设置 Cookie: session=eyJ1c2VybmFtZSI6ImFsaWNlIn0.tJ9xK3mP  │
│                                                                 │
│   验证流程(请求阶段):                                          │
│                                                                 │
│   1. 读取 Cookie → 分割 payload 和 signature                    │
│   2. 用 secret_key 重新计算签名                                  │
│   3. 对比签名 → 一致则反序列化,不一致则抛出 BadSignature        │
│   4. 检查时间戳 → 过期则抛出 SignatureExpired                   │
│                                                                 │
│   ⚠️ 注意:session 数据是编码而非加密!                          │
│      用户可以读取内容,但无法篡改(除非知道 secret_key)         │
│      敏感信息(密码、token)不应存入 session                    │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

10.4 性能考量

操作耗时内存说明
路由匹配(50 条规则)~0.03ms线性扫描,最坏 O(n)
路由匹配(500 条规则)~0.3ms规则数线性增长
上下文创建 + 销毁~0.15ms~2KBRequestContext 全生命周期
session 签名(cookie 写入)~0.05msHMAC-SHA256 计算
session 验证(cookie 读取)~0.05ms签名校验 + 反序列化
request.args 解析~0.02ms~1KBURL 查询字符串解析
request.get_json()~0.1ms~5KBJSON 反序列化

扩展性提示: 客户端 session 随数据量增长而增大 cookie 体积(4KB 浏览器限制)。用户数据超过 1KB 时应改用服务端 session(Flask-Session + Redis)。

10.5 设计动机

设计决策动机权衡
使用 ContextVar 存储上下文Python 3.7+ 原生支持,替代 thread-local需手动管理异步任务中的上下文传播
路由规则按注册顺序匹配简单直观,开发者可预测规则数量多时性能线性下降
session 存于客户端 cookie无状态,易于水平扩展数据大小受限,无法主动失效
request 作为全局代理对象API 简洁,from flask import request在请求外访问会抛出 RuntimeError
转换器系统可扩展支持自定义 URL 参数解析需手动注册到 url_map.converters

10.6 知识关联

                    浏览器请求


                  ┌───────────┐
                  │  WSGI     │
                  │  Server   │
                  └─────┬─────┘
                        │ environ

              ┌───────────────────┐
              │  RequestContext   │◄── 推送上下文栈
              │  ┌─────────────┐  │
              │  │ request     │  │◄── 代理到 current request
              │  │ session     │  │◄── 签名/验证
              │  └─────────────┘  │
              └────────┬──────────┘


              ┌───────────────────┐
              │  URL 匹配引擎      │◄── Werkzeug Map/Rule
              │  (转换器链)       │
              └────────┬──────────┘
                       │ endpoint + kwargs

              ┌───────────────────┐
              │  视图函数          │
              │  @app.route(...)  │
              └────────┬──────────┘
                       │ Response

                  返回 HTTP 响应

总结

知识点说明
@app.route注册路由
动态路由<int:id> 等类型转换器
url_for()根据视图函数名生成 URL
redirect()页面跳转
abort()返回错误状态码
request请求对象
session会话管理