Python structured concurrency with anyio

<p>Building code that needs to work with both asyncio and Trio event loops (or library code that shouldn't dictate the event loop). Also need nursery-style task cancellation that TaskGroup provides but with trio compatibility.</p>

<p>Use <code>anyio</code> as an abstraction layer over asyncio and trio:</p> <div class="highlight"><pre><span></span><code><span class="kn">import</span><span class="w"> </span><span class="nn">anyio</span> <span class="kn">from</span><span class="w"> </span><span class="nn">anyio</span><span class="w"> </span><span class="kn">import</span> <span class="n">create_task_group</span><span class="p">,</span> <span class="n">move_on_after</span><span class="p">,</span> <span class="n">fail_after</span> <span class="kn">from</span><span class="w"> </span><span class="nn">anyio.abc</span><span class="w"> </span><span class="kn">import</span> <span class="n">TaskGroup</span> <span class="c1"># Concurrent tasks with automatic cancellation</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">fetch_trace_data</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">create_task_group</span><span class="p">()</span> <span class="k">as</span> <span class="n">tg</span><span class="p">:</span> <span class="n">results</span> <span class="o">=</span> <span class="p">{}</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">fetch_trace</span><span class="p">():</span> <span class="n">results</span><span class="p">[</span><span class="s1">'trace'</span><span class="p">]</span> <span class="o">=</span> <span class="k">await</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="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">fetch_votes</span><span class="p">():</span> <span class="n">results</span><span class="p">[</span><span class="s1">'votes'</span><span class="p">]</span> <span class="o">=</span> <span class="k">await</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="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">fetch_tags</span><span class="p">():</span> <span class="n">results</span><span class="p">[</span><span class="s1">'tags'</span><span class="p">]</span> <span class="o">=</span> <span class="k">await</span> <span class="n">db</span><span class="o">.</span><span class="n">get_tags</span><span class="p">(</span><span class="n">trace_id</span><span class="p">)</span> <span class="n">tg</span><span class="o">.</span><span class="n">start_soon</span><span class="p">(</span><span class="n">fetch_trace</span><span class="p">)</span> <span class="n">tg</span><span class="o">.</span><span class="n">start_soon</span><span class="p">(</span><span class="n">fetch_votes</span><span class="p">)</span> <span class="n">tg</span><span class="o">.</span><span class="n">start_soon</span><span class="p">(</span><span class="n">fetch_tags</span><span class="p">)</span> <span class="k">return</span> <span class="n">results</span> <span class="c1"># Timeout with move_on_after (gives up, returns None-equivalent)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">try_embed_with_timeout</span><span class="p">(</span><span class="n">text</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">float</span><span class="p">]</span> <span class="o">|</span> <span class="kc">None</span><span class="p">:</span> <span class="n">result</span> <span class="o">=</span> <span class="kc">None</span> <span class="k">with</span> <span class="n">move_on_after</span><span class="p">(</span><span class="mf">5.0</span><span class="p">):</span> <span class="c1"># Gives up after 5 seconds</span> <span class="n">result</span> <span class="o">=</span> <span class="k">await</span> <span class="n">embed_text</span><span class="p">(</span><span class="n">text</span><span class="p">)</span> <span class="k">return</span> <span class="n">result</span> <span class="c1"># Timeout that raises on expiry</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">embed_or_fail</span><span class="p">(</span><span class="n">text</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">float</span><span class="p">]:</span> <span class="k">with</span> <span class="n">fail_after</span><span class="p">(</span><span class="mf">10.0</span><span class="p">):</span> <span class="k">return</span> <span class="k">await</span> <span class="n">embed_text</span><span class="p">(</span><span class="n">text</span><span class="p">)</span> <span class="c1"># Run from sync code</span> <span class="k">if</span> <span class="vm">__name__</span> <span class="o">==</span> <span class="s1">'__main__'</span><span class="p">:</span> <span class="c1"># Run with asyncio (default)</span> <span class="n">anyio</span><span class="o">.</span><span class="n">run</span><span class="p">(</span><span class="n">main</span><span class="p">)</span> <span class="c1"># Run with trio</span> <span class="n">anyio</span><span class="o">.</span><span class="n">run</span><span class="p">(</span><span class="n">main</span><span class="p">,</span> <span class="n">backend</span><span class="o">=</span><span class="s1">'trio'</span><span class="p">)</span> <span class="c1"># Library code that works with both</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">anyio_compatible_function</span><span class="p">()</span> <span class="o">-&gt;</span> <span class="kc">None</span><span class="p">:</span> <span class="c1"># Uses anyio primitives, not asyncio directly</span> <span class="k">await</span> <span class="n">anyio</span><span class="o">.</span><span class="n">sleep</span><span class="p">(</span><span class="mi">1</span><span class="p">)</span> <span class="k">async</span> <span class="k">with</span> <span class="n">anyio</span><span class="o">.</span><span class="n">open_file</span><span class="p">(</span><span class="s1">'data.txt'</span><span class="p">)</span> <span class="k">as</span> <span class="n">f</span><span class="p">:</span> <span class="n">content</span> <span class="o">=</span> <span class="k">await</span> <span class="n">f</span><span class="o">.</span><span class="n">read</span><span class="p">()</span> </code></pre></div> <p><code>anyio</code> is used by Starlette/FastAPI internally. <code>create_task_group()</code> behaves like Python 3.11's <code>asyncio.TaskGroup</code>. <code>move_on_after</code> is cleaner than try/except TimeoutError for optional operations.</p>