asyncio semaphore for concurrency limiting

<p>I have a background worker that processes items concurrently, but I don't want to overwhelm the database or external API. I need to limit the number of concurrent coroutines processing at any time.</p>

<p>Use <code>asyncio.Semaphore</code> to cap concurrent operations:</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">Coroutine</span><span class="p">,</span> <span class="n">TypeVar</span> <span class="n">T</span> <span class="o">=</span> <span class="n">TypeVar</span><span class="p">(</span><span class="s1">'T'</span><span class="p">)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">process_with_limit</span><span class="p">(</span> <span class="n">items</span><span class="p">:</span> <span class="nb">list</span><span class="p">,</span> <span class="n">processor</span><span class="p">:</span> <span class="nb">callable</span><span class="p">,</span> <span class="n">max_concurrent</span><span class="p">:</span> <span class="nb">int</span> <span class="o">=</span> <span class="mi">5</span><span class="p">,</span> <span class="p">)</span> <span class="o">-&gt;</span> <span class="nb">list</span><span class="p">:</span> <span class="w"> </span><span class="sd">"""Process items concurrently with a cap on parallelism."""</span> <span class="n">semaphore</span> <span class="o">=</span> <span class="n">asyncio</span><span class="o">.</span><span class="n">Semaphore</span><span class="p">(</span><span class="n">max_concurrent</span><span class="p">)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">bounded_process</span><span class="p">(</span><span class="n">item</span><span class="p">):</span> <span class="k">async</span> <span class="k">with</span> <span class="n">semaphore</span><span class="p">:</span> <span class="k">return</span> <span class="k">await</span> <span class="n">processor</span><span class="p">(</span><span class="n">item</span><span class="p">)</span> <span class="k">return</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">bounded_process</span><span class="p">(</span><span class="n">item</span><span class="p">)</span> <span class="k">for</span> <span class="n">item</span> <span class="ow">in</span> <span class="n">items</span><span class="p">])</span> <span class="c1"># Usage — process 100 traces, max 5 at a time:</span> <span class="n">results</span> <span class="o">=</span> <span class="k">await</span> <span class="n">process_with_limit</span><span class="p">(</span> <span class="n">pending_trace_ids</span><span class="p">,</span> <span class="n">generate_embedding</span><span class="p">,</span> <span class="n">max_concurrent</span><span class="o">=</span><span class="mi">5</span><span class="p">,</span> <span class="p">)</span> <span class="c1"># With error handling:</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">process_one</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">tuple</span><span class="p">[</span><span class="nb">str</span><span class="p">,</span> <span class="nb">bool</span><span class="p">]:</span> <span class="k">async</span> <span class="k">with</span> <span class="n">semaphore</span><span class="p">:</span> <span class="k">try</span><span class="p">:</span> <span class="k">await</span> <span class="n">generate_embedding</span><span class="p">(</span><span class="n">trace_id</span><span class="p">)</span> <span class="k">return</span> <span class="n">trace_id</span><span class="p">,</span> <span class="kc">True</span> <span class="k">except</span> <span class="ne">Exception</span> <span class="k">as</span> <span class="n">e</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">'Failed'</span><span class="p">,</span> <span class="n">trace_id</span><span class="o">=</span><span class="n">trace_id</span><span class="p">,</span> <span class="n">error</span><span class="o">=</span><span class="nb">str</span><span class="p">(</span><span class="n">e</span><span class="p">))</span> <span class="k">return</span> <span class="n">trace_id</span><span class="p">,</span> <span class="kc">False</span> </code></pre></div> <p>Key points: - <code>asyncio.Semaphore(n)</code> allows at most n coroutines in the critical section - Create the semaphore outside the coroutine so it's shared across all tasks - <code>asyncio.gather()</code> starts all tasks immediately — semaphore controls entry - For producer/consumer patterns, prefer <code>asyncio.Queue</code> over semaphore</p>