Function iter_over_async

robohusky/utils.py:199–222

Convert async generator to sync generator :param async_gen: the AsyncGenerator to convert :param event_loop: the event loop to run on :returns: Sync generator

(
        async_gen: AsyncGenerator, event_loop: AbstractEventLoop
)

def iter_over_async(
        async_gen: AsyncGenerator, event_loop: AbstractEventLoop
) -> Generator:
    """
    Convert async generator to sync generator

    :param async_gen: the AsyncGenerator to convert
    :param event_loop: the event loop to run on
    :returns: Sync generator
    """
    ait = async_gen.__aiter__()

    async def get_next():
        try:
            obj = await ait.__anext__()
            return False, obj
        except StopAsyncIteration:
            return True, None

    while True:
        done, obj = event_loop.run_until_complete(get_next())
        if done:
            break
        yield obj


def detect_language(text: str) -> str: