Consuming TaskGroup response

Question

In Python3.11 it's suggested to use TaskGroup for spawning Tasks rather than using gather. Given Gather will also return the result of a co-routine, what's the best approach with TaskGroup.

Currently I have

async with TaskGroup() as tg:
      r1 = tg.create_task(foo())
      r2 = tg.create_task(bar())
res = [r1.result(), r2.result()]

Is there a more concise approach that can be used to achieve the same result?

Answer 1

The task groups were implemented more as a cleaner way to handle task lifetimes and exception handling, enabled greatly by the new exceptions group rework. I think its a popular misconception right now, but the TaskGroup is not a drop in replacement for all of the gather use-cases. For cases where you do not care about the results (which seems to be the only example I am seeing in new documentation and tutorials) it feels much more terse.

When the task group has completed, it's still required by the user to pull results out of the completed coruntines. If you need values immediately then you can write it as you have it as res = [r1.result(), r2.result()] after the TaskGroup completes. A more terse syntax might be to gather the results after the TaskGroup completes with res = await asyncio.gather(r1, r2) (this will release the execution of your function which may or may not be desirable).

This may look redundant to use both TaskGroup and gather, but the TaskGroup is solving a different purpose than what is provided by gather alone, being that it allows for waiting for your tasks with strong safety guarantees, logic around cancellation for failures, and grouping of exceptions.

It might be possible to extend the default TaskGroup class to make this pattern easier. Here's one such idea that can keep track of which tasks were issued in the task group and provides a helper to fish out the results:

class GatheringTaskGroup(asyncio.TaskGroup):
    def __init__(self):
        super().__init__()
        self.__tasks = []

    def create_task(self, coro, *, name=None, context=None):
        task = super().create_task(coro, name=name, context=context)
        self.__tasks.append(task)
        return task

    def results(self):
        return [task.result() for task in self.__tasks]

async def foo(): return 1
async def bar(): return 2
async with GatheringTaskGroup() as tg:
    task1 = tg.create_task(foo())
    task2 = tg.create_task(bar())
print(tg.results())

[1, 2]