Python asyncio TaskGroup for concurrent operations

<p>Executing multiple independent async operations sequentially instead of concurrently. Using <code>asyncio.gather()</code> but need better error handling when one task fails — gather continues other tasks even on failure.</p>

<p>Use <code>asyncio.TaskGroup</code> (Python 3.11+) for structured concurrency with automatic cancellation:</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="c1"># asyncio.gather (old pattern)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">fetch_all_gather</span><span class="p">(</span><span class="n">ids</span><span class="p">:</span> <span class="nb">list</span><span class="p">[</span><span class="nb">str</span><span class="p">])</span> <span class="o">-&gt;</span> <span class="nb">list</span><span class="p">[</span><span class="nb">dict</span><span class="p">]:</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="o">*</span><span class="p">[</span><span class="n">fetch_trace</span><span class="p">(</span><span class="nb">id</span><span class="p">)</span> <span class="k">for</span> <span class="nb">id</span> <span class="ow">in</span> <span class="n">ids</span><span class="p">],</span> <span class="n">return_exceptions</span><span class="o">=</span><span class="kc">True</span> <span class="c1"># Gather swallows errors by default</span> <span class="p">)</span> <span class="c1"># Have to check each result manually</span> <span class="k">return</span> <span class="p">[</span><span class="n">r</span> <span class="k">for</span> <span class="n">r</span> <span class="ow">in</span> <span class="n">results</span> <span class="k">if</span> <span class="ow">not</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">r</span><span class="p">,</span> <span class="ne">Exception</span><span class="p">)]</span> <span class="c1"># TaskGroup (Python 3.11+) — preferred</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">fetch_all_taskgroup</span><span class="p">(</span><span class="n">ids</span><span class="p">:</span> <span class="nb">list</span><span class="p">[</span><span class="nb">str</span><span class="p">])</span> <span class="o">-&gt;</span> <span class="nb">list</span><span class="p">[</span><span class="nb">dict</span><span class="p">]:</span> <span class="n">results</span> <span class="o">=</span> <span class="p">[]</span> <span class="c1"># If ANY task raises, all others are cancelled immediately</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">tasks</span> <span class="o">=</span> <span class="p">[</span><span class="n">tg</span><span class="o">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">fetch_trace</span><span class="p">(</span><span class="nb">id</span><span class="p">))</span> <span class="k">for</span> <span class="nb">id</span> <span class="ow">in</span> <span class="n">ids</span><span class="p">]</span> <span class="c1"># All tasks completed successfully here</span> <span class="k">return</span> <span class="p">[</span><span class="n">task</span><span class="o">.</span><span class="n">result</span><span class="p">()</span> <span class="k">for</span> <span class="n">task</span> <span class="ow">in</span> <span class="n">tasks</span><span class="p">]</span> <span class="c1"># Real example: parallel database + Redis operations</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">get_trace_with_context</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">db</span><span class="o">.</span><span class="n">get_trace</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">db</span><span class="o">.</span><span class="n">get_votes</span><span class="p">(</span><span class="n">trace_id</span><span class="p">))</span> <span class="n">cached_views_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">redis</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="sa">f</span><span class="s1">'views:</span><span class="si">{</span><span class="n">trace_id</span><span class="si">}</span><span class="s1">'</span><span class="p">))</span> <span class="n">trace</span> <span class="o">=</span> <span class="n">trace_task</span><span class="o">.</span><span class="n">result</span><span class="p">()</span> <span class="n">votes</span> <span class="o">=</span> <span class="n">votes_task</span><span class="o">.</span><span class="n">result</span><span class="p">()</span> <span class="n">views</span> <span class="o">=</span> <span class="n">cached_views_task</span><span class="o">.</span><span class="n">result</span><span class="p">()</span> <span class="ow">or</span> <span class="mi">0</span> <span class="k">return</span> <span class="p">{</span><span class="o">**</span><span class="n">trace</span><span class="o">.</span><span class="n">dict</span><span class="p">(),</span> <span class="s1">'votes'</span><span class="p">:</span> <span class="n">votes</span><span class="p">,</span> <span class="s1">'view_count'</span><span class="p">:</span> <span class="nb">int</span><span class="p">(</span><span class="n">views</span><span class="p">)}</span> <span class="c1"># With timeout</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">fetch_with_timeout</span><span class="p">(</span><span class="n">ids</span><span class="p">:</span> <span class="nb">list</span><span class="p">[</span><span class="nb">str</span><span class="p">],</span> <span class="n">timeout</span><span class="p">:</span> <span class="nb">float</span> <span class="o">=</span> <span class="mf">5.0</span><span class="p">)</span> <span class="o">-&gt;</span> <span class="nb">list</span><span class="p">[</span><span class="nb">dict</span><span class="p">]:</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">timeout</span><span class="p">(</span><span class="n">timeout</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">tasks</span> <span class="o">=</span> <span class="p">[</span><span class="n">tg</span><span class="o">.</span><span class="n">create_task</span><span class="p">(</span><span class="n">fetch_trace</span><span class="p">(</span><span class="nb">id</span><span class="p">))</span> <span class="k">for</span> <span class="nb">id</span> <span class="ow">in</span> <span class="n">ids</span><span class="p">]</span> <span class="k">return</span> <span class="p">[</span><span class="n">t</span><span class="o">.</span><span class="n">result</span><span class="p">()</span> <span class="k">for</span> <span class="n">t</span> <span class="ow">in</span> <span class="n">tasks</span><span class="p">]</span> <span class="k">except</span> <span class="ne">TimeoutError</span><span class="p">:</span> <span class="k">return</span> <span class="p">[]</span> <span class="c1"># Partial results discarded; all tasks cancelled</span> </code></pre></div> <p><code>TaskGroup</code> raises an <code>ExceptionGroup</code> if any task fails, which cancels the rest — this is structured concurrency. Use <code>asyncio.gather(return_exceptions=True)</code> when you want partial results even on failure. <code>asyncio.timeout()</code> (3.11+) replaces <code>asyncio.wait_for()</code> for nested timeout control.</p>