web框架——flask3.x-上下文管理机制

发表于 2025-12-8 19:59:01

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

您需要登录才可以下载或查看，没有账号？立即注册

×

flask3.x——上下文管理机制

1. 上下文隔离

焦点目标：办理多哀求并发时的数据杂乱题目，让每个哀求的干系数据
互不干扰，为吸取哀求时做情况铺垫。
实现工具：用contextvars模块内里的contextvar类办理线程/协程数据隔离题目
作用：确保差异哀求的的request、current_app等数据各用各的，而不是串用
Contextvar简概：是python中用于实现线程/协程级数据隔离的焦点工具，能让差异实行单位（如哀求）拥有专属数据，互不干扰
- 主动绑定标识：会主动关联当火线程/协程的唯一标识（线程/协程ID）
- 精准数据定位：存取数据时根据唯一标识准确找到该实行单位的专属数据，而不会串拿其他单位的数据
- 直接存储当前状态：通过_cv_app（应用上下文）和_cv_request（哀求上下文）两个独立的ContextVar，直接存储当前活泼的上下文对象，不维护汗青上下文的栈结构。即同一时候只须要一个活泼的AppContext和一个活泼的RequestContext
- 简化的嵌套模子：对于暂时哀求，不再通过嵌套栈实现，而是使用ContextVar的set()方法直接更换当前上下文
- 主动规复：ContextVar的reset()方法可以主动规复到之前的状态，相称于内置了"栈"的单个层级切换

2. 上下文管理机制（新旧版本对比）

Flask版本差异上下文管理机制阐明：
- Flask 2.x及更早版本：都依靠Werkzeug内里的Local.py 。Local类和LocalStack类共同实现上下文的隔离和管理，通过LocalStack类维护栈结构，管理嵌套上下文。
- Flask 3.x及更新版本：不再依靠Werkzeug内里的Local.py。直接使用Python内置的ContextVar存储上下文对象，通过令牌管理嵌套上下文。
上下文管理机制焦点目标：
- 多个哀求的隔离
- 单个哀求的内的多个Context有序管理
上下文管理机制实现方式差异：
- Flask 3.x及更新版本：
  - 使用Python内置的ContextVar实现
  - 直接受理当前上下文状态，不再使用栈结构
  - 分别存储AppContext和RequestContext，通过令牌团结上下文类的push/pop方法管理
- Flask 2.x及更早版本：
  - 使用LocalStack类实现栈结构管理
  - 将Context对象按次序压入栈中管理
  - 通过栈的后进先出特性实现多个Context的有序切换
上下文管理机制新旧版本总结
- Flask 3.x及更新版本总结：它通过_cv_app（应用上下文）和_cv_request（哀求上下文）两个独立的ContextVar实例，直接存储当前活泼的上下文对象。当出现暂时上下文使用时，会实行set()方法将暂时上下文设为当前活泼状态，覆盖原有上下文值。当暂时哀求处置处罚完后，实行reset()方法规复到之前的上下文状态。整个过程团结try-finally块，确保无论是否发生非常，上下文都能准确规复，克制上下文杂乱。。
- Flask 2.x及更早版本总结：通过_app_ctx_stack（应用上下文栈）和_request_ctx_stack（哀求上下文栈）两个独立的LocalStack实例，以栈结构管理上下文对象。当出现暂时上下文使用时，会实行栈的push()方法将暂时上下文压入栈顶，使其成为当前活泼上下文，而当这个暂时哀求处置处罚完后，实行栈的pop()方法移除栈顶的暂时上下文，规复到之前的上下文状态。整个过程团结try-finally块，确保无论是否发生非常，上下文都能准确出栈，克制栈结构杂乱。

3. 上下文到底是什么及上下文管理机制的实现

上下文：
- 简而言之一篇文章以致一部小说，试卷出题人常常会让你举行接洽上下文分析作者当前写这句话的心情，此时就得接洽上下文即根据文本中出现的信息来举行有逻辑有头脑有步调的分析然后举行作答。
- 此过程中你肯定得记着这些信息，对应到Flask的上下文中其也就是用于存储信息然后举行步调的有内容有次序的实行。然而要存储这些信息就有了哀求上下文RequestContext和应用上下文Appcontext，而之后要举行步调有内容有次序的实行就得须要上下文管理机制。
上下文管理机制的实现：
- 如果没有这个机制会怎么样：
  - 没有上下文管理机制ContextVar大概LocalStack，单个哀求内的多段逻辑会因暂时数据污染、上下文状态杂乱而报错。
  - 比如现在有HTTP哀求过来：起首创建哀求上下文RequestContext再在其 push() 方法中创建/激活应用上下文AppContext，若反面有个暂时上下文使用（比如代码中调用某个工具函数来完成某个功能），此时会新建暂时 AppContext ，不会新增 RequestContext（ RequestContext 与当前用户哀求强绑定，全程唯一，包罗 request 、 session 等哀求数据）:
    - 若没有符合的上下文管理机制
      - Flask 2.x及更早版本：
        
        依靠_app_ctx_stack （ AppContext 栈）和 _request_ctx_stack （ RequestContext 栈）管理上下文。
        
        没有LocalStack 将 AppContext 维护成栈，就无法区分主AppContext 和暂时 AppContext ，烧毁时大概误删主 AppContext ，导致主哀求的后续使用无法正常举行。
      - Flask 3.x及更新版本：
        
        依靠_cv_app （存储 AppContext）和 _cv_request （存储 RequestContext ）两个 ContextVar管理上下文
        
        没有 ContextVar 的Token状态纪录时，暂时使用后无法规复主上下文状态，导致上下文杂乱。
    - 视化
      - Flask 2.x及更早版本:_app_ctx_stack ： [主AppContext, 暂时AppContext]，_request_ctx_stack ： [主RequestContext] （始终唯一）
      - Flask 3.x及更新版本:通过 ContextVar 分别存储 ppContext 和 RequestContext ，用Token纪录 AppContext 的嵌套状态，实现与栈等价的上下文切换， RequestContext 保持唯一

4. 上下文管理机制源码分析

Flask 3.x及更新版本焦点实现组件源码拆解
- 上下文存储方式
  - 使用两个独立的ContextVar实例作为上下文存储容器
1. # flask/globals.py
3. _cv_app: ContextVar[AppContext] = ContextVar("flask.app_ctx")
4. """
5. _cv_app= ContextVar("flask.app_ctx")
6. 表示创建了一个名为_cv_app的Contextvar变量
7. "flask.app_ctx"
8. 表示这个Contextvar数据隔离工具的标识名称为"flask.app_ctx"
9. _cv_app:ContextVar[AppContext]
10. 表示这个变量的类型为Contextvar且这个变量只能存储AppContext类型的对象
11. """
13. _cv_request: ContextVar[RequestContext] = ContextVar("flask.request_ctx")
14. """
15. _cv_request= ContextVar("flask.request_ctx")
16. 表示创建了一个名为_cv_request的Contextvar变量
17. "flask.request_ctx"
18. 表示这个Contextvar数据隔离工具的标识名称为"flask.app_ctx"
19. _cv_request: ContextVar[RequestContext]
20. 表示这个变量的类型为Contextvar且这个变量只能存储RequestContext类型的对象
21. """
复制代码
- 上下文署理方式
  - 通过LocalProxy创建了一系列全局可访问的署理对象，简化上下文访问
1. app_ctx: AppContext = LocalProxy( # type: ignore[assignment]
2. _cv_app, unbound_message=_no_app_msg
3. )
4. current_app: Flask = LocalProxy( # type: ignore[assignment]
5. _cv_app, "app", unbound_message=_no_app_msg
6. )
7. g: _AppCtxGlobals = LocalProxy( # type: ignore[assignment]
8. _cv_app, "g", unbound_message=_no_app_msg
9. )
13. request_ctx: RequestContext = LocalProxy( # type: ignore[assignment]
14. _cv_request, unbound_message=_no_req_msg
15. )
16. request: Request = LocalProxy( # type: ignore[assignment]
17. _cv_request, "request", unbound_message=_no_req_msg
18. )
19. session: SessionMixin = LocalProxy( # type: ignore[assignment]
20. _cv_request, "session", unbound_message=_no_req_msg
21. )
复制代码
- 上下文类实现
  - AppContext类
    - 应用上下文，用于存储应用级别的信息
  1. class AppContext:
  3. def __init__(self, app: Flask) -> None:
  4. """
  5. 将Flask实例保存到self.app中，便于以后应用上下文在其他地方可以便于访问应用实例
  6. 创建URL适配器，调用了app中create_url_adapter()方法，便于将请求的URL路径匹配到对应的视图函数
  7. 创建单次请求生命周期中的全局对象g，便于用于存储单次请求生命周期中重要且反复需要的数据
  8. 创建一个空列表，用于存储ContextVar操作的令牌，便于pop操作时ContextVar利用令牌恢复到之前的应用上下文状态
  9. """
  10. self.app = app
  11. self.url_adapter = app.create_url_adapter(None)
  12. self.g: _AppCtxGlobals = app.app_ctx_globals_class()
  13. self._cv_tokens: list[contextvars.Token[AppContext]] = []
  15. def push(self) -> None:
  16. #绑定应用上下文到当前上下文
  17. """
  18. _cv_app.set(self):将当前(self)应用上下文设置为_cv_app：ContextVar的值，即表示其为当前活跃的上下文。且会返回一个Token(令牌)用于恢复之前上下文的状态信息
  19. self._cv_tokens.append():将返回的Token添加到_cv_tokens列表中
  20. appcontext_pushed.send():发送应用上下文推入信号
  21. """
  22. self._cv_tokens.append(_cv_app.set(self))
  23. appcontext_pushed.send(self.app, _async_wrapper=self.app.ensure_sync)
  25. def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
  26. #弹出应用上下文
  27. """
  28. pop时如果是最后一个令牌
  29. 获取当前异常（如果有）
  30. 调用应用的 do_teardown_appcontext() 方法，执行所有注册的应用上下文清理函数
  31. """
  32. try:
  33. if len(self._cv_tokens) == 1:
  34. if exc is _sentinel:
  35. exc = sys.exc_info()[1]
  36. self.app.do_teardown_appcontext(exc)
  38. """
  39. 不论是否为最后一个上下文均执行finally块，确保上下文正确切换
  40. ctx = _cv_app.get():从ContextVar中获取当前活跃的应用上下文
  41. _cv_app.reset(self._cv_tokens.pop()):将Token列表中的令牌进行获取，进行恢复之前应用上下文的状态
  42. """
  43. finally:
  44. ctx = _cv_app.get()
  45. _cv_app.reset(self._cv_tokens.pop())
  48. """
  49. 最后检查，finally块中获取的当前活跃的应用上下文是否为当前的应用上下文。如果不是则抛异常，确保上下文的正确切换
  50. """
  51. if ctx is not self:
  52. raise AssertionError(
  53. f"Popped wrong app context. ({ctx!r} instead of {self!r})"
  54. )
  56. """
  57. 发送当前应用上下文弹出信号
  58. """
  59. appcontext_popped.send(self.app, _async_wrapper=self.app.ensure_sync)
  63. 核心工作流程总结
  64. 初始化：创建 AppContext 实例，存储应用实例、URL 适配器和全局对象 g
  65. 推入上下文：
  66. 调用 push() 方法
  67. 将当前实例设置为全局 _cv_app ContextVar 的值
  68. 存储返回的令牌
  69. 发送应用上下文推入信号
  70. 弹出上下文：
  71. 调用 pop() 方法
  72. 如果是最后一个令牌，执行应用上下文清理
  73. 使用存储的令牌恢复 ContextVar 之前的状态
  74. 验证弹出的上下文是否正确
  复制代码
  - RequestContext类
    - 哀求上下文，用于存储哀求级别的信息
  1. class RequestContext:
  3. def __init__(
  4. self,
  5. app: Flask,
  6. environ: WSGIEnvironment,
  7. request: Request | None = None,
  8. session: SessionMixin | None = None,
  9. ) -> None:
  10. """
  11. 初始化，存储应用实例、request、路由适配器、flash、session等
  12. """
  13. self.app = app
  14. self.request = request or app.request_class(environ) # 请求对象
  15. self.url_adapter = None
  16. self.flashes = None
  17. self.session = session # 会话对象
  18. self._after_request_functions = [] # 后处理函数列表
  19. self._cv_tokens: list[tuple[contextvars.Token[RequestContext], AppContext | None]] = [] # 令牌列表
  22. def push(self) -> None:
  23. #绑定请求上下文到当前上下文
  24. """
  25. 应用上下文管理逻辑
  26. 1.检查确保有活跃应用上下文存在
  27. 2.如果，不存在或者应用上下文的应用实例和当前请求上下文的应用实例不一样则创建与当前请求上下文相符合的应用上下文
  28. app_ctx = self.app.app_context()：使用应用实例的app_context()方法返回AppContext对象
  29. app_ctx.push()：执行AppContext类的push方法将当前AppContext实例设置为全局 _cv_app ContextVar 的值
  30. 3.否则，不在创建新的应用上下文
  31. 4.最后存储请求上下文令牌，和应用上下文以元组形式将其归于一组到列表，用于管理嵌套上下文和状态恢复
  32. """
  33. app_ctx = _cv_app.get(None)
  34. if app_ctx is None or app_ctx.app is not self.app:
  35. app_ctx = self.app.app_context()
  36. app_ctx.push()
  37. else:
  38. app_ctx = None
  39. self._cv_tokens.append((_cv_request.set(self), app_ctx))
  42. """
  43. 会话初始化逻辑
  44. 如果会话对象不存在
  45. 1.获取应用的会话接口 app.session_interface
  46. 2.调用open_session(self.app, self.request)尝试从请求中恢复会话
  47. 3.如果恢复失败 if self.session is None ，创建一个新的会话 make_null_session
  48. """
  49. if self.session is None:
  50. session_interface = self.app.session_interface
  51. self.session = session_interface.open_session(self.app, self.request)
  53. if self.session is None:
  54. self.session = session_interface.make_null_session(self.app)
  57. if self.url_adapter is not None:
  58. self.match_request()
  60. def pop(self, exc: BaseException | None = _sentinel) -> None: # type: ignore
  61. """
  62. 判断是否为最后一个请求上下文
  63. clear_request = len(self._cv_tokens)
  64. 判断：clear_request == 1
  65. """
  66. clear_request = len(self._cv_tokens) == 1
  69. """
  70. pop时如果是最后一个请求上下文
  71. 获取当前异常（如果有）
  72. 调用do_teardown_request执行请求上下文清理函数
  73. 检查请求对象是否有close方法，如果有则调用关闭资源
  74. """
  75. try:
  76. if clear_request:
  77. if exc is _sentinel:
  78. exc = sys.exc_info()[1]
  79. self.app.do_teardown_request(exc)
  81. request_close = getattr(self.request, "close", None)
  82. if request_close is not None:
  83. request_close()
  86. """
  87. 无论是否有异常均执行finally块
  88. 1.获取当前请求上下文，并获取令牌便于恢复之前的请求上下文状态，和获取与之相关的应用上下文
  89. 2.利用令牌恢复之前的请求上下文状态
  90. 3.如果是最后一个请求上下文，将请求对象从WSGI环境中移除
  91. 4.如果app_ctx不为None（表示替换这个请求上下文时，创建过与之相配合的应用上下文），则清除这个应用上下文
  92. 5.确保弹出的请求上下文是当前实例，如果不是则抛异常
  94. """
  95. finally:
  96. ctx = _cv_request.get()
  97. token, app_ctx = self._cv_tokens.pop()
  98. _cv_request.reset(token)
  100. if clear_request:
  101. ctx.request.environ["werkzeug.request"] = None
  103. if app_ctx is not None:
  104. app_ctx.pop(exc)
  106. if ctx is not self:
  107. raise AssertionError(
  108. f"Popped wrong request context. ({ctx!r} instead of {self!r})"
  109. )
  112. 总结
  113. RequestContext类实现了Flask的请求上下文管理，核心功能包括：
  115. 初始化请求相关的资源（请求对象、会话、URL适配器等）
  116. 在push时确保应用上下文存在并正确管理嵌套上下文
  117. 在pop时按正确顺序恢复上下文状态并清理资源
  118. 通过令牌机制支持嵌套上下文的正确管理
  119. 确保资源的正确释放和异常安全
  复制代码

web框架——flask3.x-上下文管理机制

马上注册，结交更多好友，享用更多功能，让你轻松玩转社区。

浏览过的版块

张春