SQLAlchemy many-to-many relationships with join table

<p>I have a many-to-many relationship between traces and tags. I need to create the join table correctly in SQLAlchemy, handle tag insertion with get-or-create semantics, and query traces by tag efficiently.</p>

<p>Many-to-many with secondary join table:</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">Table</span><span class="p">,</span> <span class="n">Column</span><span class="p">,</span> <span class="n">ForeignKey</span><span class="p">,</span> <span class="n">select</span> <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">UUID</span> <span class="kn">from</span><span class="w"> </span><span class="nn">sqlalchemy.orm</span><span class="w"> </span><span class="kn">import</span> <span class="n">Mapped</span><span class="p">,</span> <span class="n">mapped_column</span><span class="p">,</span> <span class="n">relationship</span> <span class="c1"># Join table definition:</span> <span class="n">trace_tags</span> <span class="o">=</span> <span class="n">Table</span><span class="p">(</span> <span class="s1">'trace_tags'</span><span class="p">,</span> <span class="n">Base</span><span class="o">.</span><span class="n">metadata</span><span class="p">,</span> <span class="n">Column</span><span class="p">(</span><span class="s1">'trace_id'</span><span class="p">,</span> <span class="n">UUID</span><span class="p">(</span><span class="n">as_uuid</span><span class="o">=</span><span class="kc">True</span><span class="p">),</span> <span class="n">ForeignKey</span><span class="p">(</span><span class="s1">'traces.id'</span><span class="p">),</span> <span class="n">primary_key</span><span class="o">=</span><span class="kc">True</span><span class="p">),</span> <span class="n">Column</span><span class="p">(</span><span class="s1">'tag_id'</span><span class="p">,</span> <span class="n">UUID</span><span class="p">(</span><span class="n">as_uuid</span><span class="o">=</span><span class="kc">True</span><span class="p">),</span> <span class="n">ForeignKey</span><span class="p">(</span><span class="s1">'tags.id'</span><span class="p">),</span> <span class="n">primary_key</span><span class="o">=</span><span class="kc">True</span><span class="p">),</span> <span class="p">)</span> <span class="k">class</span><span class="w"> </span><span class="nc">Trace</span><span class="p">(</span><span class="n">Base</span><span class="p">):</span> <span class="n">tags</span><span class="p">:</span> <span class="n">Mapped</span><span class="p">[</span><span class="nb">list</span><span class="p">[</span><span class="s1">'Tag'</span><span class="p">]]</span> <span class="o">=</span> <span class="n">relationship</span><span class="p">(</span> <span class="s1">'Tag'</span><span class="p">,</span> <span class="n">secondary</span><span class="o">=</span><span class="s1">'trace_tags'</span><span class="p">,</span> <span class="n">back_populates</span><span class="o">=</span><span class="s1">'traces'</span> <span class="p">)</span> <span class="k">class</span><span class="w"> </span><span class="nc">Tag</span><span class="p">(</span><span class="n">Base</span><span class="p">):</span> <span class="n">traces</span><span class="p">:</span> <span class="n">Mapped</span><span class="p">[</span><span class="nb">list</span><span class="p">[</span><span class="s1">'Trace'</span><span class="p">]]</span> <span class="o">=</span> <span class="n">relationship</span><span class="p">(</span> <span class="s1">'Trace'</span><span class="p">,</span> <span class="n">secondary</span><span class="o">=</span><span class="s1">'trace_tags'</span><span class="p">,</span> <span class="n">back_populates</span><span class="o">=</span><span class="s1">'tags'</span> <span class="p">)</span> <span class="c1"># Get-or-create tag (idempotent):</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">get_or_create_tag</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">name</span><span class="p">:</span> <span class="nb">str</span><span class="p">)</span> <span class="o">-&gt;</span> <span class="n">Tag</span><span class="p">:</span> <span class="n">normalized</span> <span class="o">=</span> <span class="n">normalize_tag</span><span class="p">(</span><span class="n">name</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">select</span><span class="p">(</span><span class="n">Tag</span><span class="p">)</span><span class="o">.</span><span class="n">where</span><span class="p">(</span><span class="n">Tag</span><span class="o">.</span><span class="n">name</span> <span class="o">==</span> <span class="n">normalized</span><span class="p">))</span> <span class="n">tag</span> <span class="o">=</span> <span class="n">result</span><span class="o">.</span><span class="n">scalar_one_or_none</span><span class="p">()</span> <span class="k">if</span> <span class="n">tag</span> <span class="ow">is</span> <span class="kc">None</span><span class="p">:</span> <span class="n">tag</span> <span class="o">=</span> <span class="n">Tag</span><span class="p">(</span><span class="n">name</span><span class="o">=</span><span class="n">normalized</span><span class="p">)</span> <span class="n">session</span><span class="o">.</span><span class="n">add</span><span class="p">(</span><span class="n">tag</span><span class="p">)</span> <span class="k">await</span> <span class="n">session</span><span class="o">.</span><span class="n">flush</span><span class="p">()</span> <span class="c1"># Get ID without committing</span> <span class="k">return</span> <span class="n">tag</span> <span class="c1"># Insert into join table directly (avoid lazy load issues):</span> <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="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">add_trace_tags</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">trace</span><span class="p">:</span> <span class="n">Trace</span><span class="p">,</span> <span class="n">tag_names</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="kc">None</span><span class="p">:</span> <span class="k">for</span> <span class="n">name</span> <span class="ow">in</span> <span class="n">tag_names</span><span class="p">:</span> <span class="k">if</span> <span class="ow">not</span> <span class="n">validate_tag</span><span class="p">(</span><span class="n">normalize_tag</span><span class="p">(</span><span class="n">name</span><span class="p">)):</span> <span class="k">continue</span> <span class="n">tag</span> <span class="o">=</span> <span class="k">await</span> <span class="n">get_or_create_tag</span><span class="p">(</span><span class="n">session</span><span class="p">,</span> <span class="n">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">trace_tags</span><span class="o">.</span><span class="n">insert</span><span class="p">()</span><span class="o">.</span><span class="n">values</span><span class="p">(</span><span class="n">trace_id</span><span class="o">=</span><span class="n">trace</span><span class="o">.</span><span class="n">id</span><span class="p">,</span> <span class="n">tag_id</span><span class="o">=</span><span class="n">tag</span><span class="o">.</span><span class="n">id</span><span class="p">)</span> <span class="p">)</span> <span class="c1"># Query traces by tag:</span> <span class="n">stmt</span> <span class="o">=</span> <span class="n">select</span><span class="p">(</span><span class="n">Trace</span><span class="p">)</span><span class="o">.</span><span class="n">where</span><span class="p">(</span><span class="n">Trace</span><span class="o">.</span><span class="n">tags</span><span class="o">.</span><span class="n">any</span><span class="p">(</span><span class="n">Tag</span><span class="o">.</span><span class="n">name</span> <span class="o">==</span> <span class="s1">'python'</span><span class="p">))</span> </code></pre></div> <p>Key points: - Direct insert into join table avoids lazy-load issues in async context - flush() after adding tag gets ID without committing the transaction - relationship secondary= links via the join table automatically for ORM queries - back_populates= enables bidirectional navigation - For async: always use selectinload or joinedload, not lazy loading</p>