result
result copied to clipboard
adapt `as_result` and `as_async_result` so they can accept generators
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?