Translating async generator into sync one

Question

Imagine we have an original API that returns a generator (it really is a mechanisms that brings pages/chunks of results from a server while the providing a simple generator to the user, and lets him iterate over these results one by one. For simplicity:

# Original sync generator
def get_results():
     # fetch from server
     yield 1
     yield 2
     # fetch next page
     yield 3
     yield 4
     # ....

Now there is a need to implement an asyncio version of the API, however we need to keep the old API operational as well. This is where things get complicated, we kind of want to translate an async generator into sync one, but I can't find an elegant way to do that. The best I could make work so far is "fetch all result into a list first, then provide a fake sync generator on that list". Which kind of defeats the purpose:

# Async generator
async def get_results_async():
     # await fetch from server
     yield 1
     yield 2
     # await fetch next page
     yield 3
     yield 4
     # ....


# Backward compatible sync generator
def get_results():

    async def gather_all_results():
        res = []
        async for i in get_results_async():
            res.append(i)
        return res

    res = asyncio.run(gather_all_results())
    for i in res:
        yield i

Is there a better, more elegant way to do that without fetching all the results before returning them?

Thanks

Answer 1

For the reason that asyncio is contagious, it's hard to write elegant code to integrate asyncio code into the old codes. For the scenario above, the flowing code is a little better, but I don't think it's elegant enough.

async def get_results_async():
    # await fetch from server
    yield 1
    yield 2
    # await fetch next page
    yield 3
    yield 4
    # ....


# Backward compatible sync generator
def get_results():
    gen = get_results_async()
    while True:
        try:
            yield asyncio.run(gen.__anext__())
        except StopAsyncIteration:
            break 

And you can re-use your event loop and not to create a new one.

async def get_results_async():
    # await fetch from server
    yield 1
    yield 2
    # await fetch next page
    yield 3
    yield 4
    # ....

# loop that you save in somewhere.
loop = asyncio.get_event_loop()

# Backward compatible sync generator
def get_results():
    gen = get_results_async()
    while True:
        try:
            yield loop.run_until_complete(gen.__anext__())
        except StopAsyncIteration:
            break 

Answer 2

The problem with the accepted answer is that it ceases to work. Using asyncio.run(...) will create a new event loop. asyncio.run(anext(...)) will run anext in a fresh event loop until anext returns. If your async generator calls other coroutine functions (async def), these calls will be placed in the event loop. But when anext(...) returns, asyncio.run(...) will return and hence destroy the event loop. That will leave the intermediate results hanging.

Take a look at the following minimal demonstrator code that uses the accepted answer's code:

import asyncio
from collections.abc import AsyncIterator, Iterator
from typing import TypeVar


T = TypeVar("T")


def async_iterable_to_sync_iterable(iterator: AsyncIterator[T]) -> Iterator[T]:
    try:
        while True:
            result = asyncio.run(anext(iterator))
            # result = asyncio.run(await_anext(async_iterable))
            print(f"🟡 async_iterable_to_sync_iterable: {result=}")
            yield result
    except StopAsyncIteration as e:
        print(f"🟡 async_iterable_to_sync_iterable: StopAsyncIteration")
        pass


async def dummy_relay(iterator: AsyncIterator[T]) -> AsyncIterator[T]:
    async for elem in iterator:
        print(f"🟡 dummy_relay: Relaying {elem} …")
        yield elem
    print(f"🟡 dummy_relay: Done.")


async def generate_numbers() -> AsyncIterator[int]:
    for i in range(3):
        print(f"🟡 generate_numbers: Yielding {i} …")
        yield i
    print(f"🟡 generate_numbers: Done.")


print(list(async_iterable_to_sync_iterable(generate_numbers())))
print()
print(list(async_iterable_to_sync_iterable(dummy_relay(generate_numbers()))))

In the output we only have this:

🟡 generate_numbers: Yielding 0 …
🟡 async_iterable_to_sync_iterable: result=0
🟡 generate_numbers: Yielding 1 …
🟡 async_iterable_to_sync_iterable: result=1
🟡 generate_numbers: Yielding 2 …
🟡 async_iterable_to_sync_iterable: result=2
🟡 generate_numbers: Done.
🟡 async_iterable_to_sync_iterable: StopAsyncIteration
[0, 1, 2]

🟡 generate_numbers: Yielding 0 …
🟡 dummy_relay: Relaying 0 …
🟡 async_iterable_to_sync_iterable: result=0
🟡 dummy_relay: Done.
🟡 async_iterable_to_sync_iterable: StopAsyncIteration
[0]

As you can see, generate_numbers never finishes.

In order to fix this, one needs to persist the event loop for a while. In Python ≥ 3.11, use with asyncio.Runner() to keep the event loop active during the whole duration of async_iterable_to_sync_iterable:

def async_iterable_to_sync_iterable(iterator: AsyncIterator[T]) -> Iterator[T]:
    with asyncio.Runner() as runner:
        try:
            while True:
                result = runner.run(anext(iterator))
                # result = asyncio.run(await_anext(async_iterable))
                print(f"🟡 async_iterable_to_sync_iterable: {result=}")
                yield result
        except StopAsyncIteration as e:
            print(f"🟡 async_iterable_to_sync_iterable: StopAsyncIteration")
            pass

And then the output becomes what we expect:

🟡 generate_numbers: Yielding 0 …
🟡 dummy_relay: Relaying 0 …
🟡 async_iterable_to_sync_iterable: result=0
🟡 generate_numbers: Yielding 1 …
🟡 dummy_relay: Relaying 1 …
🟡 async_iterable_to_sync_iterable: result=1
🟡 generate_numbers: Yielding 2 …
🟡 dummy_relay: Relaying 2 …
🟡 async_iterable_to_sync_iterable: result=2
🟡 generate_numbers: Done.
🟡 dummy_relay: Done.
🟡 async_iterable_to_sync_iterable: StopAsyncIteration
[0, 1, 2]