RLS with FastAPI and SQLAlchemy async

Capsule overview

You already have RLS's fundamentals in pure SQL: how it gets turned on, how a policy gets written, how PostgreSQL applies the filter automatically. This capsule teaches you to integrate it into a real production app with FastAPI 0.110+ and SQLAlchemy 2.0 async + asyncpg. You're going to write the dependency that runs SET LOCAL app.tenant_id at the start of each request, you're going to avoid asyncpg's critical gotcha that breaks RLS intermittently (cached prepared statements), you're going to configure PgBouncer correctly, and you're going to write automated isolation tests that pass in CI.

⚠️ The critical caveat repeated (yes, again): RLS here is used only for multi-tenancy, not for auth/RBAC. Every time you see the SET LOCAL app.tenant_id code, remember that the "tenant" is the unit that's stable throughout the whole request — the user's organization, not the individual user. If you find yourself writing SET LOCAL app.user_id and adding policies that depend on the user for granular permissions, stop and reconsider: you probably need RBAC in the application layer (FastAPI dependencies that check permissions), not more policies. Auth/RBAC is covered in guide #9 of the path.

By the end you'll have a working FastAPI mini-API with two tenants and production RLS: automated "malicious" tests will pass, asyncpg will be configured so it doesn't break the guarantee, and you'll have the exact code that module 8 (TaskFlow) is going to scale into the complete API.

Mental model: the dependency as the "gateway to the tenant"

In FastAPI, dependencies are code that runs before the endpoint. The typical auth dependency returns the current user; you're going to have a similar one that returns the current tenant. But the RLS dependency does something more: it modifies the DB session's state so the policies know which tenant the request belongs to.

Think of it like a turnstile at a building's entrance. Before the dev accesses the DB (the building), they have to go through the turnstile (the dependency) that presents their tenant credential. After going through, everything they do inside is marked with their tenant. Without going through the turnstile, the queries fail or return 0 rows.

The key technical detail is: the turnstile has to run INSIDE the same transaction as the endpoint's queries. SET LOCAL only lasts until the COMMIT or ROLLBACK. If the dependency sets the context in one transaction and the endpoint opens another transaction, the context is lost and the queries fail.

This matters especially with connection pools (asyncpg, PgBouncer) where "the same session" can mean "the same connection during this specific transaction" — not "the same connection during the whole request." The capsule solves this.

Base setup: the app's role

Before touching Python, you need the app_user role correctly configured in PostgreSQL (you saw it in capsule 04). To recap, the minimal setup:

-- Run as superuser/postgres
CREATE ROLE app_user LOGIN PASSWORD 'app_password_secure';
GRANT CONNECT ON DATABASE rls_demo TO app_user;
GRANT USAGE ON SCHEMA public TO app_user;
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO app_user;
GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO app_user;

-- Make sure new tables also grant permissions to the role
ALTER DEFAULT PRIVILEGES IN SCHEMA public
    GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO app_user;

The app connects as app_user, NOT as postgres. This makes sure RLS always applies (because app_user isn't the tables' owner, it doesn't need the FORCE... but it's still worth having FORCE as defense in depth).

The asyncpg gotcha: cached prepared statements

Before showing the dependency, we have to cover RLS + asyncpg's subtlest problem.

What it is: asyncpg, by default, caches prepared statements. When you run SELECT * FROM tasks WHERE id = $1 for the first time, asyncpg internally registers it as prepared statement P1 on the connection and reuses the plan in future queries. This improves performance.

Why it breaks RLS: prepared statements get planned ONCE and reused. PostgreSQL plans the query including the policy's predicate. If at planning time current_setting('app.tenant_id') returns NULL (or a different value), the plan gets "fixed" with that value. Subsequent executions can return incorrect results.

Worse still with PgBouncer in transaction mode: each transaction can take a different connection from the pool. If connection A has prepared statement P1 planned for tenant 1, and tenant 2's next request takes that same connection A, the cached plan can get applied to tenant 2's context. Result: an intermittent leak.

The solution: disable asyncpg's prepared statement cache when you use RLS.

# app/db/session.py
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.ext.asyncio import AsyncSession

DATABASE_URL = "postgresql+asyncpg://app_user:app_password_secure@localhost:5432/rls_demo"

engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    # CRITICAL for RLS: disable the prepared statements cache
    connect_args={
        "statement_cache_size": 0,
        "prepared_statement_cache_size": 0,
    },
)

SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

The trade-off: disabling the cache has a performance cost (~5-10% more latency per query). For a multi-tenant SaaS API, that cost is acceptable compared with the risk of intermittent leaks. If your performance is critical and you want to keep the cache, there are more complex alternatives (a dedicated connection per tenant, manual cache invalidation) that are outside this capsule's scope.

Confirmation you're on the right version: asyncpg 0.27+ and SQLAlchemy 2.0+. The statement_cache_size=0 parameter works in both modern versions.

The dependency that sets the tenant context

Now the integration's key code. The dependency does 4 things:

1. Gets SQLAlchemy's async session.
2. Gets the request's tenant (typically from the JWT or a header).
3. Starts an explicit transaction.
4. Runs SET LOCAL app.tenant_id = ... inside that transaction.
5. Yields the session to the endpoint, which operates inside the same transaction.
6. When it finishes, it commits (or rolls back on error).

# app/db/tenant_context.py
from typing import AsyncIterator
from fastapi import Depends, HTTPException, Request
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import SessionLocal


async def get_current_tenant_id(request: Request) -> int:
    """
    Extracts the tenant_id from the request (typically from a decoded JWT or a header).
    In production this comes from your auth system.
    """
    # A simplified version for this capsule: the X-Tenant-ID header
    tenant_id_str = request.headers.get("X-Tenant-ID")
    if not tenant_id_str:
        raise HTTPException(status_code=401, detail="Missing X-Tenant-ID header")
    try:
        return int(tenant_id_str)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid X-Tenant-ID header")


async def get_tenant_session(
    tenant_id: int = Depends(get_current_tenant_id),
) -> AsyncIterator[AsyncSession]:
    """
    A SQLAlchemy session with the tenant context set.
    Critical: SET LOCAL has to run in the same transaction as the queries.
    """
    async with SessionLocal() as session:
        async with session.begin():
            # SET LOCAL only lasts until this transaction's COMMIT/ROLLBACK.
            # That's why we use an explicit session.begin().
            await session.execute(
                text("SET LOCAL app.tenant_id = :tid"),
                {"tid": str(tenant_id)},
            )
            yield session
            # session.begin() commits automatically on exit if there was no exception.
            # If there was an exception, it rolls back. SET LOCAL disappears in both cases.

Three details that kill you if you skip them:

1. An explicit session.begin(), not implicit. SQLAlchemy 2.0 async can operate in "autocommit" mode or with explicit transactions. For RLS, the explicit transaction is mandatory — SET LOCAL needs a transaction that groups the setting with the queries.

2. The yield is INSIDE the session.begin(), not outside. If the yield were outside, the endpoint would operate in a new transaction where the SET LOCAL no longer applies.

3. Parameterize the setting's value, don't concatenate it. SET LOCAL app.tenant_id = '1' with string concatenation (f"SET LOCAL app.tenant_id = '{tenant_id}'") is vulnerable to SQL injection if tenant_id comes from unvalidated input. The correct form is text("SET LOCAL app.tenant_id = :tid") with a bindparam. But careful: asyncpg doesn't allow parameterizing setting names, only values, and SET LOCAL with parameters has caveats. The most robust form is to validate tenant_id as an int beforehand (which the get_current_tenant_id dependency does with int(tenant_id_str)) and then concatenate safely.

# The 100% safe version: validate the type and then concatenate (knowing it's an int)
async def get_tenant_session(
    tenant_id: int = Depends(get_current_tenant_id),  # Already validated as int
) -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        async with session.begin():
            # tenant_id is guaranteed to be an int by the previous dependency.
            # Safe concatenation because the type is validated.
            await session.execute(text(f"SET LOCAL app.tenant_id = {tenant_id}"))
            yield session

Both versions are valid. The first is more defensive (standard SQLAlchemy parameterization); the second is more readable. Pick according to your team's style, but understand the why.

The endpoint using the dependency

With the dependency ready, the endpoint stays clean:

# app/api/tasks.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.models import Task
from app.db.tenant_context import get_tenant_session

router = APIRouter()


@router.get("/tasks")
async def list_tasks(db: AsyncSession = Depends(get_tenant_session)):
    # We do NOT need .where(Task.tenant_id == ...).
    # RLS adds it automatically.
    result = await db.execute(
        select(Task).order_by(Task.created_at.desc()).limit(50)
    )
    return list(result.scalars().all())


@router.get("/tasks/{task_id}")
async def get_task(
    task_id: int,
    db: AsyncSession = Depends(get_tenant_session),
):
    result = await db.execute(select(Task).where(Task.id == task_id))
    task = result.scalar_one_or_none()
    if task is None:
        # This can be "doesn't exist" or "exists but belongs to another tenant."
        # To an attacker, both cases are indistinguishable. Good.
        raise HTTPException(status_code=404, detail="Task not found")
    return task


@router.post("/tasks")
async def create_task(
    title: str,
    db: AsyncSession = Depends(get_tenant_session),
):
    # We need to know the tenant_id for the INSERT (it doesn't autofill).
    # We get it from the setting the dependency already set.
    from sqlalchemy import text
    result = await db.execute(text("SELECT current_setting('app.tenant_id')::BIGINT"))
    tenant_id = result.scalar_one()

    task = Task(tenant_id=tenant_id, title=title)
    db.add(task)
    await db.flush()
    await db.refresh(task)
    return task

Things to notice:

• list_tasks doesn't mention tenant_id. RLS filters automatically. This is what closes the ceiling of capsule 03's model: even if a dev forgets the filter, the app stays secure.
• get_task returns a 404 indistinguishable between "doesn't exist" and "exists but belongs to another tenant." RLS makes the other tenant's row invisible (result: scalar_one_or_none() returns None). The endpoint needs no special logic.
• create_task DOES need the tenant_id to insert. It gets it by reading the current setting. Alternatively, you could pass it as a separate dependency (tenant_id: int = Depends(get_current_tenant_id)) and use it directly — more readable.

Defense in depth: keep the WHERE tenant_id too

Even though RLS protects you, don't remove the WHERE tenant_id from the code. Keeping it is defense in depth for three reasons:

1. Implicit documentation. Whoever reads the code immediately understands it's multi-tenant code. Without the WHERE, it looks like single-tenant code that mysteriously works.

2. Protection against infrastructure changes. If in the future someone (a standalone script, a temporary job, a migration) runs queries without going through the dependency that sets the context, RLS can fail in unexpected ways. The explicit WHERE keeps protecting.

3. Similar performance. PostgreSQL is smart enough to detect that the code's WHERE and the policy's predicate are redundant. There's no significant overhead from duplicating it.

# The defense-in-depth version
@router.get("/tasks")
async def list_tasks(
    tenant_id: int = Depends(get_current_tenant_id),
    db: AsyncSession = Depends(get_tenant_session),
):
    result = await db.execute(
        select(Task)
        .where(Task.tenant_id == tenant_id)  # Defense in depth
        .order_by(Task.created_at.desc())
        .limit(50)
    )
    return list(result.scalars().all())

The rule: RLS is the safety net. The code is still "correct by intent." If the code is written well, both protect and nobody notices the redundancy. If the code fails, RLS rescues it.

PgBouncer and connection pooling: the correct configuration

PgBouncer is a connection pooler widely used in production. Its "transaction" mode (the recommended default) makes each transaction take a connection from the pool, use it, and return it when it finishes. That means two successive requests from the same client can take different connections.

For RLS this is critical:

• ✅ SET LOCAL works fine with PgBouncer transaction mode because it only lasts the transaction. When the transaction ends, the setting gets cleared and the connection goes back "clean" to the pool.
• ❌ SET (without LOCAL) does not work with PgBouncer transaction mode. The setting persists on the connection and the next transaction that takes that connection inherits the previous tenant's setting. This is a textbook cross-tenant leak.

An absolute rule: always SET LOCAL, never SET in code that touches RLS.

Recommended PgBouncer configuration for RLS

# pgbouncer.ini
[databases]
rls_demo = host=postgres-primary port=5432 dbname=rls_demo

[pgbouncer]
listen_port = 6432
listen_addr = 0.0.0.0
auth_type = scram-sha-256
auth_file = /etc/pgbouncer/userlist.txt

# Transaction mode: each transaction takes a connection from the pool.
# Compatible with SET LOCAL.
pool_mode = transaction

# Important with asyncpg + RLS:
# also disable prepared statements on PgBouncer's side.
# (asyncpg already disables them on the client with statement_cache_size=0,
# but this line adds defense.)
server_reset_query = DISCARD ALL

# Typical connection limits
max_client_conn = 200
default_pool_size = 20

Configuration on the app's side: the connection URL points at PgBouncer (localhost:6432) instead of PostgreSQL directly (localhost:5432):

DATABASE_URL = "postgresql+asyncpg://app_user:app_password_secure@localhost:6432/rls_demo"

Something that does NOT work: PgBouncer in pool_mode = session with asyncpg + RLS. Session mode keeps the connection assigned to the client for the whole session, which breaks the pool model and eliminates the benefit. Always stay in transaction mode.

Automated isolation tests

This is the most critical part of having production RLS. Without automated tests, there's no continuous guarantee that the isolation works after changes.

Async pytest setup

# tests/conftest.py
import asyncio
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from app.db.models import Base, Tenant, Task

TEST_DATABASE_URL = "postgresql+asyncpg://app_user:app_password_secure@localhost:5432/rls_demo_test"


@pytest_asyncio.fixture(scope="function")
async def engine():
    engine = create_async_engine(
        TEST_DATABASE_URL,
        echo=False,
        connect_args={"statement_cache_size": 0},
    )
    yield engine
    await engine.dispose()


@pytest_asyncio.fixture(scope="function")
async def session_maker(engine):
    return async_sessionmaker(engine, expire_on_commit=False)

An isolation test between tenants

# tests/test_rls_isolation.py
import pytest
from sqlalchemy import select, text

from app.db.models import Tenant, Task


@pytest.mark.asyncio
async def test_tenant_a_cannot_see_tenant_b_data(session_maker):
    """
    Given two tenants with tasks in each one,
    when tenant A's context is set,
    then queries don't return tenant B's tasks.
    """
    # Arrange: create two tenants and tasks (with bypass privileges for the setup)
    async with session_maker() as setup_session:
        async with setup_session.begin():
            # Bypass RLS for the setup (simulates an admin creating tenants)
            await setup_session.execute(text("SET LOCAL app.tenant_id = '0'"))  # placeholder
            # In real setups, this setup is done by an admin/seed with BYPASSRLS

    # A more realistic setup: use a separate admin role or raw psql to create the data
    # For this capsule, we assume the data exists (created in a previous seed).

    # Act 1: tenant A tries to list tasks
    async with session_maker() as session:
        async with session.begin():
            await session.execute(text("SET LOCAL app.tenant_id = '1'"))
            result = await session.execute(select(Task))
            tasks_visible_to_a = list(result.scalars().all())

    # Act 2: tenant B tries to list tasks
    async with session_maker() as session:
        async with session.begin():
            await session.execute(text("SET LOCAL app.tenant_id = '2'"))
            result = await session.execute(select(Task))
            tasks_visible_to_b = list(result.scalars().all())

    # Assert: no task of tenant B shows up in tenant A's view
    a_task_ids = {t.id for t in tasks_visible_to_a}
    b_task_ids = {t.id for t in tasks_visible_to_b}

    assert a_task_ids.isdisjoint(b_task_ids), (
        f"BREACH: shared task IDs between tenants: {a_task_ids & b_task_ids}"
    )

    # Every task visible to A has to have tenant_id = 1
    for t in tasks_visible_to_a:
        assert t.tenant_id == 1, f"Task {t.id} leaked from tenant {t.tenant_id} to A"

    # Every task visible to B has to have tenant_id = 2
    for t in tasks_visible_to_b:
        assert t.tenant_id == 2, f"Task {t.id} leaked from tenant {t.tenant_id} to B"


@pytest.mark.asyncio
async def test_malicious_query_with_explicit_other_tenant_returns_empty(session_maker):
    """
    Given tenant A's context,
    when a query with WHERE tenant_id = B runs,
    then it returns 0 rows (the policy wins over the WHERE).
    """
    async with session_maker() as session:
        async with session.begin():
            await session.execute(text("SET LOCAL app.tenant_id = '1'"))

            # A "malicious" query: try to read tenant 2 from tenant 1's context
            result = await session.execute(
                select(Task).where(Task.tenant_id == 2)
            )
            leaked = list(result.scalars().all())

    assert leaked == [], (
        f"BREACH: malicious query returned {len(leaked)} tasks from another tenant"
    )


@pytest.mark.asyncio
async def test_insert_with_other_tenant_id_is_rejected(session_maker):
    """
    Given tenant A's context,
    when you try to insert a task with tenant_id = B,
    then PostgreSQL rejects the INSERT (WITH CHECK).
    """
    from sqlalchemy.exc import IntegrityError, ProgrammingError

    async with session_maker() as session:
        async with session.begin():
            await session.execute(text("SET LOCAL app.tenant_id = '1'"))

            # A malicious attempt: insert a task with someone else's tenant_id
            intruder_task = Task(tenant_id=2, title="INTRUDER")
            session.add(intruder_task)

            with pytest.raises((IntegrityError, ProgrammingError)) as exc_info:
                await session.flush()

            # PostgreSQL returns an error mentioning "row-level security policy"
            assert "row-level security" in str(exc_info.value).lower() or \
                   "row level security" in str(exc_info.value).lower()


@pytest.mark.asyncio
async def test_query_without_tenant_context_returns_empty(session_maker):
    """
    Given app.tenant_id is NOT set (a policy with missing_ok),
    when a SELECT runs,
    then it returns 0 rows (NO error, NO leak).
    """
    async with session_maker() as session:
        async with session.begin():
            # We deliberately do NOT set app.tenant_id
            result = await session.execute(select(Task))
            tasks = list(result.scalars().all())

    assert tasks == [], "With no tenant context, queries have to return 0 rows"

Run:

pytest tests/test_rls_isolation.py -v

Expected output:

test_rls_isolation.py::test_tenant_a_cannot_see_tenant_b_data PASSED
test_rls_isolation.py::test_malicious_query_with_explicit_other_tenant_returns_empty PASSED
test_rls_isolation.py::test_insert_with_other_tenant_id_is_rejected PASSED
test_rls_isolation.py::test_query_without_tenant_context_returns_empty PASSED

These four tests are your continuous guarantee. If someone breaks the setup (changes the policy, disables FORCE, connects as the owner by mistake), at least one will fail. Run them in CI on every PR.

Why does this matter in real work?

1. It's the exact code TaskFlow is going to use. Module 8 scales this same structure into a complete API. If you understand the dependency, the asyncpg gotcha, and the isolation tests, you already have TaskFlow's foundation ready.

2. It's the code that closes enterprise deals. Buyers who ask for an "isolation guarantee" want to see: (a) the policies in SQL, (b) the dependency in code, (c) the automated tests that prove the isolation. Those three artifacts are the defensible answer.

3. The asyncpg gotcha is invisible until production. In development with little traffic, prepared statements cause no visible problems. The intermittent leak shows up under real load. Having statement_cache_size=0 configured from the start saves you a painful postmortem.

4. A badly configured PgBouncer destroys RLS. Teams that don't understand the difference between transaction mode and session mode with SET vs SET LOCAL cause leaks that seem like "magic": one customer sees another's data intermittently, with no clear pattern, until the connection pooling gets debugged.

Traps and common mistakes

Mistake 1 (conceptual and EVEN MORE IMPORTANT): repeating the caveat — RLS is NOT for auth

Symptom: a dev gets to this capsule, sees SET LOCAL app.tenant_id in the dependency, and thinks "I can do the same with SET LOCAL app.user_id for per-user permissions." Three months later they have 30 policies with per-user permission logic and the team's productivity collapses.

Why it happens: the dependency looks like a general pattern applicable to any "request context." But per-user permissions have a very different complexity from the tenant's: they change per action, they depend on relationships (is the user the owner of this project?), they require multiple policies per table.

How to tell: ask "does my policy's predicate depend only on the tenant, or also on the user?". If it depends on the user, it is NOT multi-tenancy.

How to fix it: RLS only with app.tenant_id. For user permissions, FastAPI dependencies that check explicitly:

# The CORRECT pattern for per-user permissions (do NOT use RLS)
async def can_edit_project(
    project_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_tenant_session),
):
    result = await db.execute(
        select(Project).where(Project.id == project_id)
    )
    project = result.scalar_one_or_none()
    if project is None:
        raise HTTPException(status_code=404)
    if project.owner_id != current_user.id and not current_user.is_admin:
        raise HTTPException(status_code=403, detail="Not authorized to edit")
    return project

Auth/RBAC is covered in guide #9. Don't mix it in here.

Mistake 2 (operational): not disabling asyncpg's prepared statements cache

Symptom: everything works in development. In production, under load, some customers report seeing other tenants' data intermittently. The bug doesn't reproduce locally. The logs show nothing odd.

Why it happens: asyncpg caches prepared statements and under load with PgBouncer, one tenant's cached plan can get applied to another's queries.

How to tell: review the engine's creation. Does it have connect_args={"statement_cache_size": 0}? If not, you're vulnerable to the gotcha.

How to fix it: add statement_cache_size=0 whenever you use RLS. Document in the repo why (link to this capsule).

Mistake 3 (operational): using SET instead of SET LOCAL

Symptom: the app uses SET app.tenant_id = ... instead of SET LOCAL. In development it works because each request uses its own connection. In production with PgBouncer, the settings persist on connections that go back to the pool. Client A sets tenant=1, returns the connection, client B takes it and inherits tenant=1 without knowing. Leak.

Why it happens: SET gets confused with SET LOCAL. The conceptual difference is: SET is "for this session" (as long as the connection exists), SET LOCAL is "for this transaction." With pooling, "session" and "request" are different things.

How to tell: search for SET (without LOCAL) in your code that touches RLS settings. If you find it, it's a bug.

How to fix it: always SET LOCAL. No exceptions.

Mistake 4 (operational): the yield outside the session.begin()

Symptom: the dependency looks correct, but the endpoint's queries fail with an error or return 0 rows. EXPLAIN shows current_setting returning NULL.

Why it happens: the yield session is outside the async with session.begin():. The transaction where SET LOCAL ran has already closed before the endpoint runs its queries. The endpoint operates in a new transaction with no context.

How to tell: look at the yield's indentation in the dependency. It has to be inside the async with session.begin():, not outside.

How to fix it:

# ❌ WRONG: the yield outside begin()
async def get_tenant_session(tenant_id: int = Depends(get_current_tenant_id)):
    async with SessionLocal() as session:
        async with session.begin():
            await session.execute(text(f"SET LOCAL app.tenant_id = {tenant_id}"))
        yield session  # ← The transaction already closed here

# ✅ RIGHT: the yield inside begin()
async def get_tenant_session(tenant_id: int = Depends(get_current_tenant_id)):
    async with SessionLocal() as session:
        async with session.begin():
            await session.execute(text(f"SET LOCAL app.tenant_id = {tenant_id}"))
            yield session  # ← The same transaction context

Mistake 5 (operational): tests that pass locally but not in CI

Symptom: the isolation tests pass locally but fail in CI with strange errors about policies or connection pooling.

Why it happens: very often the local setup connects as postgres (the owner) and RLS doesn't apply. The tests "pass" because RLS is silently disabled. In CI with a different user, RLS applies and the tests reveal bugs.

How to tell: verify your test role is app_user (not postgres). And that FORCE ROW LEVEL SECURITY is enabled.

How to fix it: configure the test environment (CI and local) to always use the non-owner role:

# tests/conftest.py
import os

# Force the app_user role in tests, fail if it isn't configured.
TEST_DATABASE_URL = os.getenv(
    "TEST_DATABASE_URL",
    "postgresql+asyncpg://app_user:app_password_secure@localhost:5432/rls_demo_test"
)
assert "app_user" in TEST_DATABASE_URL or "appuser" in TEST_DATABASE_URL, (
    "Tests must run as app_user (not postgres/owner) for RLS to apply"
)

Mistake 6 (conceptual): assuming RLS protects against attackers with sophisticated SQL injection

Symptom: someone argues "since we have RLS, we can relax input sanitization because PostgreSQL protects us."

Why it happens: RLS does protect against many cross-tenant SQL injection techniques (UNION, OR 1=1, etc.). But it does NOT protect against other vectors: mass deletion within the attacker's own tenant (DELETE FROM tasks), exfiltrating the attacker's own tenant's data to a controlled endpoint, privilege escalation if the app_user role gets compromised.

How to tell: does the app use bind parameters for every query? Is there input validation? RLS does NOT replace that.

How to fix it: RLS is ONE layer of defense, not THE layer. Keep: input sanitization, bind parameters for every query, least privilege for the app_user role, log auditing. RLS adds to all of that, it doesn't replace it.

Exercises

Exercise 1: implement the dependency from scratch

Implement the get_tenant_session function for a new FastAPI app. Assume get_current_tenant_id already exists (it returns an int). It has to:

1. Open a SQLAlchemy async session.
2. Start an explicit transaction.
3. Run SET LOCAL app.tenant_id with the validated ID.
4. Yield the session to the endpoint.
5. Handle errors correctly (automatic rollback).

# app/db/tenant_context.py
from typing import AsyncIterator
from fastapi import Depends
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.db.session import SessionLocal
from app.auth import get_current_tenant_id


async def get_tenant_session(
    tenant_id: int = Depends(get_current_tenant_id),
) -> AsyncIterator[AsyncSession]:
    """
    Yields a SQLAlchemy session with tenant context set via SET LOCAL.

    Critical:
    - SET LOCAL must be in the same transaction as queries.
    - tenant_id must be validated as int by upstream dependency.
    - statement_cache_size=0 must be configured in the engine.
    """
    async with SessionLocal() as session:
        async with session.begin():
            # tenant_id is int (validated). Safe to interpolate.
            await session.execute(
                text(f"SET LOCAL app.tenant_id = {tenant_id}")
            )
            yield session
        # session.begin() commits on success, rolls back on exception.
        # SET LOCAL is discarded automatically when transaction ends.

Exercise 2: identify bugs in a badly written dependency

This dependency has 3 different bugs. Identify them and fix them.

# The buggy version
async def get_tenant_session(
    request: Request,
):
    tenant_id = request.headers.get("X-Tenant-ID")
    async with SessionLocal() as session:
        await session.execute(
            text(f"SET app.tenant_id = '{tenant_id}'")
        )
        yield session

async def get_current_tenant_id(request: Request) -> int:
    tenant_id_str = request.headers.get("X-Tenant-ID")
    if not tenant_id_str:
        raise HTTPException(status_code=401, detail="Missing X-Tenant-ID")
    try:
        return int(tenant_id_str)
    except ValueError:
        raise HTTPException(status_code=400, detail="Invalid X-Tenant-ID")


async def get_tenant_session(
    tenant_id: int = Depends(get_current_tenant_id),
):
    async with SessionLocal() as session:
        async with session.begin():
            await session.execute(
                text(f"SET LOCAL app.tenant_id = {tenant_id}")
            )
            yield session

Exercise 3: configure the engine to avoid the asyncpg gotcha

You have this engine configuration. Modify it so it's safe with production RLS.

# The NOT-safe version for RLS
from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "postgresql+asyncpg://app_user:secret@localhost/rls_demo"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,
    pool_size=20,
    max_overflow=10,
)

from sqlalchemy.ext.asyncio import create_async_engine

DATABASE_URL = "postgresql+asyncpg://app_user:secret@localhost/rls_demo"

engine = create_async_engine(
    DATABASE_URL,
    echo=True,
    pool_size=20,
    max_overflow=10,
    # CRITICAL for RLS: disable asyncpg's prepared statements cache.
    # Without this, cached plans can get applied with the wrong tenant
    # context under load, causing intermittent leaks.
    connect_args={
        "statement_cache_size": 0,
        "prepared_statement_cache_size": 0,
    },
)

import logging
logging.getLogger("asyncpg").setLevel(logging.DEBUG)

Exercise 4: write an end-to-end isolation test

Write a pytest test that uses FastAPI's TestClient to:

1. Make a request to GET /tasks with X-Tenant-ID: 1.
2. Make a request to GET /tasks with X-Tenant-ID: 2.
3. Verify the two responses share no tasks (there's no leak).
4. Make a request to GET /tasks/{id} with X-Tenant-ID: 1 but using a task ID that belongs to tenant 2.
5. Verify it returns a 404 (no leak via a direct ID).

# tests/test_api_isolation.py
import pytest
from httpx import AsyncClient
from app.main import app  # FastAPI app


@pytest.mark.asyncio
async def test_api_isolation_between_tenants(seeded_db):
    """
    The seeded_db fixture assumes tasks already exist for tenants 1 and 2.
    """
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Tenant 1's request
        response_a = await client.get(
            "/tasks",
            headers={"X-Tenant-ID": "1"},
        )
        assert response_a.status_code == 200
        tasks_a = response_a.json()

        # Tenant 2's request
        response_b = await client.get(
            "/tasks",
            headers={"X-Tenant-ID": "2"},
        )
        assert response_b.status_code == 200
        tasks_b = response_b.json()

        # No task should appear in both lists
        ids_a = {t["id"] for t in tasks_a}
        ids_b = {t["id"] for t in tasks_b}
        assert ids_a.isdisjoint(ids_b), (
            f"BREACH: Tasks visible to both tenants: {ids_a & ids_b}"
        )

        # Verify each list has tasks (not empty lists)
        assert len(tasks_a) > 0, "Tenant 1 should see its tasks"
        assert len(tasks_b) > 0, "Tenant 2 should see its tasks"


@pytest.mark.asyncio
async def test_cannot_get_other_tenant_task_by_id(seeded_db):
    """
    Tenant 1 tries to access a tenant 2 task by a direct ID.
    It has to return a 404 (indistinguishable from "doesn't exist").
    """
    async with AsyncClient(app=app, base_url="http://test") as client:
        # First, get a task ID from tenant 2 (with its own context)
        response_b = await client.get(
            "/tasks",
            headers={"X-Tenant-ID": "2"},
        )
        assert response_b.status_code == 200
        tasks_b = response_b.json()
        assert len(tasks_b) > 0
        target_task_id = tasks_b[0]["id"]

        # Tenant 1 tries to access tenant 2's task by ID
        response_a = await client.get(
            f"/tasks/{target_task_id}",
            headers={"X-Tenant-ID": "1"},
        )

        # It has to be a 404 — RLS makes the task invisible, the endpoint doesn't find it
        assert response_a.status_code == 404, (
            f"BREACH: Tenant 1 was able to access task {target_task_id} "
            f"of tenant 2. Response: {response_a.text}"
        )

Exercise 5: diagnose an intermittent leak

In production, the team reports this symptom: "the Acme customer reported seeing 3 strange tasks in their dashboard they never created. We deleted them manually. Three days later, the Globex customer reported something similar. The isolation tests pass in CI. The logs show no errors."

What are the 3 most likely causes? How would you diagnose them?

Exercise 6: extend the tests to cover cross-tenant DELETE

Exercise 4's tests cover SELECT (a list and by ID). Extend them to verify:

1. A tenant can NOT delete another tenant's tasks even if it knows the ID.
2. A tenant can NOT update another tenant's tasks even if it knows the ID.
3. A tenant can NOT insert tasks attributed to another tenant.

# tests/test_api_isolation_writes.py
import pytest
from httpx import AsyncClient
from app.main import app


@pytest.mark.asyncio
async def test_cannot_delete_other_tenant_task(seeded_db):
    """Tenant 1 tries to delete a tenant 2 task."""
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Get a task ID from tenant 2
        response_b = await client.get("/tasks", headers={"X-Tenant-ID": "2"})
        target_task_id = response_b.json()[0]["id"]

        # Tenant 1 tries to delete it
        response_delete = await client.delete(
            f"/tasks/{target_task_id}",
            headers={"X-Tenant-ID": "1"},
        )

        # It has to be a 404 (RLS makes the task invisible to the delete)
        assert response_delete.status_code == 404

        # Verify the task STILL exists in tenant 2
        response_b_after = await client.get(
            f"/tasks/{target_task_id}",
            headers={"X-Tenant-ID": "2"},
        )
        assert response_b_after.status_code == 200, (
            "BREACH: tenant 2's task was deleted by tenant 1"
        )


@pytest.mark.asyncio
async def test_cannot_update_other_tenant_task(seeded_db):
    """Tenant 1 tries to update a tenant 2 task."""
    async with AsyncClient(app=app, base_url="http://test") as client:
        # Get a task from tenant 2
        response_b = await client.get("/tasks", headers={"X-Tenant-ID": "2"})
        target_task = response_b.json()[0]
        target_task_id = target_task["id"]
        original_title = target_task["title"]

        # Tenant 1 tries to update it
        response_update = await client.patch(
            f"/tasks/{target_task_id}",
            json={"title": "MODIFIED BY ATTACKER"},
            headers={"X-Tenant-ID": "1"},
        )

        # It has to be a 404
        assert response_update.status_code == 404

        # Verify the title did NOT change in tenant 2
        response_b_after = await client.get(
            f"/tasks/{target_task_id}",
            headers={"X-Tenant-ID": "2"},
        )
        assert response_b_after.status_code == 200
        assert response_b_after.json()["title"] == original_title, (
            f"BREACH: tenant 2's task title was modified to "
            f"'{response_b_after.json()['title']}'"
        )


@pytest.mark.asyncio
async def test_cannot_insert_task_with_other_tenant_id(seeded_db):
    """
    Tenant 1 tries to insert a task with an explicit tenant 2 tenant_id.
    The POST /tasks endpoint should ignore/reject the body's tenant_id
    (always use the context's). But if by a bug it accepts the body's,
    RLS rejects it with WITH CHECK.
    """
    async with AsyncClient(app=app, base_url="http://test") as client:
        response_create = await client.post(
            "/tasks",
            json={
                "title": "INTRUDER",
                "tenant_id": 2,  # An attempt to attribute it to another tenant
            },
            headers={"X-Tenant-ID": "1"},
        )

        # The endpoint should:
        # - Ignore the body's tenant_id and use the context's (recommended)
        # - Or fail with a 4xx from validation
        # In either case, the task must NOT end up attributed to tenant 2

        if response_create.status_code == 201 or response_create.status_code == 200:
            # If it got created, verify tenant_id is 1, not 2
            created = response_create.json()
            assert created["tenant_id"] == 1, (
                f"BREACH: task created with tenant_id={created['tenant_id']} "
                f"from request of tenant 1"
            )
        else:
            # A 4xx error is also valid (validation rejected the input)
            assert 400 <= response_create.status_code < 500


@pytest.mark.asyncio
async def test_db_rejects_direct_insert_with_other_tenant_id(seeded_db, session_maker):
    """
    A direct DB-level test: trying an INSERT with someone else's tenant_id
    has to fail with an RLS WITH CHECK error.
    """
    from sqlalchemy import text
    from sqlalchemy.exc import IntegrityError, ProgrammingError

    async with session_maker() as session:
        async with session.begin():
            await session.execute(text("SET LOCAL app.tenant_id = '1'"))

            with pytest.raises((IntegrityError, ProgrammingError)) as exc_info:
                await session.execute(text(
                    "INSERT INTO tasks (tenant_id, title) "
                    "VALUES (2, 'DIRECT INTRUSION')"
                ))
                await session.flush()

            error_msg = str(exc_info.value).lower()
            assert "row-level security" in error_msg or "row level security" in error_msg

Summary and next step

In this capsule you learned:

• The dependency is the "gateway" to the tenant: it runs SET LOCAL app.tenant_id in an explicit transaction BEFORE the endpoint runs its queries.
• asyncpg's critical gotcha: the prepared statements cache can cause intermittent leaks under load. The solution: statement_cache_size=0 in the engine.
• PgBouncer in transaction mode + SET LOCAL is the correct combination. SET (without LOCAL) never with pooling.
• Defense in depth: keep the WHERE tenant_id in the code even though RLS also filters. It documents intent and protects in cases where RLS doesn't apply.
• Automated isolation tests are mandatory: reads, writes, INSERT with someone else's tenant, a query with no context. They pass in CI on every PR.
• The caveat repeated (yes, three times in the module): RLS only for multi-tenancy. For user permissions, RBAC in the application layer. Auth is covered in guide #9.

Before moving on you should be able to:

• Implement get_tenant_session correctly (an explicit transaction, the yield in the right place, SET LOCAL).
• Configure the SQLAlchemy + asyncpg engine to avoid the prepared statements gotcha.
• Configure PgBouncer (or understand why your setup doesn't need it) compatibly with RLS.
• Write end-to-end isolation tests with FastAPI's TestClient.
• Diagnose intermittent leaks in production (the three most likely causes).
• Distinguish when to add more WHERE tenant_id vs when to trust RLS.

Next capsule — Schema-per-tenant: when and how. You already have production RLS working. Now you're going to learn the spectrum's "most extreme" option: schema-per-tenant. When you really need it (spoiler: very rarely outside pure enterprise), how it gets implemented with search_path and SQLAlchemy, what happens when you have to migrate 500 schemas with Alembic, and what the real operational cost is (not the "elegance" that gets sold). It's the capsule that completes your judgment about the 3 models: you already saw the simplest (capsule 03), the default (RLS, capsules 04-05), and you're going to see the heaviest (schema-per-tenant) to have the complete picture.

Resources

1. SQLAlchemy 2.0 — Async ORM tutorial — the official reference for the async ORM.
2. asyncpg — Connection pool — the documentation for the pool's parameters, including statement_cache_size.
3. SQLAlchemy — connect_args reference — how to pass parameters to the underlying driver (asyncpg in this case).
4. PgBouncer — Pool modes documentation — the differences between transaction, session, and statement modes. Critical to understand for using RLS with pooling.
5. FastAPI — Dependencies with yield — the pattern used in get_tenant_session.
6. Supabase — Client-side auth + RLS — a production implementation model many APIs draw inspiration from.
7. Crunchy Data — Connection pooling with RLS — a specific analysis of the PgBouncer + RLS combination.
8. GitLab — Database security through RLS — postmortems and implementation patterns at scale (search the blog).

Module 4 — SQL Patterns for Production APIs Guide

Next capsule: Schema-per-tenant: when and how — the spectrum's heaviest option, when it's really worth paying the operational cost.