result icon indicating copy to clipboard operation
result copied to clipboard

adapt `as_result` and `as_async_result` so they can accept generators

Open demetere opened this issue 1 year ago • 0 comments

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?

demetere avatar May 30 '24 16:05 demetere