如何用迭代器包装异步

Question

我有以下简化代码：

async def asynchronous_function(*args, **kwds):
    statement = await prepare(query)
    async with conn.transaction():
        async for record in statement.cursor():
            ??? yield record ???

...

class Foo:

    def __iter__(self):
        records = ??? asynchronous_function ???
        yield from records

...

x = Foo()
for record in x:
    ...

上面的???不知道怎么填。我想产生记录数据，但是如何包装异步代码真的不明显。

Answer 1

虽然 asyncio 确实旨在全面使用，但有时根本不可能立即将大型软件（及其所有依赖项）转换为 async。幸运的是，有一些方法可以将遗留的同步代码与新编写的 asyncio 部分结合起来。一种直接的方法是在专用线程中运行事件循环，并使用asyncio.run_coroutine_threadsafe 向它提交任务。

使用这些低级工具，您可以编写一个通用适配器，将任何异步迭代器转换为同步迭代器。例如：

import asyncio, threading, queue

# create an asyncio loop that runs in the background to
# serve our asyncio needs
loop = asyncio.get_event_loop()
threading.Thread(target=loop.run_forever, daemon=True).start()

def wrap_async_iter(ait):
    """Wrap an asynchronous iterator into a synchronous one"""
    q = queue.Queue()
    _END = object()

    def yield_queue_items():
        while True:
            next_item = q.get()
            if next_item is _END:
                break
            yield next_item
        # After observing _END we know the aiter_to_queue coroutine has
        # completed.  Invoke result() for side effect - if an exception
        # was raised by the async iterator, it will be propagated here.
        async_result.result()

    async def aiter_to_queue():
        try:
            async for item in ait:
                q.put(item)
        finally:
            q.put(_END)

    async_result = asyncio.run_coroutine_threadsafe(aiter_to_queue(), loop)
    return yield_queue_items()

那么您的代码只需调用wrap_async_iter 将异步迭代器包装到同步迭代器中：

async def mock_records():
    for i in range(3):
        yield i
        await asyncio.sleep(1)

for record in wrap_async_iter(mock_records()):
    print(record)

在您的情况下，Foo.__iter__ 将使用 yield from wrap_async_iter(asynchronous_function(...))。

Answer 2

如果您想接收来自异步生成器的所有记录，您可以使用async for，或者简而言之，asynchronous comprehensions：

async def asynchronous_function(*args, **kwds):
    # ...
    yield record


async def aget_records():
    records = [
        record 
        async for record 
        in asynchronous_function()
    ]
    return records

如果你想同步获取异步函数的结果（即阻塞），你可以运行这个函数in asyncio loop：

def get_records():
    records = asyncio.run(aget_records())
    return records

---

但是，请注意，一旦您在事件循环中运行某个协程，您将失去与其他协程同时（即并行）运行该协程的能力，从而获得所有相关的好处。

正如 Vincent 在 cmets 中已经指出的那样，asyncio 不是让代码更快的魔杖，它是一种有时可用于以低开销同时运行不同 I/O 任务的工具。

您可能有兴趣阅读 this answer 以了解 asyncio 背后的主要思想。