contextvars — 上下文变量¶
此模块提供用于管理、存储和访问上下文本地状态的 API。ContextVar 类用于声明和使用*上下文变量*。copy_context() 函数和 Context 类应用于在异步框架中管理当前上下文。
具有状态的上下文管理器应使用上下文变量而不是 threading.local()，以防止在并发代码中使用时，其状态意外泄漏到其他代码。
有关更多详细信息，请参阅 PEP 567。
3.7 版新增。
上下文变量¶
- class contextvars.ContextVar(name[, *, default])¶
此类用于声明新的上下文变量，例如
var: ContextVar[int] = ContextVar('var', default=42)
所需的 name 参数用于内省和调试目的。
当在当前上下文中找不到变量的值时，可选的仅限关键字的 default 参数由
ContextVar.get()返回。**重要提示：**上下文变量应在顶级模块级别创建，而绝不能在闭包中创建。
Context对象持有对上下文变量的强引用，这会阻止上下文变量被正确地垃圾回收。- name¶
变量的名称。这是一个只读属性。
3.7.1 版新增。
- get([default])¶
返回当前上下文中上下文变量的值。
如果当前上下文中没有该变量的值，则该方法将
返回方法的 default 参数的值（如果提供）；或
返回上下文变量的默认值（如果创建时指定了默认值）；或
引发
LookupError。
- set(value)¶
调用以在当前上下文中为上下文变量设置新值。
所需的 value 参数是上下文变量的新值。
返回一个
Token对象，该对象可用于通过ContextVar.reset()方法将变量恢复为其先前值。
- reset(token)¶
将上下文变量重置为创建 token 的
ContextVar.set()使用之前的状态。例如
var = ContextVar('var') token = var.set('new value') # code that uses 'var'; var.get() returns 'new value'. var.reset(token) # After the reset call the var has no value again, so # var.get() would raise a LookupError.
- class contextvars.Token¶
ContextVar.set()方法返回 Token 对象。它们可以传递给ContextVar.reset()方法，以将变量的值恢复为相应的 set 之前的状态。- var¶
只读属性。指向创建令牌的
ContextVar对象。
- old_value¶
只读属性。设置为创建令牌的
ContextVar.set()方法调用之前变量的值。如果在调用之前未设置变量，则它指向Token.MISSING。
- MISSING¶
Token.old_value使用的标记对象。
手动上下文管理¶
- contextvars.copy_context()¶
返回当前
Context对象的副本。以下代码段获取当前上下文的副本，并打印其中设置的所有变量及其值
ctx: Context = copy_context() print(list(ctx.items()))
该函数具有 O(1) 复杂度，即对于具有少量上下文变量的上下文和具有大量上下文变量的上下文，其工作速度相同。
- class contextvars.Context¶
ContextVars到其值的映射。Context()创建一个不包含任何值的空上下文。要获取当前上下文的副本，请使用copy_context()函数。每个线程都有一个不同的顶级
Context对象。这意味着当在不同线程中赋值时，ContextVar对象的行为类似于threading.local()。Context 实现了
collections.abc.Mapping接口。- run(callable, *args, **kwargs)¶
在调用 run 方法的上下文对象中执行
callable(*args, **kwargs)代码。返回执行结果，如果发生异常则传播异常。callable 对任何上下文变量所做的任何更改都将包含在上下文对象中。
var = ContextVar('var') var.set('spam') def main(): # 'var' was set to 'spam' before # calling 'copy_context()' and 'ctx.run(main)', so: # var.get() == ctx[var] == 'spam' var.set('ham') # Now, after setting 'var' to 'ham': # var.get() == ctx[var] == 'ham' ctx = copy_context() # Any changes that the 'main' function makes to 'var' # will be contained in 'ctx'. ctx.run(main) # The 'main()' function was run in the 'ctx' context, # so changes to 'var' are contained in it: # ctx[var] == 'ham' # However, outside of 'ctx', 'var' is still set to 'spam': # var.get() == 'spam'
当从多个操作系统线程调用同一个上下文对象，或者递归调用时，该方法会引发
RuntimeError。
- copy()¶
返回上下文对象的浅拷贝。
- var in context
如果 context 为 var 设置了值，则返回
True；否则返回False。
- context[var]
返回 var
ContextVar变量的值。如果在上下文对象中未设置该变量，则会引发KeyError。
- get(var[, default])¶
如果 var 在上下文对象中具有值，则返回 var 的值。否则返回 default。如果未给出 default，则返回
None。
- iter(context)
返回一个迭代器，用于迭代上下文对象中存储的变量。
- len(proxy)
返回上下文对象中设置的变量数量。
- keys()¶
返回上下文对象中所有变量的列表。
- values()¶
返回上下文对象中所有变量值的列表。
- items()¶
返回一个包含上下文对象中所有变量及其值的二元组列表。
asyncio 支持¶
上下文变量在 asyncio 中得到原生支持，无需任何额外配置即可使用。例如，下面是一个简单的回显服务器，它使用上下文变量使远程客户端的地址在处理该客户端的任务中可用。
import asyncio
import contextvars
client_addr_var = contextvars.ContextVar('client_addr')
def render_goodbye():
# The address of the currently handled client can be accessed
# without passing it explicitly to this function.
client_addr = client_addr_var.get()
return f'Good bye, client @ {client_addr}\n'.encode()
async def handle_request(reader, writer):
addr = writer.transport.get_extra_info('socket').getpeername()
client_addr_var.set(addr)
# In any code that we call is now possible to get
# client's address by calling 'client_addr_var.get()'.
while True:
line = await reader.readline()
print(line)
if not line.strip():
break
writer.write(line)
writer.write(render_goodbye())
writer.close()
async def main():
srv = await asyncio.start_server(
handle_request, '127.0.0.1', 8081)
async with srv:
await srv.serve_forever()
asyncio.run(main())
# To test it you can use telnet:
# telnet 127.0.0.1 8081