如何用迭代器包装 asyncio

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:
    ...

我不知道如何填写???上面的内容。我想产生记录数据，但是如何包装asyncio代码真的不是很明显。

Answer 1

虽然 asyncio 确实旨在全面使用，但有时根本不可能立即将大型软件（及其所有依赖项）转换为异步。幸运的是，有一些方法可以将遗留同步代码与新编写的 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(...)).