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 参数是上下文变量的新值。

返回A 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

令牌 对象由返回 ContextVar.set() 方法。它们可以传递给 ContextVar.reset() 方法将变量的值还原为相应的 set .

Token.var

只读属性。指向 ContextVar 创建令牌的对象。

Token.old_value

只读属性。设置为变量在 ContextVar.set() 创建令牌的方法调用。它指向 Token.MISSING 是否在调用之前未设置变量。

Token.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)

执行 callable(*args, **kwargs) 上下文对象中的代码 run 方法被调用。返回执行结果,或者在发生异常时传播异常。

对任何上下文变量的任何更改 可赎回的 生成将包含在上下文对象中::

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 当从多个OS线程调用同一个上下文对象时,或者当以递归方式调用时。

copy()

返回上下文对象的浅副本。

var in context

返回 True 如果 context 有价值 var 集合;返回 False 否则。

context[var]

返回的值 var ContextVar 变量。如果未在上下文对象中设置该变量,则 KeyError 提高了。

get(var[, default])

返回的值 var 如果 var 具有上下文对象中的值。返回 default 否则。如果 default 未给出,返回 None .

iter(context)

对存储在上下文对象中的变量返回迭代器。

len(proxy)

返回在上下文对象中设置的变量数。

keys()

返回上下文对象中所有变量的列表。

values()

返回上下文对象中所有变量值的列表。

items()

返回包含上下文对象中所有变量及其值的2元组列表。

异步支援

本机支持上下文变量 asyncio 并且可以在没有任何额外配置的情况下使用。例如,这里有一个简单的echo服务器,它使用上下文变量使处理该客户机的任务中的远程客户机地址可用:

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