`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¶

Token 对象由 ContextVar.set() 方法返回。它们可以被传递给 ContextVar.reset() 方法，以将变量的值恢复到相应 *set* 调用之前的值。

token 支持上下文管理器协议，以便在退出 with 代码块时恢复相应的上下文变量值。

var = ContextVar('var', default='default value')

with var.set('new value'):
    assert var.get() == 'new value'

assert var.get() == 'default value'

在 3.14 版本加入: 添加了作为上下文管理器的用法支持。

var¶: 一个只读属性。指向创建该 token 的 ContextVar 对象。

old_value¶: 一个只读属性。设为在创建该 token 的 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 对象栈。当前上下文是当前线程栈顶的 Context 对象。栈中的所有 Context 对象都被认为是*已进入*的。

进入一个上下文，可以通过调用其 run() 方法来完成，这会通过将其推入当前线程上下文栈的顶部，使该上下文成为当前上下文。

退出当前上下文，可以通过从传递给 run() 方法的回调函数中返回来完成，这会通过从上下文栈的顶部弹出该上下文，将当前上下文恢复到进入该上下文之前的状态。

由于每个线程都有自己的上下文栈，当在不同线程中赋值时，ContextVar 对象的行为与 threading.local() 类似。

尝试进入一个已经进入的上下文，包括在其他线程中进入的上下文，会引发一个 RuntimeError。

退出一个上下文后，它可以在以后（从任何线程）重新进入。

通过 ContextVar.set() 方法对 ContextVar 值的任何更改都记录在当前上下文中。ContextVar.get() 方法返回与当前上下文关联的值。退出一个上下文会有效地撤销在进入该上下文期间对上下文变量所做的任何更改（如果需要，可以通过重新进入该上下文来恢复值）。

Context 实现了 collections.abc.Mapping 接口。

run(callable, *args, **kwargs)¶

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

示例

import contextvars

var = contextvars.ContextVar('var')
var.set('spam')
print(var.get())  # 'spam'

ctx = contextvars.copy_context()

def main():
    # 'var' was set to 'spam' before
    # calling 'copy_context()' and 'ctx.run(main)', so:
    print(var.get())  # 'spam'
    print(ctx[var])  # 'spam'

    var.set('ham')

    # Now, after setting 'var' to 'ham':
    print(var.get())  # 'ham'
    print(ctx[var])  # 'ham'

# 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:
print(ctx[var])  # 'ham'

# However, outside of 'ctx', 'var' is still set to 'spam':
print(var.get())  # 'spam'

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 支持¶

上下文变量在 asyncio 中得到原生支持，无需任何额外配置即可使用。例如，下面是一个简单的回显服务器，它使用一个上下文变量来使远程客户端的地址在处理该客户端的 Task 中可用：

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}\r\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(b'HTTP/1.1 200 OK\r\n')  # status line
    writer.write(b'\r\n')  # headers
    writer.write(render_goodbye())  # body
    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 or curl:
#     telnet 127.0.0.1 8081
#     curl 127.0.0.1:8081

`contextvars` — 上下文变量¶

上下文变量¶

手动上下文管理¶

asyncio 支持¶

目录

上一主题

下一主题

本页

contextvars — 上下文变量¶

上下文变量¶

手动上下文管理¶

asyncio 支持¶

`contextvars` — 上下文变量¶