Enforcing tenant isolation with Postgres row-level security

Create a policy USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')::bigint), enable and force row-level security, and set the tenant per transaction with set_config('app.tenant_id', :tenant, true) from a session after_begin listener — PostgreSQL then filters every statement, including the ones someone forgot to filter. This guide belongs to dynamic schema and multi-tenant routing.

Quick Answer

Shared-table multi-tenancy usually starts with a tenant_id filter on every query, and eventually a query ships without one. Row-level security moves the rule into the database, where it cannot be forgotten.

Before — isolation depends on every query remembering:

from sqlalchemy import select

from shop.models import Order


async def recent_orders(session, tenant_id: int):
    return (await session.scalars(
        select(Order).where(Order.tenant_id == tenant_id).order_by(Order.id.desc()).limit(50)
    )).all()


async def export_all_orders(session):
    # Written in a hurry for a support ticket. No tenant filter.
    return (await session.scalars(select(Order))).all()

After — a policy in the database, the tenant set per transaction:

-- migration, run as the table owner
ALTER TABLE orders ENABLE ROW LEVEL SECURITY;
ALTER TABLE orders FORCE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON orders
    USING (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')::bigint)
    WITH CHECK (tenant_id = NULLIF(current_setting('app.tenant_id', true), '')::bigint);

import contextvars

from sqlalchemy import event, text
from sqlalchemy.orm import Session

current_tenant: contextvars.ContextVar[int | None] = contextvars.ContextVar(
    "current_tenant", default=None
)


@event.listens_for(Session, "after_begin")
def _set_tenant(session, transaction, connection) -> None:
    tenant_id = current_tenant.get()
    if tenant_id is None:
        return   # no tenant: the policy matches no rows
    connection.execute(
        text("SELECT set_config('app.tenant_id', :tenant_id, true)"),
        {"tenant_id": str(tenant_id)},
    )

Now export_all_orders() returns only the current tenant's orders, and a code path that runs without a tenant sees nothing at all.

Execution Context & Async Workflow Integration

A row-level security policy is a predicate PostgreSQL adds to every query against the table, for every role the policy applies to. USING filters rows that SELECT, UPDATE and DELETE can see; WITH CHECK rejects INSERT and UPDATE rows that would not satisfy it. The predicate here reads a custom configuration setting, app.tenant_id, which any session can set and which the policy reads with current_setting().

The second argument to current_setting('app.tenant_id', true) is missing_ok. When the setting was never set on the connection, it returns NULL instead of raising. There is a subtlety with pooled connections: once any earlier transaction on that connection has set the parameter locally, PostgreSQL remembers that the parameter exists, and afterwards it returns an empty string rather than NULL. Casting '' to bigint raises, so the policy wraps the value in NULLIF(..., ''). With that, an unset tenant becomes NULL in both cases, tenant_id = NULL is never true, and the policy matches no rows. That is the fail-closed behaviour you want: forgetting to set the tenant produces empty results, not every tenant's data.

The setting must be scoped to a transaction, which is what the third argument of set_config(..., true) does — is_local. A transaction-local setting disappears at COMMIT or ROLLBACK. The alternative, a plain SET app.tenant_id = 42, lasts for the whole database session, and with a connection pool the database session outlives your request: the next request that checks out that connection inherits tenant 42. SET LOCAL is also transaction-scoped, but it cannot take a bound parameter, and interpolating a tenant identifier into SQL text is exactly the kind of mistake this design is meant to remove.

after_begin is the right hook because it fires once per transaction, on the connection the transaction is using, before any of the transaction's statements. It is registered on the synchronous Session class, which every AsyncSession wraps, so it applies to async code without change; inside the listener, connection.execute() is synchronous because it already runs inside SQLAlchemy's greenlet bridge. The tenant itself comes from a ContextVar set by middleware, which asyncio copies into every task the request creates — the same mechanism described for per-request state in switching schemas per request with schema_translate_map.

The pattern also works behind PgBouncer in transaction pooling mode, because transaction-local settings are exactly as long-lived as a transaction-pooled server connection assignment. Session-level SET does not, for the same reason it leaks through SQLAlchemy's pool — and the related prepared-statement issues are covered in handling asyncpg prepared statement errors with PgBouncer.

Resolving Warnings, Errors & Common Mistakes

The bypass rows matter most, because nothing reports them. PostgreSQL applies policies to every role except superusers, roles with BYPASSRLS, and — unless FORCE ROW LEVEL SECURITY is set on the table — the table's owner. Many deployments run the application as the role that ran the migrations, which owns every table, so policies are created, enabled and silently ignored. Check which role the application actually uses:

from sqlalchemy import text


async def rls_status(session) -> list[dict]:
    rows = await session.execute(text(
        "SELECT c.relname AS table_name, c.relrowsecurity AS enabled, "
        "       c.relforcerowsecurity AS forced, "
        "       pg_get_userbyid(c.relowner) = current_user AS app_is_owner, "
        "       (SELECT rolbypassrls OR rolsuper FROM pg_roles WHERE rolname = current_user) "
        "         AS app_bypasses "
        "FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace "
        "WHERE n.nspname = 'public' AND c.relkind = 'r'"
    ))
    return [dict(row._mapping) for row in rows]

Any row with enabled but app_bypasses, or app_is_owner without forced, is a table whose policy does nothing for this application.

Advanced: Performance, Defaults and Cross-Tenant Work

A policy is a predicate like any other, and PostgreSQL's planner treats it that way: current_setting() is a stable function, so tenant_id = current_setting(...)::bigint can use an index on tenant_id. Make sure that index exists, and for tables usually queried by tenant and time, lead composite indexes with tenant_id. Keeping the explicit .where(Order.tenant_id == tenant_id) in application queries is still worthwhile: it documents intent, and it gives the planner a plain parameter to estimate with.

The WITH CHECK clause rejects writes for the wrong tenant, and a column default can remove the need to set tenant_id in Python at all:

ALTER TABLE orders
    ALTER COLUMN tenant_id SET DEFAULT NULLIF(current_setting('app.tenant_id', true), '')::bigint;

In the model, declare it as server-generated so the ORM does not try to supply a value and reads back what the database chose:

from sqlalchemy import BigInteger, FetchedValue
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


class Base(DeclarativeBase):
    pass


class Order(Base):
    __tablename__ = "orders"
    id: Mapped[int] = mapped_column(primary_key=True)
    tenant_id: Mapped[int] = mapped_column(BigInteger, server_default=FetchedValue(), index=True)
    total_cents: Mapped[int]

Some work legitimately spans tenants: billing runs, analytics exports, support tooling. Two designs keep that explicit. A separate database role with its own policy — for example USING (true) for a reporting role — used by a separate engine makes cross-tenant access a deployment decision rather than a code path. Or a job iterates tenants and sets each one in turn, which keeps even batch work inside the isolation model:

from shop.db import Session
from shop.tenancy import current_tenant


async def run_for_every_tenant(tenant_ids: list[int], job) -> None:
    for tenant_id in tenant_ids:
        token = current_tenant.set(tenant_id)
        try:
            async with Session() as session:
                await job(session)
                await session.commit()
        finally:
            current_tenant.reset(token)

Each iteration opens a new session, so each gets a new transaction and a fresh after_begin with the right tenant. Reusing one session across tenants would keep the first tenant's setting for as long as its transaction stayed open.

Testing Tenant Isolation

Isolation is a security property, and it deserves tests that would fail if it broke — which means tests that run under the same conditions as production. Most RLS test suites that pass while enforcing nothing have one flaw: they connect as a superuser or the table owner.

Create the application role in the test database the same way as in production, run migrations as the owner, and give the application-level fixtures an engine that connects as the application role. Then write tests from an attacker's point of view:

import pytest
from sqlalchemy import select, text

from shop.models import Order
from shop.tenancy import current_tenant


@pytest.mark.asyncio
async def test_unfiltered_query_sees_only_current_tenant(app_session_factory, seed_orders):
    await seed_orders(tenant_id=1, count=3)
    await seed_orders(tenant_id=2, count=5)

    token = current_tenant.set(1)
    try:
        async with app_session_factory() as session:
            orders = (await session.scalars(select(Order))).all()   # no filter on purpose
    finally:
        current_tenant.reset(token)

    assert {order.tenant_id for order in orders} == {1}
    assert len(orders) == 3


@pytest.mark.asyncio
async def test_no_tenant_sees_nothing(app_session_factory, seed_orders):
    await seed_orders(tenant_id=1, count=3)
    async with app_session_factory() as session:
        assert (await session.scalars(select(Order))).all() == []


@pytest.mark.asyncio
async def test_setting_does_not_leak_through_the_pool(app_engine, app_session_factory, seed_orders):
    await seed_orders(tenant_id=1, count=3)
    token = current_tenant.set(1)
    try:
        async with app_session_factory() as session:
            await session.scalars(select(Order))
    finally:
        current_tenant.reset(token)
    async with app_engine.connect() as conn:          # likely the same pooled connection
        leaked = await conn.scalar(text("SELECT current_setting('app.tenant_id', true)"))
    assert leaked in (None, "")

seed_orders must insert as a role that can write any tenant's rows — the owner, or with the tenant set per insert — otherwise the fixture itself is blocked by the policy. Add one more test that asserts rls_status() from the previous section reports no bypasses for the application role, so a future change to how the test database is provisioned cannot quietly turn the whole suite into a no-op. The general fixture patterns are in running tests against a Postgres testcontainer.

Frequently Asked Questions

Related

• Dynamic Schema and Multi-Tenant Routing — The parent guide: schema routing, replicas and tenant models.
• Switching schemas per request with schema_translate_map — The schema-per-tenant alternative.
• Routing reads to replicas with async engines — Keeping the tenant setting consistent across engines.
• Using async_scoped_session with asyncio tasks — How ContextVars flow into tasks a request spawns.