40.ContextVar

ContextVar

contextvars模块提供了管理、存储和访问上下文局部状态的API。上下文变量(Context Variables)是一种新的功能,它在Python 3.7版本中引入,用于替代threading.local(),以避免在并发代码中意外地将状态传递给其他代码。

以下是一些关键点:

  • contextvars.ContextVar 类用于声明新的上下文变量。
  • contextvars.copy_context() 函数和 contextvars.Context 类用于在异步框架中管理当前上下文。
  • 上下文变量应该在模块顶层创建,而不是在闭包中,以确保它们可以被正确地垃圾回收。
  • ContextVar.get() 方法用于获取当前上下文中的变量值,如果变量未设置,则返回默认值或抛出LookupError。
  • ContextVar.set() 方法用于在当前上下文中设置变量的新值,并返回一个可以用于恢复之前值的Token对象。
  • ContextVar.reset(token) 方法使用Token对象将变量重置为其之前的状态。

此外,contextvars模块还提供了手动上下文管理的功能,如copy_context()函数,它可以复制当前上下文对象,以及Context类,它实现了collections.abc.Mapping接口,允许你执行代码并在特定上下文中管理变量。

在异步编程中,asyncio原生支持上下文变量,无需额外配置即可使用。

ContextVar

contextvars.ContextVar 用于在异步编程中(如在协程中)管理上下文变量。上下文变量是在请求的生命周期中长时间保持的变量,例如,在 Web 应用程序中,它们可能包括用户会话信息。

ContextVar 类本身是一个实现上下文变量的抽象类。它有一个 get() 方法来获取当前变量的值,一个 set() 方法来设置当前变量的值,以及一个 reset() 方法来重置当前变量的值为默认值。这些方法都是协程安全的,可以在异步函数中使用。

要使用 ContextVar,你需要创建一个继承自 ContextVar 的类,并实现 get()、set() 和 reset() 方法。例如:

from contextvars import ContextVar

class UserSessionContextVar(ContextVar):
    async def get_user_id(self):
        return self.get()

    async def set_user_id(self, user_id):
        self.set(user_id)

    async def reset_user_id(self):
        self.reset()

# 然后,你可以在异步函数中使用这个上下文变量:
user_session_context = UserSessionContextVar()

async def login(user_id):
    user_session_context.set(user_id)
    # 在这里使用 user_session_context.get() 获取当前用户ID

async def get_current_user_id():
    return user_session_context.get()

# 这样,你可以在多个异步函数之间传递用户会话信息,而无需在每个函数中手动传递变量。
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
b''

contextvars.Token

contextvars.Token 对象在 Python 的 contextvars 模块中用于在上下文变量(Context Variable)的值被改变后,能够恢复到之前的状态。当你通过 ContextVar.set() 方法设置一个新的值给上下文变量时,这个方法会返回一个 Token 对象。这个 Token 对象代表了上下文变量在设置新值之前的状态。

以下是 Token 对象的主要作用:

  • 恢复之前的状态:你可以使用 ContextVar.reset() 方法和这个 Token 对象来将上下文变量恢复到它在设置新值之前的状态。这在你需要临时改变一个变量的值,然后在特定时刻(例如,离开一个代码块)恢复到原始值的场景中非常有用。
  • 保持上下文变量的引用:Token 对象持有对上下文变量的强引用,这确保了在 Token 对象存在期间,上下文变量不会被垃圾回收器回收,即使没有其他引用。
  • 上下文管理:Token 对象通常与 with 语句一起使用,以确保在退出 with 代码块时自动恢复上下文变量的值。这样,你可以在代码块的开始设置一个新值,在代码块结束时自动恢复到旧值,而不需要显式调用 reset() 方法。

下面是一个使用 Token 的例子:

from contextvars import ContextVar
var = ContextVar('var', default=42)
print(var.get())
token = var.set('new value')
print(var.get())
# code that uses 'var'; var.get() returns 'new value'.
var.reset(token)
print(var.get())
# After the reset call the var has no value again, so
# var.get() would raise a LookupError.
import contextvars

# 创建一个上下文变量
var = contextvars.ContextVar("example", default="default")

# 获取当前值
print(var.get())  # 输出: default

# 设置新值并获取 Token
token = var.set("new_value")
# 在这个代码块中,var 的值是 "new_value"
print(var.get())  # 输出: new_value
var.reset(token)
# 这里可以执行其他操作,var 的值仍然是 "new_value"

# 离开 with 代码块后,var 的值会自动恢复到 "default"
print(var.get())  # 输出: default

# 在这个例子中,with var.set("new_value"): 语句块开始时,var 的值被设置为 "new_value"。
# 当离开这个 with 语句块时,var 的值会自动恢复到之前的 "default",这是由 Token 对象在 with 语句块结束时自动处理的。
default
new_value
default