这篇“python中如何使用contextvars模块”文章的知识点大部分人都不太理解，所以小编给大家总结了以下内容，内容详细，步骤清晰，具有一定的借鉴价值，希望大家阅读完这篇文章能有所收获，下面我们一起来看看这篇“python中如何使用contextvars模块”文章吧。

前记

在Python3.7后官方库出现了

contextvars

上下文,可以理解为我们说话的语境, 在聊天的过程中, 有些话脱离了特定的语境,他的意思就变了,程序的运行也是如此.在线程中也是有他的上下文,只不过称为堆栈,如在python中就是保存在thread.local变量中,而协程也有他自己的上下文,但是没有暴露出来,不过有了

contextvars

contextvars

使用

contextvars

contextvars

更新说明

• 切换web框架

  sanic

  为

  starlette

• 增加一个自己编写且可用于

  starlette

  ,

  fastapi

  的context说明

• 更新fast_tools.context的最新示例以及简单的修改行文。

1.有无上下文传变量的区别

如果有用过

Flask

Flask

Flask

threading.local

threading.local

werkzeug

gevent

Local

Flask

request

from flask import Flask, request
app = Flask(__name__)
@app.route('/')
def root():
    so1n_name = request.get('so1n_name')
    return f'Name is {so1n_name}'

与之相比的是

Python

Djano

request

from django.http import HttpResponse
def root(request):
    so1n_name = request.get('so1n_name')
    return HttpResponse(f'Name is {so1n_name}')

通过上面两者的对比可以发现， 在

Django

Flask

可能会有人说, 也就是传个变量的区别,为了省传这个变量,而花许多功夫去维护一个上下文变量,有点不值得,那可以看看下面的例子,如果层次多就会出现’一个参数传一天’的情况(不过分层做的好或者需求不坑爹一般不会出现像下面的情况,一个好的程序员能做好代码的分层, 但可能也有出现一堆烂需求的时候)

# 伪代码,举个例子一个request传了3个函数
from django.http import HttpResponse
def is_allow(request, uid):
    if request.ip == '127.0.0.1' and check_permissions(uid):
        return True
    else:
        return False
def check_permissions(request, uid):
    pass

def root(request):
    user_id = request.GET.get('uid')
    if is_allow(request, id):
    	return HttpResponse('ok')
    else
        return HttpResponse('error')

此外， 除了防止

一个参数传一天

Flask

import logging
from typing import Any
from flask import g  # type: ignore
from flask.logging import default_handler
# 这是一个Python logging.Filter的对象, 日志在生成之前会经过Filter步骤, 这时候我们可以为他绑定request_id变量
class RequestIDLogFilter(logging.Filter):
    """
    Log filter to inject the current request id of the request under `log_record.request_id`
    """
    def filter(self, record: Any) -> Any:
        record.request_id = g.request_id or None
        return record
# 配置日志的format格式, 这里多配了一个request_id变量
format_string: str = (
    "[%(asctime)s][%(levelname)s][%(filename)s:%(lineno)d:%(funcName)s:%(request_id)s]" " %(message)s"
)
# 为flask的默认logger设置format和增加一个logging.Filter对象
default_handler.setFormatter(logging.Formatter(format_string))
default_handler.addFilter(RequestIDLogFilter())
# 该方法用于设置request_id
def set_request_id() -> None:
    g.request_id = request.headers.get("X-Request-Id", str(uuid4()))
# 初始化FLask对象, 并设置before_request
app: Flask = Flask("demo")
app.before_request(set_request_id)

2.如何使用contextvars模块

这里举了一个例子, 但这个例子也有别的解决方案. 只不过通过这个例子顺便说如何使用contextvar模块

首先看看未使用

contextvars

starlette

contextvars

Redis

Redis

# demo/web_tools.py
# 通过中间件把变量给存进去
class RequestContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(
            self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        request.stat.redis = REDIS_POOL
        response = await call_next(request)
        return response
# demo/server.py
# 调用变量
@APP.route('/')
async def homepage(request):
    # 伪代码,这里是执行redis命令
    await request.stat.redis.execute()
    return JSONResponse({'hello': 'world'})

代码非常简便, 也可以正常的运行, 但你下次在重构时, 比如简单的把redis这个变量名改为new_redis, 那IDE不会识别出来, 需要一个一个改。 同时, 在写代码的时候, IDE永远不知道这个方法调用到的变量的类型是什么, IDE也无法智能的帮你检查(如输入request.stat.redis.时,IDE不会出现execute,或者出错时,IDE并不会提示). 这非常不利于项目的工程化, 而通过

contextvars

TypeHints

说了那么多, 下面以一个

Redis

contextvars

TypeHints

# demo/context.py
# 该文件存放contextvars相关
import contextvars
if TYPE_CHECKING:
    from demo.redis_dal import RDS  # 这里是一个redis的封装实例
# 初始化一个redis相关的全局context
redis_pool_context = contextvars.ContextVar('redis_pool')
# 通过函数调用可以获取到当前协程运行时的context上下文
def get_redis() -> 'RDS':
    return redis_pool_context.get()
# demo/web_tool.py
# 该文件存放starlette相关模块
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.middleware.base import RequestResponseEndpoint
from starlette.responses import Response
from demo.redis_dal import RDS
# 初始化一个redis客户端变量,当前为空
REDIS_POOL = None  # type: Optional[RDS]
class RequestContextMiddleware(BaseHTTPMiddleware):
    async def dispatch(
            self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        # 通过中间件,在进入路由之前,把redis客户端放入当前协程的上下文之中
        token = redis_pool_context.set(REDIS_POOL)
        try:
        	response = await call_next(request)
            return response
        finally:
        	# 调用完成,回收当前请求设置的redis客户端的上下文
            redis_pool_context.reset(token)
async def startup_event() -> None:
    global REDIS_POOL
    REDIS_POOL = RDS() # 初始化客户端,里面通过asyncio.ensure_future逻辑延后连接
async def shutdown_event() -> None:
    if REDIS_POOL:
        await REDIS_POOL.close() # 关闭redis客户端
# demo/server.py
# 该文件存放starlette main逻辑
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from demo.web_tool import RequestContextMiddleware
from demo.context import get_redis
APP = Starlette()
APP.add_middleware(RequestContextMiddleware)
@APP.route('/')
async def homepage(request):
    # 伪代码,这里是执行redis命令
    # 只要验证 id(get_redis())等于demo.web_tool里REDID_POOL的id一致,那证明contextvars可以为asyncio维护一套上下文状态
    await get_redis().execute()
    return JSONResponse({'hello': 'world'})

3.如何优雅的使用contextvars

从上面的示例代码来看, 使用

contextvar

TypeHint

自己在使用了

contextvars

fastapi

starlette

以下是调用者的代码示例, 这里的实例化变量由一个http client代替, 且都会每次请求分配一个客户端实例, 但在实际使用中并不会为每一个请求都分配一个客户端实例, 很影响性能:

import asyncio
import uuid
from contextvars import Context, copy_context
from functools import partial
from typing import Optional, Set
import httpx
from fastapi import FastAPI, Request, Response
from fast_tools.context import ContextBaseModel, ContextMiddleware, HeaderHelper
app: FastAPI = FastAPI()
check_set: Set[int] = set()
class ContextModel(ContextBaseModel):
    """
	通过该实例可以屏蔽大部分与contextvars相关的操作,如果要添加一个变量,则在该实例添加一个属性即可.
	属性必须要使用Type Hints的写法,不然不会识别(强制使用Type Hints)
    """
    # 用于把自己的实例(如上文所说的redis客户端)存放于contextvars中
    http_client: httpx.AsyncClient
    # HeaderHepler用于把header的变量存放于contextvars中
    request_id: str = HeaderHelper.i("X-Request-Id", default_func=lambda request: str(uuid.uuid4()))
    ip: str = HeaderHelper.i("X-Real-IP", default_func=lambda request: request.client.host)
    user_agent: str = HeaderHelper.i("User-Agent")

    async def before_request(self, request: Request) -> None:
        # 请求之前的钩子, 通过该钩子可以设置自己的变量
        self.http_client = httpx.AsyncClient()
        check_set.add(id(self.http_client))

    async def before_reset_context(self, request: Request, response: Optional[Response]) -> None:
        # 准备退出中间件的钩子, 这步奏后会清掉上下文
        await self.http_client.aclose()
context_model: ContextModel = ContextModel()
app.add_middleware(ContextMiddleware, context_model=context_model)
async def test_ensure_future() -> None:
    assert id(context_model.http_client) in check_set
def test_run_in_executor() -> None:
    assert id(context_model.http_client) in check_set
def test_call_soon() -> None:
    assert id(context_model.http_client) in check_set
@app.get("/")
async def root() -> dict:
    # 在使用asyncio.ensure_future开启另外一个子协程跑任务时, 也可以复用上下文
    asyncio.ensure_future(test_ensure_future())
    loop: "asyncio.AbstractEventLoop" = asyncio.get_event_loop()
    # 使用call_soon也能复用上下文
    loop.call_soon(test_call_soon)
    # 使用run_in_executor也能复用上下文, 但必须使用上下文的run方法, copy_context表示复制当前的上下文
    ctx: Context = copy_context()
    await loop.run_in_executor(None, partial(ctx.run, test_run_in_executor))  # type: ignore
    return {
        "message": context_model.to_dict(is_safe_return=True),  # not return CustomQuery
        "client_id": id(context_model.http_client),
    }
if __name__ == "__main__":
    import uvicorn  # type: ignore
    uvicorn.run(app)

可以从例子中看到, 通过封装的上下文调用会变得非常愉快, 只要通过一两步方法就能设置好自己的上下文属性, 同时不用考虑如何编写上下文的生命周期. 另外也能通过这个例子看出, 在asyncio生态中, contextvars能运用到包括子协程, 多线程等所有的场景中.

4.contextvars的原理

在第一次使用时,我就很好奇contextvars是如何去维护程序的上下文的,好在contextvars的作者出了一个向下兼容的contextvars库,虽然他不支持asyncio,但我们还是可以通过代码了解到他的基本原理.

4.1 ContextMeta,ContextVarMeta和TokenMeta

代码仓中有

ContextMeta

ContextVarMeta

TokenMeta

Context

ContextVar

Token

class ContextMeta(type(collections.abc.Mapping)):
    # contextvars.Context is not subclassable.
    def __new__(mcls, names, bases, dct):
        cls = super().__new__(mcls, names, bases, dct)
        if cls.__module__ != 'contextvars' or cls.__name__ != 'Context':
            raise TypeError("type 'Context' is not an acceptable base type")
        return cls

4.2 Token

上下文的本质是一个堆栈, 每次set一次对象就向堆栈增加一层数据, 每次reset就是pop掉最上层的数据, 而在

Contextvars

Token

class Token(metaclass=TokenMeta):
    MISSING = object()
    def __init__(self, context, var, old_value):
        # 分别存放上下文变量, 当前set的数据以及上次set的数据
        self._context = context
        self._var = var
        self._old_value = old_value
        self._used = False
    @property
    def var(self):
        return self._var
    @property
    def old_value(self):
        return self._old_value
    def __repr__(self):
        r = '<Token '
        if self._used:
            r += ' used'
        r += ' var={!r} at {:0x}>'.format(self._var, id(self))
        return r

可以看到

Token

context

contextvar.context

Token

Token

4.3 全局唯一context

前面说过, Python中由

threading.local()

threading.local()

_state

threading.local()

_state

context

copy_context

threading.local()

def copy_context():
    return _get_context().copy()
def _get_context():
    ctx = getattr(_state, 'context', None)
    if ctx is None:
        ctx = Context()
        _state.context = ctx
    return ctx
def _set_context(ctx):
    _state.context = ctx
_state = threading.local()

关于

threading.local()

contextvars

threading.local()

threading.local()

在一个线程里面使用线程的局部变量会比直接使用全局变量的性能好，因为局部变量只有线程自己能看见，不会影响其他线程，而全局变量的修改必须加锁, 性能会变得很差, 比如下面全局变量的例子:

pet_dict = {}
def get_pet(pet_name):
    return pet_dict[pet_name]
def set_pet(pet_name):
    return pet_dict[pet_name]

这份代码就是模仿一个简单的全局变量调用, 如果是多线程调用的话, 那就需要加锁啦, 每次在读写之前都要等到持有锁的线程放弃了锁后再去竞争, 而且还可能污染到了别的线程存放的数据.

而线程的局部变量则是让每个线程有一个自己的

pet_dict

get_pet

set_pet

pet_dict = {}
def get_pet(pet_name, pid):
    return pet_dict[pid][pet_name]
def set_pet(pet_name, pid):
    return pet_dict[pid][pet_name]

不过这样子使用起来非常方便, 同时示例例子没有对异常检查和初始化等处理, 如果值比较复杂, 我们还要维护异常状况， 这样太麻烦了.

这时候

threading.local()

threading.local()

import threading
thread_local=threading.local()
def get_pet(pet_name):
    return thread_local[pet_name]
def set_pet(pet_name):
    return thread_local[pet_name]

可以看到代码就像调用全局变量一样， 但是又不会产生竞争状态。

4.4contextvar自己封装的Context

contextvars

dict

class Context(collections.abc.Mapping, metaclass=ContextMeta):
    def __init__(self):
        self._data = immutables.Map()
        self._prev_context = None
    def run(self, callable, *args, **kwargs):
        if self._prev_context is not None:
            raise RuntimeError(
                'cannot enter context: {} is already entered'.format(self))
        self._prev_context = _get_context()
        try:
            _set_context(self)
            return callable(*args, **kwargs)
        finally:
            _set_context(self._prev_context)
            self._prev_context = None
    def copy(self):
        new = Context()
        new._data = self._data
        return new

首先, 在

__init__

查看immutables.Map()的示例代码可以看到,每次对原对象的修改时,原对象并不会发生改变,并会返回一个已经发生改变的新对象.

map2 = map.set('a', 10)
print(map, map2)
# will print:
#   <immutables.Map({'a': 1, 'b': 2})>
#   <immutables.Map({'a': 10, 'b': 2})>
map3 = map2.delete('b')
print(map, map2, map3)
# will print:
#   <immutables.Map({'a': 1, 'b': 2})>
#   <immutables.Map({'a': 10, 'b': 2})>
#   <immutables.Map({'a': 10})>

此外,context还有一个叫

run

loop.run_in_executor

run

immutables.Map

run

def run(self, callable, *args, **kwargs):
    # 已经存在旧的context,抛出异常,防止多线程循环调用
    if self._prev_context is not None:
        raise RuntimeError(
            'cannot enter context: {} is already entered'.format(self))
    self._prev_context = _get_context()  # 保存当前的context
    try:
        _set_context(self) # 设置新的context
        return callable(*args, **kwargs)  # 执行函数
    finally:
        _set_context(self._prev_context)  # 设置为旧的context
        self._prev_context = None

4.5 ContextVar

我们一般在使用contextvars模块时,经常使用的就是

ContextVar

Context

set – 为当前上下文设置变量

def set(self, value):
    ctx = _get_context()  # 获取当前上下文对象`Context`
    data = ctx._data
    try:
        old_value = data[self]  # 获取Context旧对象
    except KeyError:
        old_value = Token.MISSING  # 获取不到则填充一个object(全局唯一)
    updated_data = data.set(self, value) # 设置新的值
    ctx._data = updated_data
    return Token(ctx, self, old_value) # 返回带有旧值的token

get – 从当前上下文获取变量

def get(self, default=_NO_DEFAULT):
    ctx = _get_context()  # 获取当前上下文对象`Context`
    try:
        return ctx[self]  # 返回获取的值
    except KeyError:
        pass
    if default is not _NO_DEFAULT:
        return default    # 返回调用get时设置的值
    if self._default is not _NO_DEFAULT:
        return self._default  # 返回初始化context时设置的默认值
    raise LookupError  # 都没有则会抛错

reset – 清理本次用到的上下文数据

def reset(self, token):
       if token._used:
       	# 判断token是否已经被使用
           raise RuntimeError("Token has already been used once")
       if token._var is not self:
       	# 判断token是否是当前contextvar返回的
           raise ValueError(
               "Token was created by a different ContextVar")
       if token._context is not _get_context():
       	# 判断token的上下文是否跟contextvar上下文一致
           raise ValueError(
               "Token was created in a different Context")
       ctx = token._context
       if token._old_value is Token.MISSING:
       	# 如果没有旧值则删除该值
           ctx._data = ctx._data.delete(token._var)
       else:
       	# 有旧值则当前contextvar变为旧值
           ctx._data = ctx._data.set(token._var, token._old_value)
       token._used = True  # 设置flag,标记token已经被使用了

则此,contextvar的原理了解完了,接下来再看看他是如何在asyncio运行的.

5.contextvars asyncio

由于向下兼容的

contextvars

5.1在asyncio中获取context

相比起contextvars复杂的概念,在asyncio中,我们可以很简单的获取到当前协程的task, 然后通过task就可以很方便的获取到task的context了,由于Pyhon3.7对asyncio的高级API 重新设计,所以可以看到需要对获取当前task进行封装

PY37 = sys.version_info >= (3, 7)
if PY37:
    def asyncio_current_task(loop=None):
        """Return the current task or None."""
        try:
            return asyncio.current_task(loop)
        except RuntimeError:
            # simulate old behaviour
            return None
else:
    asyncio_current_task = asyncio.Task.current_task

不同的版本有不同的获取task方法， 之后我们就可以通过调用

asyncio_current_task().context

5.2 对上下文的操作

同样的,在得到上下文后, 我们这里也需要set, get, reset的操作,不过十分简单, 类似dict一样的操作即可, 它没有token的逻辑：

set

def set(key, value):
    """
    Sets the given value inside Task.context[key]. If the key does not exist it creates it.
    :param key: identifier for accessing the context dict.
    :param value: value to store inside context[key].
    :raises
    """
    current_task = asyncio_current_task()
    if not current_task:
        raise ValueError(NO_LOOP_EXCEPTION_MSG.format(key))

    current_task.context[key] = value

get

def get(key, default=None):
    """
    Retrieves the value stored in key from the Task.context dict. If key does not exist,
    or there is no event loop running, default will be returned
    :param key: identifier for accessing the context dict.
    :param default: None by default, returned in case key is not found.
    :return: Value stored inside the dict[key].
    """
    current_task = asyncio_current_task()
    if not current_task:
        raise ValueError(NO_LOOP_EXCEPTION_MSG.format(key))
    return current_task.context.get(key, default)

clear – 也就是

contextvar.ContextVars

def clear():
    """
    Clear the Task.context.
    :raises ValueError: if no current task.
    """
    current_task = asyncio_current_task()
    if not current_task:
        raise ValueError("No event loop found")
    current_task.context.clear()

5.2 copying_task_factory和chainmap_task_factory

在Python的更高级版本中,已经支持设置context了,所以这两个方法可以不再使用了.他们最后都用到了

task_factory

task_factory

def task_factory(loop, coro, copy_context=False, context_factory=None):
    """
    By default returns a task factory that uses a simple dict as the task context,
    but allows context creation and inheritance to be customized via ``context_factory``.
    """
    # 生成context工厂函数
    context_factory = context_factory or partial(
        dict_context_factory, copy_context=copy_context)
    # 创建task, 跟asyncio.ensure_future一样
    task = asyncio.tasks.Task(coro, loop=loop)
    if task._source_traceback:
        del [-1]

    # 获取task的context
    try:
        context = asyncio_current_task(loop=loop).context
    except AttributeError:
        context = None
    # 从context工厂中处理context并赋值在task
    task.context = context_factory(context)
    return task

aiotask-context

dict_context_factory

chainmap_context_factory

aiotask-context

dict_context_factory

def dict_context_factory(parent_context=None, copy_context=False):
    """A traditional ``dict`` context to keep things simple"""
    if parent_context is None:
        # initial context
        return {}
    else:
        # inherit context
        new_context =&