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 参数的值（如果提供）；或者
返回上下文变量的默认值，如果它是用 1 创建的；或者
引发 LookupError。

set(value)

调用以在当前上下文中为上下文变量设置新值。

所需的 value 参数是上下文变量的新值。

返回一个 Token 对象，该对象可用于通过 ContextVar.reset() 方法将变量恢复到其先前的值。

reset(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

Token 对象由 ContextVar.set() 方法返回。它们可以传递给 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() 函数。

上下文实现了 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(): 返回包含上下文对象中所有变量及其值的 2 元组列表。

异步支持

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

contextvars — 上下文变量 — Python 文档

目录

contextvars — 上下文变量

上下文变量

手动上下文管理

异步支持