Python FastAPI dependency injection for database transactions

<p>I need my FastAPI routes to participate in database transactions where multiple operations must succeed or fail together. I want a dependency that manages the transaction lifecycle.</p>

<p>Transaction-scoped dependency:</p> <div class="highlight"><pre><span></span><code><span class="kn">from</span><span class="w"> </span><span class="nn">contextlib</span><span class="w"> </span><span class="kn">import</span> <span class="n">asynccontextmanager</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">Annotated</span> <span class="kn">from</span><span class="w"> </span><span class="nn">fastapi</span><span class="w"> </span><span class="kn">import</span> <span class="n">Depends</span> <span class="kn">from</span><span class="w"> </span><span class="nn">sqlalchemy.ext.asyncio</span><span class="w"> </span><span class="kn">import</span> <span class="n">AsyncSession</span> <span class="kn">from</span><span class="w"> </span><span class="nn">app.database</span><span class="w"> </span><span class="kn">import</span> <span class="n">async_session_factory</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">get_db</span><span class="p">():</span> <span class="w"> </span><span class="sd">"""Provides a session with auto-commit on success, rollback on error."""</span> <span class="k">async</span> <span class="k">with</span> <span class="n">async_session_factory</span><span class="p">()</span> <span class="k">as</span> <span class="n">session</span><span class="p">:</span> <span class="k">async</span> <span class="k">with</span> <span class="n">session</span><span class="o">.</span><span class="n">begin</span><span class="p">():</span> <span class="k">yield</span> <span class="n">session</span> <span class="c1"># Commits on success, rolls back on exception</span> <span class="n">DbSession</span> <span class="o">=</span> <span class="n">Annotated</span><span class="p">[</span><span class="n">AsyncSession</span><span class="p">,</span> <span class="n">Depends</span><span class="p">(</span><span class="n">get_db</span><span class="p">)]</span> <span class="c1"># For manual transaction control:</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">get_db_manual</span><span class="p">():</span> <span class="w"> </span><span class="sd">"""Provides a session without auto-commit -- caller controls transaction."""</span> <span class="k">async</span> <span class="k">with</span> <span class="n">async_session_factory</span><span class="p">()</span> <span class="k">as</span> <span class="n">session</span><span class="p">:</span> <span class="k">yield</span> <span class="n">session</span> <span class="c1"># Route that uses auto-commit:</span> <span class="nd">@router</span><span class="o">.</span><span class="n">post</span><span class="p">(</span><span class="s1">'/votes'</span><span class="p">)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">cast_vote</span><span class="p">(</span><span class="n">vote</span><span class="p">:</span> <span class="n">VoteCreate</span><span class="p">,</span> <span class="n">db</span><span class="p">:</span> <span class="n">DbSession</span><span class="p">):</span> <span class="n">new_vote</span> <span class="o">=</span> <span class="n">Vote</span><span class="p">(</span><span class="o">**</span><span class="n">vote</span><span class="o">.</span><span class="n">model_dump</span><span class="p">())</span> <span class="n">db</span><span class="o">.</span><span class="n">add</span><span class="p">(</span><span class="n">new_vote</span><span class="p">)</span> <span class="c1"># No commit needed -- session.begin() handles it</span> <span class="k">return</span> <span class="n">new_vote</span> <span class="c1"># Route with multiple dependent operations:</span> <span class="nd">@router</span><span class="o">.</span><span class="n">post</span><span class="p">(</span><span class="s1">'/traces/</span><span class="si">{trace_id}</span><span class="s1">/validate'</span><span class="p">)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">validate_trace</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="n">db</span><span class="p">:</span> <span class="n">DbSession</span><span class="p">):</span> <span class="n">trace</span> <span class="o">=</span> <span class="k">await</span> <span class="n">db</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="n">Trace</span><span class="p">,</span> <span class="n">trace_id</span><span class="p">)</span> <span class="k">if</span> <span class="ow">not</span> <span class="n">trace</span><span class="p">:</span> <span class="k">raise</span> <span class="n">HTTPException</span><span class="p">(</span><span class="mi">404</span><span class="p">)</span> <span class="n">trace</span><span class="o">.</span><span class="n">status</span> <span class="o">=</span> <span class="s1">'validated'</span> <span class="n">trace</span><span class="o">.</span><span class="n">trust_score</span> <span class="o">=</span> <span class="mf">1.0</span> <span class="c1"># Update contributor stats in same transaction:</span> <span class="k">await</span> <span class="n">update_contributor_stats</span><span class="p">(</span><span class="n">db</span><span class="p">,</span> <span class="n">trace</span><span class="o">.</span><span class="n">contributor_id</span><span class="p">)</span> <span class="c1"># Both updates commit together or both roll back</span> </code></pre></div> <p>Key points: - session.begin() as context manager auto-commits on exit, rolls back on exception - All operations in one route handler share one transaction (one session) - Raise exceptions to trigger rollback -- FastAPI exception handlers still fire - For read-only routes, no transaction needed -- but session.begin() is low overhead - expire_on_commit=False prevents attribute access after commit from triggering lazy loads</p>