Python asyncio TaskGroup for concurrent structured operations

<p>I need to run multiple async operations concurrently and collect all results. I want to handle cases where some may fail independently and need Python 3.11+ structured concurrency approach.</p>

<p>asyncio.TaskGroup 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="c1"># TaskGroup -- structured concurrency (Python 3.11+)</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="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 done -- 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 or when tasks are independent:</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="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="c1"># Handle ExceptionGroup (Python 3.11+):</span> <span class="k">try</span><span class="p">:</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">task1</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">fetch_traces</span><span class="p">())</span> <span class="n">task2</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">fetch_users</span><span class="p">())</span> <span class="k">except</span><span class="o">*</span> <span class="ne">ValueError</span> <span class="k">as</span> <span class="n">eg</span><span class="p">:</span> <span class="k">for</span> <span class="n">exc</span> <span class="ow">in</span> <span class="n">eg</span><span class="o">.</span><span class="n">exceptions</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">'Validation error'</span><span class="p">,</span> <span class="n">exc_info</span><span class="o">=</span><span class="n">exc</span><span class="p">)</span> </code></pre></div> <p>Key points: - TaskGroup cancels all tasks if any raises -- use gather(return_exceptions=True) for independent failures - ExceptionGroup wraps multiple failures -- catch with except* syntax - asyncio.timeout() inside tasks prevents indefinite hangs - TaskGroup preferred over gather when all tasks must succeed together</p>