Aioresult: Capture the result of a Trio or anyio task

Show off your work
1

In the spirit of “show off your work”, I present to you: aioresult, a tiny library for capturing the result of a task. Works with Trio directly, or also anyio but anyio isn’t required.

From the docs:

The main class of aioresult is the ResultCapture class. If you are directly awaiting a task then there is no need to use this class – you can just use the return value:

result1 = await foo(1)
result2 = await foo(2)
print("results:", result1, result2)

If you want to run your tasks in parallel then you would typically use a nursery, but then it’s harder to get hold of the results:

async with trio.open_nursery() as n:
    n.start_soon(foo, 1)
    n.start_soon(foo, 2)
# At this point the tasks have completed, but the results are lost
print("results: ??")

To get access to the results, the usual advice is to either modify the routines so that they store their result somewhere rather than returning it or to create a little wrapper function that stores the return value of the function you actually care about. ResultCapture is a simple helper to do this:

async with trio.open_nursery() as n:
    result1 = ResultCapture.start_soon(n, foo, 1)
    result2 = ResultCapture.start_soon(n, foo, 2)
# At this point the tasks have completed, and results are stashed in ResultCapture objects 
print("results", result1.result(), result2.result())

You can get very similar effect to asyncio.gather() by using a nursery and an array of ResultCapture objects:

async with trio.open_nursery() as n:
    results = [ResultCapture.start_soon(n, foo, i) for i in range(10)]
print("results:", *[r.result() for r in results])

Unlike asyncio’s gather, you benefit from the safer behaviour of Trio nurseries if one of the tasks throws an exception. ResultCapture is also more flexible because you don’t have to use a list, for example you could use a dictionary:

async with trio.open_nursery() as n:
    results = {i: ResultCapture.start_soon(n, foo, i) for i in range(10)}
    print("results:", *[f"{i} -> {r.result()}," for i, r in results.items()])

Any exception thrown by the task will propagate out as usual, typically to the enclosing nursery. See Exception handling and design rationale for details.

It’s just meant to save a bit of boilerplate code rather than do anything really clever, so the implementation is dead simple. Rather than running your routine directly in the nursery, you run ResultCapture.run() instead, which is effectively implemented like this:

class ResultCapture:
    def run(self):
        try:
            self._result = await self._fn(*self.args)
        except BaseException as e:
            self._exception = e
            raise
        finally:
            self._done_event.set()

The ResultCapture.start_soon() class method is a trivial helper that just constructs a
ResultCapture instance and then calls nursery.start_soon(rc.run).

There’s also a very basic Future class because it was implemented almost automatically in the
process of writing ResultCapture.

Finally, there’s another class StartableResultCapture for also capturing the start value of a
task. I’ve called this release 0.9 because I’ve realised that I can replace StartableResultCapture
with a very short helper function, so I’ll make that change and release as 1.0.

Any comments or feedback would be very gratefully received. As would thoughts on whether this could
go in Awesome Trio.

2

I finally released 1.0 - a year later than I’d planned! - with that simplification to startable tasks I mentioned. Probably more interesting is that I added three utility functions for waiting: wait_any(), wait_all() and results_to_channel(). Here’s an example of using the first two:

async with trio.open_nursery() as n:
    results = [ResultCapture.start_soon(n, foo, i) for i in range(10)]

    first_completed = await wait_any(results)
    # One task (at least) is now done
    print("task completed:", first_completed.args[0], "result:", first_completed.result())

    even_results = results[::2]
    await wait_all(even_results)
    # All tasks in even positions are now done.
    print("even results:", *[r.result() for r in even_results]

# All tasks are now done.
print("all results:", *[r.result() for r in results])