python - 使用 asyncio 协同程序进行方法链接

Question

我想实现 method chaining ，但不适用于通常的功能 - 对于 asyncio 协程。

import asyncio


class Browser:
    @asyncio.coroutine
    def go(self):
        # some actions
        return self

    @asyncio.coroutine
    def click(self):
        # some actions
        return self

调用链的“直观”方式行不通，因为单个方法返回协程(生成器)，而不是自身:

@asyncio.coroutine
def main():
    br = yield from Browser().go().click()  # this will fail

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

调用链的正确方式是:

br = yield from (yield from Browser().go()).click()

但是当链增长时它看起来很难看并且变得不可读。

有什么办法可以做得更好吗？欢迎任何想法。

Answer 1

我创建了解决方案，可以完成所需的工作。想法是使用 Browser() 的包装器，它使用 __getattr__ 和 __call__ 来收集 Action (比如获取属性或调用)并返回 self 以捕获下一个 Action 。收集完所有 Action 后，我们使用 __iter__“捕获”yiled from wrapper 并处理所有收集到的 Action 。

import asyncio


def chain(obj):
    """
    Enables coroutines chain for obj.
    Usage: text = yield from chain(obj).go().click().attr
    Note: Returns not coroutine, but object that can be yield from.
    """
    class Chain:
        _obj = obj
        _queue = []

        # Collect getattr of call to queue:
        def __getattr__(self, name):
            Chain._queue.append({'type': 'getattr', 'name': name})
            return self

        def __call__(self, *args, **kwargs):
            Chain._queue.append({'type': 'call', 'params': [args, kwargs]})
            return self

        # On iter process queue:
        def __iter__(self):
            res = Chain._obj
            while Chain._queue:
                action = Chain._queue.pop(0)
                if action['type'] == 'getattr':
                    res = getattr(res, action['name'])
                elif action['type'] == 'call':
                    args, kwargs = action['params']
                    res = res(*args, **kwargs)
                if asyncio.iscoroutine(res):
                    res = yield from res
            return res
    return Chain()

用法:

class Browser:
    @asyncio.coroutine
    def go(self):
        print('go')
        return self

    @asyncio.coroutine
    def click(self):
        print('click')
        return self

    def text(self):
        print('text')
        return 5


@asyncio.coroutine
def main():
    text = yield from chain(Browser()).go().click().go().text()
    print(text)


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

输出:

go
click
go
text
5

请注意，chain() 不返回真正的协程，而是返回可在 yield from 上像协程一样使用的对象。我们应该包装 chain() 的结果以获得正常的协程，它可以传递给任何需要协程的异步函数:

@asyncio.coroutine
def chain_to_coro(chain):
    return (yield from chain)


@asyncio.coroutine
def main():
    ch = chain(Browser()).go().click().go().text()
    coro = chain_to_coro(ch)

    results = yield from asyncio.gather(*[coro], return_exceptions=True)
    print(results)

输出:

go
click
go
text
[5]