contextvars 简介

contextvars 模块来管理上下文变量，可以用于在异步编程中传递上下文信息。

import contextvars

# 创建名为 `var` 的上下文变量，并将其初始值设置为 `'A'`。
var = contextvars.ContextVar('var')
var.set('A')
print(var.get()) # 使用 `var.get()` 获取当前上下文中的值

A

contextvars.copy_context() 复制当前的上下文：

ctx = contextvars.copy_context()
print(list(ctx.items()))

[(<ContextVar name='parent_header' at 0x7f4773500540>, {'msg_id': 'b1d638c6-c0a7bff70da2b2b58d0a7b32_2541_3', 'msg_type': 'execute_request', 'username': 'runner', 'session': 'b1d638c6-c0a7bff70da2b2b58d0a7b32', 'date': datetime.datetime(2024, 11, 18, 8, 33, 33, 24148, tzinfo=tzutc()), 'version': '5.3'}), (<ContextVar name='var' at 0x7f477043fab0>, 'A'), (<ContextVar name='parent_header' at 0x7f477351cea0>, {'msg_id': 'b1d638c6-c0a7bff70da2b2b58d0a7b32_2541_3', 'msg_type': 'execute_request', 'username': 'runner', 'session': 'b1d638c6-c0a7bff70da2b2b58d0a7b32', 'date': datetime.datetime(2024, 11, 18, 8, 33, 33, 24148, tzinfo=tzutc()), 'version': '5.3'})]

备注

contextvars.copy_context() 函数具有 O(1) 复杂度，也就是说对于只包含几个上下文变量和很多上下文变量的情况运行速度是相同的。

def main():
    print(f"var.get(): {var.get()}")
    print(f"ctx[var]: {ctx[var]}")
    # 修改上下文变量
    token = var.set('B')
    print(f"修改后 var.get(): {var.get()}||token:{token}")
    print(f"修改后 ctx[var]: {ctx[var]}")

main()

var.get(): A
ctx[var]: A
修改后 var.get(): B||token:<Token var=<ContextVar name='var' at 0x7f477043fab0> at 0x7f4768e39d00>
修改后 ctx[var]: A

可以看到 ctx[var] 没有被修改。

ctx.run(callable, *args, **kwargs)

进入上下文，执行 callable(*args, **kwargs)，然后退出上下文。返回 callable 的返回值，如果发生异常则传播该异常。

ctx.run(main) # “main”函数对“var”所做的任何更改都将包含在“ctx”中。

ctx[var]

'B'

var.get()

'B'