检测作为子例程调用的python协程

Question

假设我有一堆协程。应该使用yield from调用这些协程。我如何发现我犯了一个错误，并像子例程一样直接调用协程？

下面是一个示例代码：

import asyncio

@asyncio.coroutine
def A(msg):
    print(msg)
    yield from asyncio.sleep(1)

@asyncio.coroutine
def B():
    while True :
        yield from A('1')
        A('2')
        yield from A('3')

loop = asyncio.get_event_loop()
loop.run_until_complete(B())

和输出：

1
3
1
3
1
3
...

像子例程一样调用协程什么也不做，但不会引发异常或阻塞事件循环，因此失败模式非常安静。

Answer 1

我在您的代码中添加了一个check_iterator函数。如果你用这个函数装饰你的协程，如果你的协程被直接调用并且不能使用__iter__访问，它将打印出信息。不过，这可能会有一些性能上的下降。

import asyncio
import functools

class IterWrap(object):
    def __init__(self, f, *args, **kwds):
        self.iterated = False
        self.info = [f, args, kwds]
        self.f = f(*args, **kwds)

    def __iter__(self):
        self.iterated = True
        return iter(self.f)

    def __del__(self):
        if not self.iterated:
            print("Did not yield: %s" % self.info)

def check_iterator(f):
    @functools.wraps(f)
    def wrapper(*args, **kwds):
        return IterWrap(f, *args, **kwds)
    return wrapper

@check_iterator
@asyncio.coroutine
def A(msg):
    print(msg)
    yield from asyncio.sleep(1)

@asyncio.coroutine
def B():
    while True :
        yield from A('1')
        A('2')
        yield from A('3')

loop = asyncio.get_event_loop()
loop.run_until_complete(B())

在Python 3.4上运行此命令的结果如下：

1
Did not yield: [<function A at 0xb6e3189c>, ('2',), {}]
3
1
Did not yield: [<function A at 0xb6e3189c>, ('2',), {}]
3
1
Did not yield: [<function A at 0xb6e3189c>, ('2',), {}]
3
1
Did not yield: [<function A at 0xb6e3189c>, ('2',), {}]
3