pytest async fixtures with database transaction rollback

<p>I need fast async pytest tests for my FastAPI application using a real database. I want tests to roll back after each test rather than truncating tables, and override the FastAPI database dependency with the test session.</p>

<p>Transaction rollback pattern for fast isolated tests:</p> <div class="highlight"><pre><span></span><code><span class="c1"># conftest.py</span> <span class="kn">import</span><span class="w"> </span><span class="nn">pytest</span> <span class="kn">import</span><span class="w"> </span><span class="nn">pytest_asyncio</span> <span class="kn">from</span><span class="w"> </span><span class="nn">httpx</span><span class="w"> </span><span class="kn">import</span> <span class="n">AsyncClient</span><span class="p">,</span> <span class="n">ASGITransport</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">create_async_engine</span><span class="p">,</span> <span class="n">async_sessionmaker</span> <span class="kn">from</span><span class="w"> </span><span class="nn">app.main</span><span class="w"> </span><span class="kn">import</span> <span class="n">app</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">get_db</span> <span class="kn">from</span><span class="w"> </span><span class="nn">app.models.base</span><span class="w"> </span><span class="kn">import</span> <span class="n">Base</span> <span class="n">TEST_DB</span> <span class="o">=</span> <span class="s1">'postgresql+asyncpg://test:test@localhost:5432/test_db'</span> <span class="nd">@pytest_asyncio</span><span class="o">.</span><span class="n">fixture</span><span class="p">(</span><span class="n">scope</span><span class="o">=</span><span class="s1">'session'</span><span class="p">)</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">engine</span><span class="p">():</span> <span class="n">engine</span> <span class="o">=</span> <span class="n">create_async_engine</span><span class="p">(</span><span class="n">TEST_DB</span><span class="p">)</span> <span class="k">async</span> <span class="k">with</span> <span class="n">engine</span><span class="o">.</span><span class="n">begin</span><span class="p">()</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span> <span class="k">await</span> <span class="n">conn</span><span class="o">.</span><span class="n">run_sync</span><span class="p">(</span><span class="n">Base</span><span class="o">.</span><span class="n">metadata</span><span class="o">.</span><span class="n">create_all</span><span class="p">)</span> <span class="k">yield</span> <span class="n">engine</span> <span class="k">async</span> <span class="k">with</span> <span class="n">engine</span><span class="o">.</span><span class="n">begin</span><span class="p">()</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span> <span class="k">await</span> <span class="n">conn</span><span class="o">.</span><span class="n">run_sync</span><span class="p">(</span><span class="n">Base</span><span class="o">.</span><span class="n">metadata</span><span class="o">.</span><span class="n">drop_all</span><span class="p">)</span> <span class="k">await</span> <span class="n">engine</span><span class="o">.</span><span class="n">dispose</span><span class="p">()</span> <span class="nd">@pytest_asyncio</span><span class="o">.</span><span class="n">fixture</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">session</span><span class="p">(</span><span class="n">engine</span><span class="p">):</span> <span class="k">async</span> <span class="k">with</span> <span class="n">engine</span><span class="o">.</span><span class="n">begin</span><span class="p">()</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span> <span class="k">async</span> <span class="k">with</span> <span class="n">async_sessionmaker</span><span class="p">(</span><span class="n">conn</span><span class="p">,</span> <span class="n">expire_on_commit</span><span class="o">=</span><span class="kc">False</span><span class="p">)()</span> <span class="k">as</span> <span class="n">sess</span><span class="p">:</span> <span class="k">yield</span> <span class="n">sess</span> <span class="k">await</span> <span class="n">sess</span><span class="o">.</span><span class="n">rollback</span><span class="p">()</span> <span class="c1"># Rollback after each test</span> <span class="nd">@pytest_asyncio</span><span class="o">.</span><span class="n">fixture</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">client</span><span class="p">(</span><span class="n">session</span><span class="p">):</span> <span class="n">app</span><span class="o">.</span><span class="n">dependency_overrides</span><span class="p">[</span><span class="n">get_db</span><span class="p">]</span> <span class="o">=</span> <span class="k">lambda</span><span class="p">:</span> <span class="n">session</span> <span class="k">async</span> <span class="k">with</span> <span class="n">AsyncClient</span><span class="p">(</span><span class="n">transport</span><span class="o">=</span><span class="n">ASGITransport</span><span class="p">(</span><span class="n">app</span><span class="o">=</span><span class="n">app</span><span class="p">),</span> <span class="n">base_url</span><span class="o">=</span><span class="s1">'http://test'</span><span class="p">)</span> <span class="k">as</span> <span class="n">c</span><span class="p">:</span> <span class="k">yield</span> <span class="n">c</span> <span class="n">app</span><span class="o">.</span><span class="n">dependency_overrides</span><span class="o">.</span><span class="n">clear</span><span class="p">()</span> </code></pre></div> <div class="highlight"><pre><span></span><code><span class="c1"># pytest.ini</span> <span class="k">[pytest]</span> <span class="na">asyncio_mode</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="s">auto</span> </code></pre></div> <p>Key points: - Transaction rollback is 10-50x faster than DROP/CREATE or TRUNCATE between tests - dependency_overrides swaps the real DB session for the test session - scope='session' on engine shares connection pool across all tests - Use pytest-asyncio with asyncio_mode = auto to avoid decorating every test</p>