Flaskz介绍7：运行上下文管理
在 Web 请求、后台任务和独立脚本中，经常需要在一次运行过程中共享数据，例如当前用户、请求 ID、数据库 Session 或临时缓存。
Flaskz Context 基于 Python ContextVar 实现，提供框架无关的运行上下文管理机制，可用于 Flask、FastAPI、Celery 和普通 Python 脚本。
主要功能包括：
- 数据共享与隔离：在同一运行上下文中共享数据，并隔离不同运行流程。
- 上下文生命周期管理：支持创建、访问和结束 Context。
- 嵌套与命名 Context：支持上下文栈，并通过 context_key 访问指定 Context。
- 数据与控制选项分离：分别管理业务数据（Values）和控制选项（Options）。
- 资源自动清理：Context 结束时自动运行注册的清理函数。
- 异步支持：基于 ContextVar，支持 asyncio 运行环境。
Context 生命周期
Context 通过 begin_context() 创建，通过 end_context() 结束。
context_key = begin_context() # 创建 Context
try:
print(has_context()) # True
print(get_context_key()) # 当前 Context Key
# Business logic
finally:
end_context(context_key=context_key) # 结束 Context
begin_context() 支持传入初始数据和指定 context_key：
context_key = begin_context( {"user_id": 1001}, context_key="request-001")
如果没有指定 context_key，会自动生成唯一标识。如果指定的 Key 已存在，则抛出 ValueError。
在 Flask、FastAPI、Celery 等应用中，通常通过框架的请求或任务生命周期钩子统一调用 begin_context() 和 end_context()，业务代码只需要访问已有的 Context。
Context 数据管理
Context 提供了一组 API，用于在同一个运行上下文中共享数据。
# 设置单个值
set_context_value("user_id", 1001)
# 批量设置
set_context_values({
"username": "admin",
"role": "administrator"
})
# 获取值
user_id = get_context_value("user_id")
# 获取值，支持默认值
role = get_context_value("role", "guest")
# 判断值是否存在
has_context_value("user_id")
# 获取全部值（浅拷贝）
values = get_context_values()
# 删除指定值
remove_context_value("user_id")
# 清空全部业务数据
clear_context_values()
这些操作默认针对当前 Context，不会自动创建新的 Context。
典型场景是跨函数共享请求数据：
def authenticate(user):
set_context_value("current_user", user)
def process_order():
user = get_context_value("current_user")
# Process order
这样就不需要在多个业务函数之间反复传递运行上下文数据。
Context 控制选项
除了业务数据，Context 还提供独立的 Options 存储空间，用于保存运行过程中的控制参数。
# 保存业务数据
set_context_value("user_id", 1001)
# 设置控制选项
set_context_option("cache_db_session", True)
# 获取控制选项
enabled = get_context_option("cache_db_session", False)
Values 和 Options 相互独立，即使使用相同的 Key，也不会发生冲突。
调用 clear_context_values() 只会清空业务数据，不影响控制选项和已注册的资源清理函数。
嵌套与命名 Context
通过上下文栈管理 Context，允许在一个运行流程中创建多个独立的上下文。
outer_key = begin_context({"name": "outer"})
inner_key = begin_context({"name": "inner"})
# 默认访问当前 Context
print(get_context_value("name")) # inner
# 通过 context_key 访问外层 Context
print(get_context_value("name", context_key=outer_key)) # outer
end_context(context_key=inner_key)
print(get_context_value("name")) # outer
end_context(context_key=outer_key)
默认情况下，操作的是栈顶 Context；通过 context_key，可以访问和修改栈中的其他 Context。
这种方式适合在同一个请求或任务中管理不同作用域的数据。
Context 资源清理
Context 支持通过 register_context_cleanup() 注册资源清理函数。
当 Context 结束时，系统自动运行已注册的清理函数，运行顺序与注册顺序相反。
context_key = begin_context()
try:
session = Session(engine)
# 注册资源清理
register_context_cleanup(session.close, context_key=context_key)
# 保存 Session
set_context_value("db_session", session)
# Business logic
finally:
# 自动运行 session.close()
end_context(context_key=context_key)
这种方式适合管理数据库 Session、文件对象、网络连接等需要主动释放的资源。
Context 负责资源清理，但不会自动处理数据库事务提交或回滚。当前清理回调采用同步 ExitStack 实现。
context_scope() 简化生命周期管理
对于独立脚本或局部运行代码，可以通过 context_scope() 自动创建和关闭 Context。
with context_scope({"task_id": "task-001"}) as context_key:
set_context_value("status", "running")
# Business logic
# 自动结束 Context
context_scope() 会在代码块退出时自动调用 end_context()，包括发生异常的情况。
异步运行支持
flaskz Context 基于 ContextVar，可以在 asyncio 环境中使用。
import asyncio
async def worker(task_id):
context_key = begin_context({"task_id": task_id})
try:
await asyncio.sleep(0.1)
print(get_context_value("task_id"))
finally:
end_context(context_key=context_key)
async def main():
await asyncio.gather(
worker("task-1"),
worker("task-2")
)
asyncio.run(main())
异步子任务会继承创建时的 ContextVar 状态。由于 Context 中包含可变字典，继承的 Context 数据可能被多个任务共享。
因此，需要独立数据空间时，应为子任务创建新的 Context。
典型使用场景
| 场景 | Context 的作用 |
|---|---|
| Flask / FastAPI 请求 | 共享请求数据、当前用户、数据库 Session |
| Celery 后台任务 | 管理任务数据和资源生命周期 |
| 定时任务 / 独立脚本 | 管理单次任务的临时数据与资源 |
| 异步任务 | 提供协程级上下文管理与数据隔离 |
| 复杂业务流程 | 通过嵌套 Context 管理不同作用域的数据 |
总结
flaskz Context 将运行上下文的生命周期管理、数据共享、控制选项和资源清理整合为一套统一的 API。
在实际应用中，通常由框架或任务调度器负责 Context 的创建与结束，业务代码使用 get_context_value()、set_context_value() 等 API 访问上下文数据。
通过这种方式，业务逻辑可以独立于具体的 Web 或任务框架，同时保持明确的数据作用域和资源生命周期。