Skip to content

Conversation

@demetere
Copy link

@demetere demetere commented May 30, 2024

Hello!

I'm seeking an alternative solution for a specific use case involving async generators. Our work heavily relies on async generators, and I needed a way to decorate an async generator with as_async_result to yield a Result for every yielded object. To make this functionality broadly applicable, I extended it to work with normal generators as well.

Here's an example:

import asyncio
from typing import AsyncGenerator

from result import Ok, as_async_result

@as_async_result(Exception)
async def random_generator() -> AsyncGenerator[int, None]:
    await asyncio.sleep(1)
    yield 1
    yield 2

async def normal_run():
    async for i in random_generator():
        assert isinstance(i, Ok)

@as_async_result(Exception)
async def error_generator() -> AsyncGenerator[Exception, None]:
    await asyncio.sleep(1)
    yield 1
    yield Exception('Error')

async def error_run():
    result = [i async for i in error_generator()]
    assert isinstance(result[0], Ok)
    assert isinstance(result[1], Exception)
    # here we can unwrap and in case of error reraise error so we can proceed to upstream or do any logic

if __name__ == '__main__':
    asyncio.run(normal_run())

For context, in our real-world scenario, we create a MongoDB Async Cursor to fetch documents asynchronously and yield them. Using this approach, if an error occurs during fetching, such as a connection error, we can safely stop the process and propagate the error upstream.

What do you think?

@montasaurus montasaurus mentioned this pull request Jan 8, 2025
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant