asyncio TaskGroup for concurrent async operations

<p>I need to run multiple async operations concurrently in Python and collect all their results. Some operations may fail and I want to handle errors per-task. I'm using Python 3.11+ and want to use the modern approach.</p>

<p>Use <code>asyncio.TaskGroup</code> (Python 3.11+) for structured concurrency:</p> <div class="highlight"><pre><span></span><code><span class="kn">import</span><span class="w"> </span><span class="nn">asyncio</span> <span class="kn">from</span><span class="w"> </span><span class="nn">typing</span><span class="w"> </span><span class="kn">import</span> <span class="n">Any</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">fetch_trace_details</span><span class="p">(</span><span class="n">trace_id</span><span class="p">:</span> <span class="nb">str</span><span class="p">)</span> <span class="o">-&gt;</span> <span class="nb">dict</span><span class="p">:</span> <span class="w"> </span><span class="sd">"""Concurrently fetch trace data, tags, and vote counts."""</span> <span class="k">async</span> <span class="k">with</span> <span class="n">asyncio</span><span class="o">.</span><span class="n">TaskGroup</span><span class="p">()</span> <span class="k">as</span> <span class="n">tg</span><span class="p">:</span> <span class="n">trace_task</span> <span class="o">=</span> <span class="n">tg</span><span class="o">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">get_trace</span><span class="p">(</span><span class="n">trace_id</span><span class="p">))</span> <span class="n">tags_task</span> <span class="o">=</span> <span class="n">tg</span><span class="o">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">get_tags</span><span class="p">(</span><span class="n">trace_id</span><span class="p">))</span> <span class="n">votes_task</span> <span class="o">=</span> <span class="n">tg</span><span class="o">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">get_vote_count</span><span class="p">(</span><span class="n">trace_id</span><span class="p">))</span> <span class="c1"># All tasks done here — any exception propagates as ExceptionGroup</span> <span class="k">return</span> <span class="p">{</span> <span class="s1">'trace'</span><span class="p">:</span> <span class="n">trace_task</span><span class="o">.</span><span class="n">result</span><span class="p">(),</span> <span class="s1">'tags'</span><span class="p">:</span> <span class="n">tags_task</span><span class="o">.</span><span class="n">result</span><span class="p">(),</span> <span class="s1">'votes'</span><span class="p">:</span> <span class="n">votes_task</span><span class="o">.</span><span class="n">result</span><span class="p">(),</span> <span class="p">}</span> <span class="c1"># For Python 3.10 and earlier, use asyncio.gather:</span> <span class="n">results</span> <span class="o">=</span> <span class="k">await</span> <span class="n">asyncio</span><span class="o">.</span><span class="n">gather</span><span class="p">(</span> <span class="n">get_trace</span><span class="p">(</span><span class="n">trace_id</span><span class="p">),</span> <span class="n">get_tags</span><span class="p">(</span><span class="n">trace_id</span><span class="p">),</span> <span class="n">get_vote_count</span><span class="p">(</span><span class="n">trace_id</span><span class="p">),</span> <span class="n">return_exceptions</span><span class="o">=</span><span class="kc">True</span><span class="p">,</span> <span class="p">)</span> <span class="c1"># Handle per-task errors with gather:</span> <span class="k">for</span> <span class="n">result</span> <span class="ow">in</span> <span class="n">results</span><span class="p">:</span> <span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">result</span><span class="p">,</span> <span class="ne">Exception</span><span class="p">):</span> <span class="n">log</span><span class="o">.</span><span class="n">error</span><span class="p">(</span><span class="s1">'Task failed'</span><span class="p">,</span> <span class="n">exc_info</span><span class="o">=</span><span class="n">result</span><span class="p">)</span> <span class="k">else</span><span class="p">:</span> <span class="n">process</span><span class="p">(</span><span class="n">result</span><span class="p">)</span> </code></pre></div> <p>Key points: - <code>TaskGroup</code> cancels all tasks if any raises — use <code>gather(return_exceptions=True)</code> for independent failures - <code>ExceptionGroup</code> (Python 3.11+) wraps multiple failures — catch with <code>except*</code> - Use <code>asyncio.timeout()</code> inside tasks to prevent indefinite hangs - <code>TaskGroup</code> is preferred over <code>gather</code> when all tasks must succeed together</p>