PostgreSQL row-level security (RLS) with SQLAlchemy

<p>Need to enforce data isolation so each tenant can only see their own rows. Application-level filtering is error-prone — a missing WHERE clause leaks all data. Want database-enforced isolation.</p>

<p>Enable RLS on the table and create policies. Pass the current tenant via <code>SET LOCAL</code>:</p> <div class="highlight"><pre><span></span><code><span class="c1">-- Enable RLS</span> <span class="k">ALTER</span><span class="w"> </span><span class="k">TABLE</span><span class="w"> </span><span class="n">traces</span><span class="w"> </span><span class="n">ENABLE</span><span class="w"> </span><span class="k">ROW</span><span class="w"> </span><span class="k">LEVEL</span><span class="w"> </span><span class="k">SECURITY</span><span class="p">;</span> <span class="k">ALTER</span><span class="w"> </span><span class="k">TABLE</span><span class="w"> </span><span class="n">traces</span><span class="w"> </span><span class="k">FORCE</span><span class="w"> </span><span class="k">ROW</span><span class="w"> </span><span class="k">LEVEL</span><span class="w"> </span><span class="k">SECURITY</span><span class="p">;</span> <span class="c1">-- Policy: user sees only their own traces</span> <span class="k">CREATE</span><span class="w"> </span><span class="n">POLICY</span><span class="w"> </span><span class="n">traces_tenant_isolation</span><span class="w"> </span><span class="k">ON</span><span class="w"> </span><span class="n">traces</span> <span class="w"> </span><span class="k">USING</span><span class="w"> </span><span class="p">(</span><span class="n">contributor_id</span><span class="w"> </span><span class="o">=</span><span class="w"> </span><span class="n">current_setting</span><span class="p">(</span><span class="s1">'app.current_user_id'</span><span class="p">,</span><span class="w"> </span><span class="k">true</span><span class="p">)::</span><span class="n">uuid</span><span class="p">);</span> <span class="c1">-- Create a limited role for the app (bypasses RLS by default for superuser)</span> <span class="k">CREATE</span><span class="w"> </span><span class="k">ROLE</span><span class="w"> </span><span class="n">app_user</span><span class="p">;</span> <span class="k">GRANT</span><span class="w"> </span><span class="k">SELECT</span><span class="p">,</span><span class="w"> </span><span class="k">INSERT</span><span class="p">,</span><span class="w"> </span><span class="k">UPDATE</span><span class="p">,</span><span class="w"> </span><span class="k">DELETE</span><span class="w"> </span><span class="k">ON</span><span class="w"> </span><span class="n">traces</span><span class="w"> </span><span class="k">TO</span><span class="w"> </span><span class="n">app_user</span><span class="p">;</span> </code></pre></div> <p>In SQLAlchemy, set the session variable per request:</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">event</span><span class="p">,</span> <span class="n">text</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="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">set_tenant_context</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">user_id</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">await</span> <span class="n">session</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">text</span><span class="p">(</span><span class="sa">f</span><span class="s2">"SET LOCAL app.current_user_id = '</span><span class="si">{</span><span class="n">user_id</span><span class="si">}</span><span class="s2">'"</span><span class="p">))</span> <span class="c1"># In FastAPI dependency</span> <span class="k">async</span> <span class="k">def</span><span class="w"> </span><span class="nf">get_session_with_tenant</span><span class="p">(</span> <span class="n">current_user</span><span class="p">:</span> <span class="n">User</span> <span class="o">=</span> <span class="n">Depends</span><span class="p">(</span><span class="n">get_current_user</span><span class="p">),</span> <span class="n">session</span><span class="p">:</span> <span class="n">AsyncSession</span> <span class="o">=</span> <span class="n">Depends</span><span class="p">(</span><span class="n">get_db</span><span class="p">),</span> <span class="p">)</span> <span class="o">-&gt;</span> <span class="n">AsyncGenerator</span><span class="p">[</span><span class="n">AsyncSession</span><span class="p">,</span> <span class="kc">None</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">await</span> <span class="n">set_tenant_context</span><span class="p">(</span><span class="n">session</span><span class="p">,</span> <span class="nb">str</span><span class="p">(</span><span class="n">current_user</span><span class="o">.</span><span class="n">id</span><span class="p">))</span> <span class="k">yield</span> <span class="n">session</span> </code></pre></div> <p>Key consideration: <code>SET LOCAL</code> applies only to the current transaction, which is exactly what you want with connection pooling. Bypass RLS for admin operations with <code>SET row_security = off</code> (requires superuser or <code>BYPASSRLS</code> attribute).</p>