02-路由与请求处理
Python 3.11+
本章讲解 Flask 路由系统、动态 URL、请求对象和响应构建。
第一部分:@app.route 装饰器
1.1 实际场景
你正在开发一个博客网站,需要为首页、文章列表、文章详情等不同页面设置不同的 URL 路径。
问题:如何将 URL 路径映射到对应的处理函数?
1.2 概念说明
路由(Route)将 URL 映射到视图函数,@app.route() 装饰器用于注册路由。
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 中捕获动态值并传递给视图函数。
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
# /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。
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() 用于立即返回错误状态码。
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 请求的所有信息。
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 获取查询参数
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 获取表单数据
@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 数据
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 四种请求钩子
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 对象返回给客户端。
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 response8.3 JSON 响应
# 返回单个对象
@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 对象用于在请求之间保存用户状态。
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) │
└─────────────────────────────────────────────────────────────────┘# 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}, 2019.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' │
└─────────────────────────────────────────────────────────────────┘# 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 的 Map 和 Rule 类使用基于前缀树(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 格式 │
│ } │
│ │
└─────────────────────────────────────────────────────────────────┘# 自定义转换器示例
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 令牌 │ │
│ └───────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘# 手动管理上下文(适用于 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 | ~2KB | RequestContext 全生命周期 |
| session 签名(cookie 写入) | ~0.05ms | — | HMAC-SHA256 计算 |
| session 验证(cookie 读取) | ~0.05ms | — | 签名校验 + 反序列化 |
| request.args 解析 | ~0.02ms | ~1KB | URL 查询字符串解析 |
| request.get_json() | ~0.1ms | ~5KB | JSON 反序列化 |
扩展性提示: 客户端 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 | 会话管理 |