metapensiero.asyncio.tasklocal 0.2

使用量

替换存储threading.local对象的drop-in 每个任务变量（而不是每个线程）。您可以这样使用：

import asyncio
from metapensiero.asyncio.tasklocal import Local

mylocal = Local()

@asyncio.coroutine
def task_a():
    mylocal.a_var = 'foo'
    yield from asyncio.sleep(0.001) # switches to task_b
    assert mylocal.a_var == 'foo'
    print(mylocal.a_var)

@asyncio.coroutine
def task_b():
    mylocal.a_var = 'bar'
    yield from asyncio.sleep(0.001) # switches back to task_a
    assert mylocal.a_var == 'bar'
    print(mylocal.a_var)

loop = asyncio.get_event_loop()
a = asyncio.Task(task_a())
b = asyncio.Task(task_b())
loop.run_until_complete(asyncio.wait((a, b)))

# this prints out:
# foo
# bar

实际上，被跟踪为“当前”的对象类型是可配置的 通过将不同的Discriminator实例传递给Local 上课。

默认情况下，它将使用 包裹。它将使用标准的asyncio循环检测技术 （通过将其委托给asyncio.Task.current_task()）但是如果您需要的话 要跟踪特定事件循环的任务，可以提供自己的 TaskDiscriminator实例到Local实例 初始化，如下所示：

from metapensiero.asyncio.tasklocal import Local, TaskDiscriminator

mylocal = Local(discriminator=TaskDiscriminator(loop=my_loop))

您还可以通过子类化 BaseDiscriminator类：

from metapensiero.asyncio.tasklocal import BaseDiscriminator

class ASampleSessionDiscriminator(BaseDiscriminator):

    def current(self):
        return current_session()

mylocal = Local(discriminator=ASampleDiscriminator())

您必须至少实现current方法。

像threading.local类一样，这个类支持 在这种情况下，它将为每个 被跟踪的对象：

call_counter = 0

class CustomLocal(Local):

    a_value = None

    def __init__(self, a_value):
        self.a_value = a_value
        nonlocal call_counter
        call_counter += 1

mylocal = CustomLocal('foo')

在这里，mylocal.a_value将被初始化为 跟踪对象（默认情况下为Asyncio的任务）。这里call_counter 将计算每个跟踪对象的数量，其中mylocal 对象已被访问。

测试

要运行测试，应在包根目录下运行以下命令：

python setup.py test

构建状态