SQLAlchemy 2.0 async bulk insert with returning

<p>I need to bulk insert thousands of rows into PostgreSQL using SQLAlchemy 2.0 async. I want to do it efficiently with a single query (not one INSERT per row) and get back the generated IDs without a second SELECT.</p>

<p>Use <code>insert().values()</code> with <code>returning()</code> for efficient bulk inserts:</p> <div class="highlight"><pre><span></span><code><span class="kn">from</span><span class="w"> </span><span class="nn">sqlalchemy</span><span class="w"> </span><span class="kn">import</span> <span class="n">insert</span><span class="p">,</span> <span class="n">select</span> <span class="kn">from</span><span class="w"> </span><span class="nn">app.models.trace</span><span class="w"> </span><span class="kn">import</span> <span class="n">Trace</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">bulk_insert_traces</span><span class="p">(</span><span class="n">session</span><span class="p">:</span> <span class="n">AsyncSession</span><span class="p">,</span> <span class="n">traces</span><span class="p">:</span> <span class="nb">list</span><span class="p">[</span><span class="nb">dict</span><span class="p">])</span> <span class="o">-&gt;</span> <span class="nb">list</span><span class="p">[</span><span class="n">uuid</span><span class="o">.</span><span class="n">UUID</span><span class="p">]:</span> <span class="k">if</span> <span class="ow">not</span> <span class="n">traces</span><span class="p">:</span> <span class="k">return</span> <span class="p">[]</span> <span class="c1"># Single bulk INSERT with RETURNING id</span> <span class="n">stmt</span> <span class="o">=</span> <span class="p">(</span> <span class="n">insert</span><span class="p">(</span><span class="n">Trace</span><span class="p">)</span> <span class="o">.</span><span class="n">values</span><span class="p">(</span><span class="n">traces</span><span class="p">)</span> <span class="c1"># list of dicts</span> <span class="o">.</span><span class="n">returning</span><span class="p">(</span><span class="n">Trace</span><span class="o">.</span><span class="n">id</span><span class="p">)</span> <span class="p">)</span> <span class="n">result</span> <span class="o">=</span> <span class="k">await</span> <span class="n">session</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">stmt</span><span class="p">)</span> <span class="n">inserted_ids</span> <span class="o">=</span> <span class="n">result</span><span class="o">.</span><span class="n">scalars</span><span class="p">()</span><span class="o">.</span><span class="n">all</span><span class="p">()</span> <span class="k">await</span> <span class="n">session</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span> <span class="k">return</span> <span class="n">inserted_ids</span> <span class="c1"># Usage:</span> <span class="n">rows</span> <span class="o">=</span> <span class="p">[</span> <span class="p">{</span> <span class="s1">'title'</span><span class="p">:</span> <span class="sa">f</span><span class="s1">'Trace </span><span class="si">{</span><span class="n">i</span><span class="si">}</span><span class="s1">'</span><span class="p">,</span> <span class="s1">'context_text'</span><span class="p">:</span> <span class="s1">'Context...'</span><span class="p">,</span> <span class="s1">'solution_text'</span><span class="p">:</span> <span class="s1">'Solution...'</span><span class="p">,</span> <span class="s1">'contributor_id'</span><span class="p">:</span> <span class="n">user_id</span><span class="p">,</span> <span class="s1">'status'</span><span class="p">:</span> <span class="s1">'validated'</span><span class="p">,</span> <span class="p">}</span> <span class="k">for</span> <span class="n">i</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="mi">1000</span><span class="p">)</span> <span class="p">]</span> <span class="n">ids</span> <span class="o">=</span> <span class="k">await</span> <span class="n">bulk_insert_traces</span><span class="p">(</span><span class="n">session</span><span class="p">,</span> <span class="n">rows</span><span class="p">)</span> </code></pre></div> <p>For upsert (INSERT ... ON CONFLICT DO UPDATE):</p> <div class="highlight"><pre><span></span><code><span class="kn">from</span><span class="w"> </span><span class="nn">sqlalchemy.dialects.postgresql</span><span class="w"> </span><span class="kn">import</span> <span class="n">insert</span> <span class="k">as</span> <span class="n">pg_insert</span> <span class="n">stmt</span> <span class="o">=</span> <span class="n">pg_insert</span><span class="p">(</span><span class="n">Tag</span><span class="p">)</span><span class="o">.</span><span class="n">values</span><span class="p">(</span><span class="n">name</span><span class="o">=</span><span class="s1">'python'</span><span class="p">)</span> <span class="n">stmt</span> <span class="o">=</span> <span class="n">stmt</span><span class="o">.</span><span class="n">on_conflict_do_nothing</span><span class="p">(</span><span class="n">index_elements</span><span class="o">=</span><span class="p">[</span><span class="s1">'name'</span><span class="p">])</span> <span class="k">await</span> <span class="n">session</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">stmt</span><span class="p">)</span> </code></pre></div> <p>Key points: - Single bulk INSERT is 10-100x faster than a loop of individual inserts - <code>RETURNING</code> avoids a second SELECT query to get generated IDs - PostgreSQL-specific <code>pg_insert</code> for upsert (<code>on_conflict_do_update</code>) - Batch by 500-1000 rows to avoid hitting PostgreSQL parameter limits</p>