【Python模块】—

【Python模块】——contextvars

Python模块之contextvars

0. 前言
1. 源码
2. contextvar模块详解
- 2.1 contextvar.ContextVar
- 2.2 contextvar.Token
- 2.3 contextvar.Context
- - 代码示例：
  - 解释
- 2.4 asyncio模块
- - 2.4.1 使用contextvars隔离异步任务的变量
  - 2.4.2 使用contextvars代替threading.local()
3. 参考资料

0. 前言

在阅读flask上下文管理器相关的文章，发现flask在2.3.0使用contextvar代替了threading.local实现线程变量跟踪。contextvar，顾名思义，是用来记录上下文的变量。自python3.7.0引入contextvar模块后，既可以像threading.local为每个线程维护各自的变量，还支持异步任务的变量跟踪。

1. 源码

https://peps.python.org/pep-0567/#abstract
https://docs.python.org/zh-cn/3/library/contextvars.html

2. contextvar模块详解

2.1 contextvar.ContextVar

使用ContextVar创建一个上下文管理对象

ctx = ContextVar('ctx', default=42) # 创建一个上下文变量，name=ctx，默认值=42

try:
    print(ctx.get())
    # 通过ctx.set()设置ctx的值
    token = ctx.set(100)
    # 通过ctx.get()获取ctx的值
    print(ctx.get())
finally:
    # 通过ctx.reset()重置ctx的值，恢复到设置token操作之前的值
    ctx.reset(token)

print(ctx.get())

# 运行结果：
"""
42
100
42
"""

注意：ctx.get()如果此前没有给ContextVar赋值或者分配默认值，调用此方法会提示LookupError

2.2 contextvar.Token

contextvar.Token官方解释: contextvars.Token is an opaque object that should be used to restore the ContextVar to its previous value, or to remove it from the context if the variable was not set before. It can be created only by calling ContextVar.set().

contextvar.Token 可以用来保存ContextVar之前的值，如果之前没有设置过则会被移除。
Token变量维护了两个属性：var和old_value
- token.var用来指向创建token的ContextVar；
- token.old_value用来保存之前的值，如果之前未分配过值，则填充token.MISSING

2.3 contextvar.Context

contextvar.Context可以用来创建一个空的上下文；如果想使用当前上下文的拷贝，可以通过ctx = contextvars.copy_context()实现。

代码示例：

from contextvars import ContextVar, copy_context

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

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

    var.set('ham')

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

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:

print("not in main, var.get() ==", var.get())

# However, outside of 'ctx', 'var' is still set to 'spam':
# var.get() == 'spam'
print("not in main, ctx[var] ==", ctx[var])

"""
执行结果：
before var.get() == spam
before ctx[var] == spam
after var.get() == ham
after ctx[var] == ham
not in main, var.get() == spam
not in main, ctx[var] == ham
"""

解释

通过ContextVar创建变量var，并赋值=spam；
使用ctx=copy_context()拷贝当前上下文，并使用ctx.run(main)执行main函数
- main函数中对var的任何操作，都会被记录到ctx中，可通过ctx[var]查看
执行完后，var退回原先的值，而ctx仍然记录着main函数中的操作记录

2.4 asyncio模块

2.4.1 使用contextvars隔离异步任务的变量

from contextvars import ContextVar
import asyncio

ctx = ContextVar('ctx')

async def ctx_get():
    print(f'Request ID (Inner)=={ctx.get()}')

async def ctx_set(value):
    ctx.set(value)
    await ctx_get()  # 设置完后去获取ctx
    print(f'Request ID (Outer)=={ctx.get()}')

async def main():
    tasks = []
    for value in range(1, 5):
        tasks.append(asyncio.create_task(ctx_set(value)))
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    asyncio.run(main())

"""
执行结果：
Request ID (Inner)==1
Request ID (Outer)==1
Request ID (Inner)==2
Request ID (Outer)==2
Request ID (Inner)==3
Request ID (Outer)==3
Request ID (Inner)==4
Request ID (Outer)==4
"""

2.4.2 使用contextvars代替threading.local()

threading.local()实例

import threading
import time


local = threading.local()
def task(num):
    local.num = num
    ident = threading.get_ident()
    time.sleep(0.1)
    print(f'线程{ident}的local.num为{local.num}')


for i in range(5):
    t = threading.Thread(target=task, args=(i,))
    t.start()

contextvars实例

import threading
import contextvars

ctx = contextvars.ContextVar("ctx")

def task_get():
    ident = threading.get_ident()
    print(f"线程{ident}")
    return ctx.get()


def task(num):
    ctx.set(num)
    time.sleep(0.1)
    ident = threading.get_ident()
    print("线程{}的ctx为{}".format(ident, task_get()))

for i in range(5):
    t = threading.Thread(target=task, args=(i,))
    t.start()