python如何使用contextvars模块源码分析_python_

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

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

# 伪代码,举个例子一个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')

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)

# 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'})

# 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'})

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(conte
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：pytest自动化测试数据驱动yaml/excel/csv/json_python_
                                                                
下一篇：Python实现自动化处理每月考勤缺卡数据_python_