17-类视图与信号
Python 3.11+
本章讲解 Flask 类视图（Class-Based Views）和信号系统（Signals）的使用方法与底层原理。
第一部分：类视图基础（L1）
1.1 为什么需要类视图
函数视图在处理复杂业务时面临以下局限：
| 问题 | 函数视图 | 类视图 |
|---|---|---|
| HTTP 方法分支 | if request.method == "GET" 冗长 | get() / post() 方法自动分发 |
| 代码复用 | 复制粘贴或提取函数 | 继承基类 |
| 状态管理 | 闭包或全局变量 | 实例属性 |
| 装饰器应用 | 逐个函数装饰 | 统一 decorators 属性 |
问题：如何让视图更结构化、更易复用？
1.2 View 基类与 as_view()
Flask 提供 flask.views.View 作为所有类视图的基类，核心方法是 dispatch_request() 和类方法 as_view()。
python
# app/views/base_example.py
from flask import Flask, View, Response
from typing import Any
class BaseExampleView(View):
"""最基础的类视图"""
methods: list[str] = ["GET"]
def dispatch_request(self) -> str:
return "Hello from View class!"
app: Flask = Flask(__name__)
# as_view() 将类转换为 Flask 可注册的视图函数
app.add_url_rule("/base", view_func=BaseExampleView.as_view("base_example"))as_view() 做了什么：
类定义 as_view() 调用 注册到 Flask
+---------------+ +---------------------+ +----------------+
| BaseView | | as_view(name="base")| | app.add_url_ |
| | | | | rule() |
| dispatch_ | --------->| 返回闭包 view_func: | ------->| |
| request() | | 1. 创建类实例 | | URL → view_func|
| methods=[] | | 2. 调用 dispatch | +----------------+
+---------------+ +---------------------+python
# as_view() 内部简化实现
from typing import Callable
class ViewDemo:
"""演示 as_view() 的原理"""
methods: list[str] = ["GET"]
@classmethod
def as_view(cls, name: str, **initkwargs: Any) -> Callable:
"""将类转换为视图函数"""
def view(**kwargs: Any) -> Any:
# 每次请求创建新实例（避免状态污染）
self: ViewDemo = cls(**initkwargs)
return self.dispatch_request(**kwargs)
# 将方法列表附加到视图函数上
view.methods = cls.methods # type: ignore[attr-defined]
view.__name__ = name
view.__module__ = cls.__module__
return view1.3 MethodView：RESTful 风格
MethodView 继承自 View，自动将 HTTP 方法映射到同名的类方法。
python
# app/views/methodview_example.py
from flask import Flask, request
from flask.views import MethodView
from typing import Any
class SimpleResource(MethodView):
"""MethodView 基本示例"""
def get(self) -> dict[str, str]:
"""处理 GET 请求"""
return {"method": "GET"}
def post(self) -> tuple[dict[str, Any], int]:
"""处理 POST 请求"""
data: dict[str, Any] = request.get_json() or {}
return {"received": data, "method": "POST"}, 201
def put(self) -> dict[str, str]:
"""处理 PUT 请求"""
return {"method": "PUT"}
def delete(self) -> tuple[str, int]:
"""处理 DELETE 请求"""
return "", 204
app: Flask = Flask(__name__)
app.add_url_rule(
"/api/simple",
view_func=SimpleResource.as_view("simple_resource")
)HTTP 方法到类方法的分发流程：
HTTP 请求 MethodView.dispatch_request() 类方法
+----------------+ +------------------------------+ +----------+
| GET /api/simple| --------->| getattr(self, "get") |----->| get() |
| POST /api/simple|-------->| getattr(self, "post") |----->| post() |
| PUT /api/simple |-------->| getattr(self, "put") |----->| put() |
| DELETE /api/simple|------>| getattr(self, "delete") |----->| delete() |
+----------------+ +------------------------------+ +----------+python
# MethodView dispatch_request 的简化实现
from flask.views import View
from flask import abort
class MethodViewDemo(View):
"""演示 MethodView 的方法分发机制"""
def dispatch_request(self, *args: Any, **kwargs: Any) -> Any:
# 获取 HTTP 方法并转为小写
method: str = request.method.lower()
# 查找对应的方法
handler: Any | None = getattr(self, method, None)
if handler is None:
abort(405)
return handler(*args, **kwargs)1.4 类视图装饰器
类视图支持通过 decorators 属性批量应用装饰器，避免逐个方法装饰。
python
# app/views/decorated_view.py
from functools import wraps
from flask import Flask, request, jsonify
from flask.views import MethodView, View
from flask_login import login_required, current_user
from typing import Callable, Any
def log_request(func: Callable) -> Callable:
"""请求日志装饰器"""
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
print(f"[LOG] {request.method} {request.path}")
return func(*args, **kwargs)
return wrapper
def require_json(func: Callable) -> Callable:
"""要求 JSON 请求体"""
@wraps(func)
def wrapper(*args: Any, **kwargs: Any) -> Any:
if not request.is_json:
return jsonify({"error": "Content-Type must be application/json"}), 400
return func(*args, **kwargs)
return wrapper
class DecoratedResource(MethodView):
"""使用 decorators 属性批量应用装饰器"""
# 装饰器按声明顺序应用（从上到下）
decorators: list[Callable] = [log_request, require_json]
def post(self) -> tuple[dict[str, Any], int]:
data: dict[str, Any] = request.get_json() or {}
return {"status": "ok", "data": data}, 201
def put(self) -> dict[str, str]:
return {"status": "updated"}
app: Flask = Flask(__name__)
app.add_url_rule(
"/api/decorated",
view_func=DecoratedResource.as_view("decorated_resource")
)装饰器应用顺序：
请求到达
|
v
+------------------+
| log_request | ← 最先执行（外层）
+------------------+
|
v
+------------------+
| require_json | ← 第二层
+------------------+
|
v
+------------------+
| post() / put() | ← 实际业务逻辑
+------------------+第二部分：实战（L1）
2.1 完整的 RESTful API 类视图
下面构建一个完整的用户资源类视图，包含 CRUD 操作、数据验证和错误处理。
python
# app/views/user_resource.py
from flask import Flask, request, jsonify, abort
from flask.views import MethodView
from dataclasses import dataclass, field, asdict
from typing import Any
@dataclass
class User:
id: int
name: str
email: str
age: int | None = None
class UserDatabase:
"""模拟用户数据库"""
def __init__(self) -> None:
self._store: dict[int, User] = {}
self._next_id: int = 1
def get_all(self) -> list[dict[str, Any]]:
return [asdict(u) for u in self._store.values()]
def get_by_id(self, user_id: int) -> User | None:
return self._store.get(user_id)
def create(self, data: dict[str, Any]) -> User:
user: User = User(id=self._next_id, **data)
self._store[user.id] = user
self._next_id += 1
return user
def update(self, user_id: int, data: dict[str, Any]) -> User | None:
user: User | None = self._store.get(user_id)
if user is None:
return None
for key, value in data.items():
if hasattr(user, key):
setattr(user, key, value)
return user
def delete(self, user_id: int) -> bool:
if user_id in self._store:
del self._store[user_id]
return True
return False
db: UserDatabase = UserDatabase()
class UserResource(MethodView):
"""用户资源 — 完整 CRUD"""
def get(self, user_id: int | None = None) -> Any:
"""
GET /api/users → 获取用户列表
GET /api/users/<id> → 获取单个用户
"""
if user_id is None:
return jsonify({"users": db.get_all()}), 200
user: User | None = db.get_by_id(user_id)
if user is None:
abort(404, description=f"User {user_id} not found")
return jsonify(asdict(user)), 200
def post(self) -> tuple[dict[str, Any], int]:
"""POST /api/users → 创建用户"""
data: dict[str, Any] | None = request.get_json()
if not data or "name" not in data or "email" not in data:
abort(400, description="name and email are required")
user: User = db.create(data)
return asdict(user), 201
def put(self, user_id: int) -> Any:
"""PUT /api/users/<id> → 完整更新用户"""
data: dict[str, Any] | None = request.get_json()
if not data:
abort(400, description="Request body is required")
user: User | None = db.update(user_id, data)
if user is None:
abort(404, description=f"User {user_id} not found")
return asdict(user), 200
def patch(self, user_id: int) -> Any:
"""PATCH /api/users/<id> → 部分更新用户"""
data: dict[str, Any] | None = request.get_json()
if not data:
abort(400, description="Request body is required")
user: User | None = db.update(user_id, data)
if user is None:
abort(404, description=f"User {user_id} not found")
return asdict(user), 200
def delete(self, user_id: int) -> tuple[str, int]:
"""DELETE /api/users/<id> → 删除用户"""
if not db.delete(user_id):
abort(404, description=f"User {user_id} not found")
return "", 204
app: Flask = Flask(__name__)
# 同一个类注册两个 URL 规则
app.add_url_rule(
"/api/users",
view_func=UserResource.as_view("user_list"),
methods=["GET", "POST"]
)
app.add_url_rule(
"/api/users/<int:user_id>",
view_func=UserResource.as_view("user_detail"),
methods=["GET", "PUT", "PATCH", "DELETE"]
)2.2 URL 变量传递到类方法
URL 中的变量自动作为参数传递给类方法，支持类型转换。
python
# app/views/url_variables.py
from flask import Flask
from flask.views import MethodView
from typing import Any
class ItemResource(MethodView):
"""演示 URL 变量如何传递到类方法"""
# GET /api/items/42/reviews/7
def get(self, item_id: int, review_id: int | None = None) -> dict[str, Any]:
if review_id is None:
return {"item_id": item_id, "action": "list_reviews"}
return {"item_id": item_id, "review_id": review_id, "action": "get_review"}
# PUT /api/items/42
def put(self, item_id: int) -> dict[str, Any]:
return {"item_id": item_id, "action": "update_item"}
app: Flask = Flask(__name__)
app.add_url_rule("/api/items/<int:item_id>", view_func=ItemResource.as_view("item"))
app.add_url_rule(
"/api/items/<int:item_id>/reviews/<int:review_id>",
view_func=ItemResource.as_view("item_review")
)URL 变量传递流程：
URL 规则 路由匹配 方法调用
+----------------------------+ +------------------------+ +----------------------+
| /api/items/<int:item_id> | | 请求: /api/items/42 | | get(item_id=42) |
| |--->| 提取: item_id=42 |--->| put(item_id=42) |
+----------------------------+ +------------------------+ +----------------------+
| /api/items/<int:item_id>/ | | 请求: /api/items/42/ | | get( |
| reviews/<int:review_id> |--->| 提取: item_id=42, |--->| item_id=42, |
+----------------------------+ | review_id=7 | | review_id=7 |
+------------------------+ | ) |
+----------------------+2.3 类视图在 Blueprint 中的注册
类视图与蓝图配合使用，实现模块化组织。
python
# app/blueprints/api_v1.py
from flask import Blueprint
from flask.views import MethodView
from typing import Any
api_v1: Blueprint = Blueprint("api_v1", __name__, url_prefix="/api/v1")
class PostList(MethodView):
"""文章列表"""
def get(self) -> dict[str, Any]:
return {"posts": [], "version": "v1"}
def post(self) -> tuple[dict[str, str], int]:
return {"status": "created"}, 201
class PostDetail(MethodView):
"""文章详情"""
def get(self, post_id: int) -> dict[str, Any]:
return {"post_id": post_id, "title": "Sample Post"}
def put(self, post_id: int) -> dict[str, Any]:
return {"post_id": post_id, "status": "updated"}
def delete(self, post_id: int) -> tuple[str, int]:
return "", 204
# 在蓝图中注册类视图
api_v1.add_url_rule("/posts", view_func=PostList.as_view("post_list"))
api_v1.add_url_rule(
"/posts/<int:post_id>",
view_func=PostDetail.as_view("post_detail")
)python
# app/main.py
from flask import Flask
from app.blueprints.api_v1 import api_v1
app: Flask = Flask(__name__)
app.register_blueprint(api_v1)
# 最终路由:
# GET /api/v1/posts
# POST /api/v1/posts
# GET /api/v1/posts/<id>
# PUT /api/v1/posts/<id>
# DELETE /api/v1/posts/<id>第三部分：信号系统（L1）
3.1 什么是信号
信号（Signal）实现了发布-订阅模式（Pub-Sub），允许应用的不同组件之间松耦合地通信。
发布者（信号发送方） 订阅者（信号接收方）
+----------------+ +------------------+
| 触发信号 | | 接收信号 |
| signal.send() | --------------> | receiver() |
+----------------+ +------------------+
| |
v v
+----------------+ +------------------+
| 携带数据 | | 处理数据 |
| sender + kwargs| --------------> | **kwargs |
+----------------+ +------------------+信号与直接调用的区别：
| 特性 | 直接调用 | 信号 |
|---|---|---|
| 耦合度 | 紧耦合（需知道调用谁） | 松耦合（发布者不知道订阅者） |
| 扩展性 | 需修改调用方代码 | 新增订阅者无需改发布者 |
| 数量关系 | 一对一 | 一对多 |
3.2 Flask 核心信号
Flask 内置多个信号，基于 Blinker 库实现。
python
# app/signals/core_signals.py
from flask import Flask
from flask.signals import (
request_started,
request_finished,
got_request_exception,
template_rendered,
before_render_template,
)
app: Flask = Flask(__name__)
# 请求开始时触发
@request_started.connect_via(app)
def on_request_started(sender: Flask, **extra: object) -> None:
"""请求开始 — 记录请求信息"""
from flask import request
print(f"[request_started] {request.method} {request.path}")
# 请求结束时触发
@request_finished.connect_via(app)
def on_request_finished(sender: Flask, response: object, **extra: object) -> None:
"""请求结束 — 记录响应状态"""
from flask import Response
if isinstance(response, Response):
print(f"[request_finished] status={response.status_code}")
# 发生异常时触发
@got_request_exception.connect_via(app)
def on_exception(sender: Flask, exception: Exception, **extra: object) -> None:
"""异常捕获 — 记录错误"""
print(f"[got_request_exception] {type(exception).__name__}: {exception}")
# 模板渲染前触发
@before_render_template.connect_via(app)
def on_before_template(sender: Flask, template: object, context: dict, **extra: object) -> None:
"""模板渲染前 — 可修改上下文"""
print(f"[before_render_template] template={template}")
# 模板渲染后触发
@template_rendered.connect_via(app)
def on_template_rendered(sender: Flask, template: object, context: dict, **extra: object) -> None:
"""模板渲染后 — 可检查渲染结果"""
print(f"[template_rendered] template={template}")Flask 信号生命周期：
请求到达
|
v
+-------------------------+
| request_started | ← 请求开始
+-------------------------+
|
v
+-------------------------+
| before_request hooks |
+-------------------------+
|
v
+-------------------------+
| 视图函数执行 |
+-------------------------+
|
+----→ [异常] → got_request_exception
|
v
+-------------------------+
| before_render_template | ← 渲染模板前
+-------------------------+
|
v
+-------------------------+
| template_rendered | ← 渲染模板后
+-------------------------+
|
v
+-------------------------+
| after_request hooks |
+-------------------------+
|
v
+-------------------------+
| request_finished | ← 请求结束
+-------------------------+
|
v
响应返回客户端3.3 订阅信号：signal.connect()
除了装饰器，还可以使用 connect() 方法订阅信号。
python
# app/signals/manual_connect.py
from flask import Flask
from flask.signals import request_started
from typing import Any
def log_request(sender: Flask, **extra: Any) -> None:
"""日志记录器"""
from flask import request
print(f"[LOG] {request.method} {request.url}")
def metric_collector(sender: Flask, **extra: Any) -> None:
"""指标收集器"""
from flask import request
print(f"[METRIC] endpoint={request.endpoint}")
app: Flask = Flask(__name__)
# 使用 connect() 订阅
request_started.connect(log_request, app)
request_started.connect(metric_collector, app)
# 也可以不限制 sender（接收所有 Flask 应用的信号）
# request_started.connect(log_request)connect() 方法签名：
python
# signal.connect 的简化签名
def connect(
receiver: Callable, # 接收函数
sender: Any = None, # 限定信号发送者（可选）
weak: bool = True, # 使用弱引用（接收器不会被信号阻止 GC）
) -> None:
"""
sender=None → 接收所有发送者的信号
sender=app → 仅接收指定 app 的信号
weak=False → 强引用，适合局部函数
"""3.4 创建自定义信号
使用 Blinker 的 Namespace 创建应用专属信号。
python
# app/signals/custom_signals.py
from blinker import Namespace, Signal
from flask import Flask
from typing import Any
# 创建信号命名空间
my_signals: Namespace = Namespace()
# 定义自定义信号
user_created: Signal = my_signals.signal("user-created")
user_deleted: Signal = my_signals.signal("user-deleted")
order_placed: Signal = my_signals.signal("order-placed")
# 订阅自定义信号
@user_created.connect
def on_user_created(sender: Any, user: dict[str, Any], **extra: Any) -> None:
"""用户创建后的处理"""
print(f"[user_created] New user: {user['name']} ({user['email']})")
# 发送欢迎邮件、初始化用户设置等
@user_deleted.connect
def on_user_deleted(sender: Any, user_id: int, **extra: Any) -> None:
"""用户删除后的处理"""
print(f"[user_deleted] User {user_id} deleted")
# 清理用户数据、发送通知等
# 在视图中发送信号
from flask.views import MethodView
from flask import request, jsonify
class UserSignup(MethodView):
"""用户注册 — 发送自定义信号"""
def post(self) -> tuple[dict[str, Any], int]:
data: dict[str, Any] = request.get_json() or {}
# ... 创建用户逻辑 ...
new_user: dict[str, Any] = {
"id": 1,
"name": data.get("name", ""),
"email": data.get("email", ""),
}
# 发送信号（通知所有订阅者）
user_created.send(self, user=new_user)
return new_user, 201
app: Flask = Flask(__name__)
app.add_url_rule(
"/api/signup",
view_func=UserSignup.as_view("user_signup")
)自定义信号发送流程：
视图函数 信号发送 订阅者
+----------------+ +------------------+ +------------------+
| UserSignup | | user_created | | on_user_created |
| .post() | | .send(self, |------->| (发送邮件) |
| | | user=new_user) | | |
| 创建用户 |-------->| |------->| on_user_created_2|
| | | | | (记录审计日志) |
| 返回响应 | | 遍历所有 receivers| | |
+----------------+ +------------------+ +------------------+第四部分：L2 实践层
4.1 类视图 vs 函数视图对比
| 维度 | 函数视图 | 类视图 |
|---|---|---|
| HTTP 方法处理 | if/elif 分支 | get() / post() 方法 |
| 装饰器 | 每个函数单独装饰 | decorators 属性统一应用 |
| 代码复用 | 提取公共函数 | 继承基类 |
| 状态保持 | 闭包 / 全局变量 | 实例属性 self.* |
| 可读性 | 简单场景清晰 | 复杂场景更清晰 |
| 学习曲线 | 低 | 中（需理解类与分发） |
| URL 变量 | 函数参数 | 方法参数 |
| Blueprint | @bp.route() | bp.add_url_rule() |
4.2 何时使用类视图
适合使用类视图的场景：
| 场景 | 原因 |
|---|---|
| RESTful API | HTTP 方法天然映射到类方法 |
| 可复用视图 | 通过继承共享逻辑 |
| 需要装饰器链 | decorators 属性统一管理 |
| 复杂业务逻辑 | 多个辅助方法组织在类中 |
适合使用函数视图的场景：
| 场景 | 原因 |
|---|---|
| 简单页面渲染 | return render_template(...) 一行搞定 |
| 重定向 | return redirect(...) 不需要类 |
| 静态响应 | return "OK" 无需结构 |
4.3 信号的典型用途
| 用途 | 使用信号 | 示例 |
|---|---|---|
| 审计日志 | request_started | 记录每次请求的用户、路径、时间 |
| 性能监控 | request_started + request_finished | 计算请求耗时 |
| 缓存失效 | 自定义信号 data_changed | 数据更新后清除相关缓存 |
| 事件驱动 | 自定义信号 order_placed | 下单后触发邮件、库存、积分 |
| 调试诊断 | got_request_exception | 统一记录异常堆栈 |
| 模板扩展 | template_rendered | 注入全局变量 |
python
# app/signals/performance_monitor.py
import time
from flask import Flask, g, request
from flask.signals import request_started, request_finished
from typing import Any
@request_started.connect_via(app)
def start_timer(sender: Flask, **extra: Any) -> None:
"""请求开始时记录时间"""
g._request_start_time: float = time.perf_counter()
@request_finished.connect_via(app)
def end_timer(sender: Flask, response: Any, **extra: Any) -> None:
"""请求结束时计算耗时"""
if hasattr(g, "_request_start_time"):
elapsed: float = time.perf_counter() - g._request_start_time
print(
f"[PERF] {request.method} {request.path} "
f"→ {response.status_code} ({elapsed:.3f}s)"
)4.4 反模式
信号的反模式：
| 反模式 | 问题 | 正确做法 |
|---|---|---|
| 滥用信号 | 代码流程难以追踪 | 直接调用比信号更清晰时用直接调用 |
| 信号中做耗时操作 | 阻塞请求响应 | 耗时操作放入 Celery 队列 |
| 信号修改请求上下文 | 可能导致意外副作用 | 信号应只读或仅记录 |
| 忘记 disconnect | 测试间信号累积 | 测试后清理订阅 |
python
# ❌ 反模式：在信号中做耗时操作
@user_created.connect
def bad_practice(sender: Any, user: dict, **extra: Any) -> None:
# 发送 100 封欢迎邮件 — 阻塞请求！
for email in huge_email_list:
send_welcome_email(email)
# ✅ 正确做法：放入异步队列
@user_created.connect
def good_practice(sender: Any, user: dict, **extra: Any) -> None:
from app.tasks import send_welcome_email_task
send_welcome_email_task.delay(user["email"]) # Celery 异步任务python
# ❌ 反模式：用信号处理关键业务逻辑
@order_placed.connect
def deduct_inventory(sender: Any, order: dict, **extra: Any) -> None:
# 库存扣减应该在事务中，不应该用信号！
db.session.execute("UPDATE inventory SET stock = stock - 1")
# ✅ 正确做法：在业务逻辑中直接处理
def place_order(data: dict) -> Order:
with db.session.begin():
order: Order = Order(**data)
db.session.add(order)
deduct_inventory(order.items) # 直接调用，确保事务一致性
order_placed.send(current_app, order=order) # 信号仅用于通知第五部分：L3 专家层
5.1 Blinker 信号库原理
Flask 信号基于 Blinker 库。理解其内部机制有助于正确使用信号。
Namespace Signal Receiver
+----------------+ +------------------+ +------------------+
| signals: dict | | receivers: dict | | receiver: callable|
| | | | | |
| signal(name) | --------->| connect(receiver)|------->| 添加到 receivers |
| | | send(**kwargs) |------->| 遍历调用 |
+----------------+ +------------------+ +------------------+python
# blinker 核心机制的简化演示
from typing import Callable, Any
from weakref import ref, WeakKeyDictionary
class SignalDemo:
"""演示 Blinker 信号的核心实现"""
def __init__(self) -> None:
# 使用 WeakKeyDictionary 避免阻止接收器 GC
self.receivers: WeakKeyDictionary[Any, Callable] = WeakKeyDictionary()
def connect(
self,
receiver: Callable,
sender: Any = None,
weak: bool = True,
) -> None:
"""订阅信号"""
key: Any = receiver if not weak else ref(receiver)
self.receivers[key] = receiver
def disconnect(self, receiver: Callable) -> None:
"""取消订阅"""
self.receivers.pop(receiver, None)
def send(self, sender: Any, **kwargs: Any) -> list[tuple[Any, Any]]:
"""发送信号，返回 [(receiver, result), ...]"""
results: list[tuple[Any, Any]] = []
for key, receiver in list(self.receivers.items()):
try:
result: Any = receiver(sender, **kwargs)
results.append((receiver, result))
except Exception:
# 默认忽略接收器异常，避免影响其他接收器
pass
return results
# 使用演示
signal: SignalDemo = SignalDemo()
def receiver1(sender: Any, **kwargs: Any) -> str:
return f"receiver1 got: {kwargs}"
def receiver2(sender: Any, **kwargs: Any) -> str:
return f"receiver2 got: {kwargs}"
signal.connect(receiver1)
signal.connect(receiver2)
results: list[tuple[Any, Any]] = signal.send("app", data={"key": "value"})
# results = [
# (receiver1, "receiver1 got: {'data': {'key': 'value'}}"),
# (receiver2, "receiver2 got: {'data': {'key': 'value'}}"),
# ]5.2 信号与请求上下文的关系
信号接收器中可以访问 Flask 的请求上下文（request、g、session），但需要注意信号触发时机与上下文可用性。
请求上下文生命周期
+-- 请求上下文被压入 --+ +-- 请求上下文被弹出 --+
| | | |
v v v v
+--------+ push +----------+ dispatch +--------+ cleanup +----------+
| 无上下文 | -----> | 上下文可用| ---------> | 信号发送| -------->| 上下文销毁 |
+--------+ +----------+ +--------+ +----------+
| |
v v
request, g, request_started
session 可用 request_finished
可访问上下文python
# app/signals/context_aware.py
from flask import Flask, g, request
from flask.signals import request_started, request_finished
from typing import Any
app: Flask = Flask(__name__)
@request_started.connect_via(app)
def setup_request_context(sender: Flask, **extra: Any) -> None:
"""在请求开始时初始化 g 对象"""
g.request_id: str = request.headers.get("X-Request-ID", "unknown")
g.user_agent: str = request.user_agent.string if request.user_agent else ""
@request_finished.connect_via(app)
def add_request_id_to_header(
sender: Flask, response: Any, **extra: Any
) -> None:
"""在请求结束时添加自定义响应头"""
from flask import Response
if isinstance(response, Response) and hasattr(g, "request_id"):
response.headers["X-Request-ID"] = g.request_id关键规则：
| 规则 | 说明 |
|---|---|
request_started | 请求上下文已建立，可安全访问 request、g |
request_finished | 请求上下文仍然存在，response 参数可修改 |
| 应用上下文外发送信号 | 无法访问 request、g |
| 多线程环境 | 每个线程有独立的请求上下文，信号接收器只能访问当前线程的上下文 |
5.3 MethodView 的方法分发机制
MethodView 的方法分发比表面看起来更精巧。
python
# flask/views.py 中 MethodView 的真实实现（简化）
from flask.views import View
from flask import abort, request
from typing import Any
class MethodViewInternals(View):
"""MethodView 内部机制的深入分析"""
# 支持的 HTTP 方法
methods: list[str] = [
"GET", "POST", "PUT", "PATCH",
"DELETE", "HEAD", "OPTIONS", "TRACE"
]
def __init__(self) -> None:
# 每次请求创建新实例
self.args: tuple = ()
self.kwargs: dict[str, Any] = {}
def dispatch_request(self, *args: Any, **kwargs: Any) -> Any:
# 1. 保存参数
self.args = args
self.kwargs = kwargs
# 2. 获取 HTTP 方法并转为小写
meth: str = request.method.lower()
# 3. 检查是否实现了该方法
handler: Any | None = getattr(self, meth, None)
if handler is None:
# 4. 对于 HEAD 请求，如果未实现 head()，回退到 get()
if meth == "head" and hasattr(self, "get"):
handler = self.get
else:
abort(405)
# 5. 调用处理方法
return handler(*args, **kwargs)
@classmethod
def as_view(cls, name: str, **initkwargs: Any) -> Any:
"""创建视图函数"""
view: Any = super().as_view(name, **initkwargs)
# 自动推断允许的 HTTP 方法
# 基于类中实际定义的方法
allowed_methods: set[str] = set()
for method_name in cls.methods:
if hasattr(cls, method_name.lower()):
allowed_methods.add(method_name)
if hasattr(view, "methods"):
view.methods = list(allowed_methods)
return viewHEAD 请求回退到 GET 的特殊处理：
请求: HEAD /api/users/1
|
v
MethodView.dispatch_request()
|
v
meth = "head"
|
v
hasattr(self, "head")? → No
|
v
meth == "head" and hasattr(self, "get")? → Yes
|
v
handler = self.get ← 自动回退
|
v
执行 get() 但 Flask 自动丢弃响应体，只保留响应头5.4 知识关联图
类视图与信号知识体系
|
+-----------------+-----------------+
| | |
类视图 装饰器 信号系统
| | |
+----+----+ +----+----+ +----+----+
| View | | decorators| | 内置信号 |
| 基类 | | 属性 | | request_ |
+----+----+ +----+----+ | started |
| | +----+----+
v v |
+----+----+ +----+----+ v
|MethodView| | 装饰器链 | +---------+
| 方法分发 | | 顺序应用 | | 自定义信号 |
+----+----+ +----+----+ | Namespace|
| | +----+----+
v v |
+----+----+ +----+----+ v
| 类视图 | | 认证装饰 | +---------+
| 在蓝图 | | 器 | | Blinker |
| 中注册 | +----+----+ | 底层原理 |
+----+----+ | +----+----+
| v |
v +---------+ v
+----+----+ | 日志装饰 | +---------+
| URL变量 | | 器 | | 请求上下文|
| 传递 | +---------+ | 关系 |
+----+----+ +---------+
|
v
+---------+
| RESTful |
| API设计 |
+---------+| 知识点 | 说明 |
|---|---|
| View | Flask 类视图基类 |
| as_view() | 将类转换为视图函数 |
| MethodView | RESTful 风格的类视图 |
| decorators | 批量应用装饰器 |
| 信号 | 发布-订阅模式 |
| request_started | 请求开始信号 |
| request_finished | 请求结束信号 |
| Blinker | Flask 信号底层库 |
| 自定义信号 | Namespace + signal() |
总结
类视图与信号——Flask 的结构化与解耦之道
+-------------------------------------------------------------+
| 请求生命周期 |
| |
| 请求 → request_started → 视图 → request_finished → 响应 |
| | | | |
| v v v |
| [信号订阅者] [类视图处理] [信号订阅者] |
| 日志/监控 MethodView 缓存/清理 |
| get/post/put |
+-------------------------------------------------------------+
类视图: 结构化你的 HTTP 方法处理
信号: 解耦你的组件通信| 知识点 | 说明 |
|---|---|
| View 基类 | 类视图的基础，需实现 dispatch_request() |
| MethodView | 自动将 HTTP 方法映射到同名类方法 |
| decorators | 类视图统一应用装饰器的属性 |
| 信号系统 | 发布-订阅模式，实现组件解耦 |
| 内置信号 | request_started, request_finished 等 |
| 自定义信号 | 使用 Blinker Namespace 创建 |
| 反模式 | 避免在信号中做耗时操作或关键业务逻辑 |