上下文管理器，它提供了一种设置上下文并在python asyncio中检索上下文的方法。
aio-task-bound-context的Python项目详细描述
aio任务绑定上下文
上下文管理器，提供设置和检索上下文的方法 在python异步中。
什么？？？
好的，举个具体的例子，flask如何处理当前请求：
fromflaskimportrequest
这个从任何地方调用的导入将导入当前请求 处理。这样做的方式与此类似：
request=Nonedefget_request():returnrequestdefset_request(value):globalrequestrequest=value
当http服务器收到请求时，它将调用set_request，然后调用
在代码中，另一个函数可以调用get_request来获取值。
关键是：这在aio中是不可能的，因为多个任务可能
同时运行，因此request有多个值，而不是
只有一个值。想象一下aio中使用的同一段代码：
importasyncioasaioasyncdefhandle_request(request):set_request(request)# generate the responseawaitaio.sleep(1)assertget_request()==request# will failset_request(None)aio.get_event_loop().run_until_complete(aio.gather(handle_request('value 1'),handle_request('value 2'),))
显然，这将是个问题。
答案
aio_task_bound_context将当前上下文值的堆栈附加到
当前的Task，以及跟踪父任务以使其上下文
可输入：
importasyncioasaiofromaio_task_bound_contextimportset_task_factory,TaskBoundContextclassRequestContext(TaskBoundContext):def__init__(self,request):self.request=requestdefget_value(self):returnself.requestasyncdefhandle_request(request):withRequestContext(request):# generate the responseawaitaio.sleep(1)assertRequestContext.current()==request# will succeedloop=aio.get_event_loop()set_task_factory(loop=loop)loop.run_until_complete(aio.gather(handle_request('value 1'),handle_request('value 2'),))
示例
请注意，所有这些示例都将在异步任务中工作，这就是 它们比简单的上下文管理器更特殊。它们都是简单的例子 在一个异步环境之外，但不要被隐藏的复杂性所欺骗。
首先，我们需要将asyncio中的默认任务工厂替换为
为任务添加额外详细信息的包装器。假设这是以前执行过的
所有示例：
importasyncioasaiofromaio_task_bound_contextimportcreate_task_factory,TaskBoundContextloop=aio.get_event_loop()loop.set_task_factory(create_task_factory(loop=loop))
如果没有定义get_value函数，“值”是TaskBoundContext。
它本身，因此如果您只想在__init__函数中设置值
作为一组值传递。
classExampleContext(TaskBoundContext):def__init__(self,*args,**kwargs):super().__init__()self.args=argsself.kwargs=kwargswithExampleContext('an arg',key='in kwargs'):assertExampleContext.current().args==['an arg']assertExampleContext.current().kwargs=={'key':'in kwargs'}
上下文管理器的“as value”是从get_value返回的值。
classExampleContext(TaskBoundContext):def__init__(self,value):super().__init__()self.value=valuedefget_value(self):returnself.valuewithExampleContext('test')asvalue:assertvalue=='test'withExampleContext('different')asvalue:assertvalue=='different'
上下文是一个层次堆栈，因此可以有多个上下文 将其值推送到/弹出上下文堆栈。
classExampleContext(TaskBoundContext):def__init__(self,value):super().__init__()self.value=valuedefget_value(self):returnself.valuewithExampleContext('first'):assertExampleContext.current()=='first'withExampleContext('second'):assertExampleContext.current()=='second'assertExampleContext.current()=='first
get_value函数可以接受单个参数，即
堆栈的值。
classLoggerContext(TaskBoundContext):def__init__(self,suffix):super().__init__()self.suffix=suffixdefget_value(self,current):ifcurrentisNone:returnlogging.getLogger(self.suffix)else:returncurrent.getChild(self.suffix)
测试
支持Python3.5+。为了在所有环境中运行测试，我们使用
pyenv，以及一些快速的virtualenv调用（是的，我们还可以使用
tox）。
要运行测试，只需运行./tests_runall.sh，它将安装相关的
如果尚未安装python版本，请为它们创建virtualenvs，以及
运行tests.py。
要手动运行测试，只需./test.py。
许可证
版权所有2018 Ricky Cook
特此免费授予获得本软件副本和相关文档文件（以下简称“软件”）的任何人使用本软件的权利，包括但不限于使用、复制、修改、合并、发布、分发、再授权和/或出售本软件副本的权利，并在符合以下条件的情况下，允许向其提供软件的人员这样做：
上述版权声明和本许可声明应包含在软件的所有副本或实质性部分中。
<软件>提供“原样”，无任何担保，明示或默示，包括但不限于适销性、适合特定用途和不侵权的保证。在任何情况下，无论是在合同、侵权行为或其他情况下，作者或著作权人都不应因软件、软件的使用或其他交易而产生或与之相关的索赔、损害或其他责任承担责任。欢迎加入QQ群-->： 979659372
