contextvars —上下文变量

---

该模块提供用于 Management，存储和访问上下文本地状态的 API。 ContextVar类用于语句和使用* Context Variables *。 copy_context()函数和Context类应用于 Management 异步框架中的当前上下文。

具有状态的上下文 Management 器在并发代码中使用时，应使用上下文变量而不是threading.local()来防止其状态意外泄漏到其他代码。

另请参见 PEP 567。

3.7 版中的新Function。

Context Variables

• • class * contextvars. ContextVar(* name * [，**， default *])
  • 此类用于语句新的上下文变量，例如：

var: ContextVar[int] = ContextVar('var', default=42)

必需的* name *参数用于自省和调试目的。

当在当前上下文中找不到该变量的值时，ContextVar.get()返回可选的仅关键字* default *参数。

重要提示： 上下文变量应在顶层模块级别创建，而不要在闭包中创建。 Context对象拥有对上下文变量的强引用，这阻止了对上下文变量进行正确的垃圾回收。

• name
  • 变量的名称。这是一个只读属性。

3.7.1 版中的新Function。

• get([默认])
  • 返回当前上下文的上下文变量的值。

如果在当前上下文中该变量没有值，则该方法将：

• 返回方法的* default *参数的值(如果提供)；要么

• 返回上下文变量的默认值(如果使用默认值创建)；要么

• 提出LookupError。

• set(* value *)

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

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

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

• reset(* token *)
  • 将上下文变量重置为使用创建* token *的ContextVar.set()之前的值。

For example:

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.

• 类别 contextvars. Token

  • • Token 对象由ContextVar.set()方法返回。可以将它们传递给ContextVar.reset()方法，以将变量的值还原为相应 set *之前的值。

• Token. var

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

• Token. old_value

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

• Token. MISSING

  • Token.old_value使用的标记对象。

手动上下文 Management

• contextvars. copy_context ( )
  • 返回当前Context对象的副本。

以下代码段获取当前上下文的副本，并打印其中设置的所有变量及其值：

ctx: Context = copy_context()
print(list(ctx.items()))

该函数具有 O(1)复杂度，即对于具有少量上下文变量的上下文以及具有很多上下文变量的上下文，其工作速度同样快。

• 类别 contextvars. Context
  • ContextVars与其值的 Map。

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'

当从多个 OS 线程对同一上下文对象进行调用时，或者以递归方式调用时，该方法将引发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 ( )

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

asyncio support

asyncio本机支持上下文变量，无需任何额外配置即可直接使用它们。例如，这是一个简单的回显服务器，它使用上下文变量使远程 Client 端的地址在处理该 Client 端的任务中可用：

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