SQLAlchemy Tenant Scoping with Session Events

Adding .filter_by(tenant_id=...) to every query is a control that fails the first time someone forgets, so this page moves it into the session itself. It is a focused part of ORM Middleware for Multi-Tenancy: binding tenant context to a SQLAlchemy session, filtering automatically, and keeping Row-Level Security underneath as the layer that cannot be forgotten.

Problem Framing

SQLAlchemy gives three natural insertion points, and choosing badly produces a scoping layer with holes. A query-level filter applied by hand is complete only if every author remembers. A with_loader_criteria option applied per query covers relationship loads too, but still has to be attached each time. The do_orm_execute event fires for every ORM execution on the session, including lazy loads and relationship traversal, which is what makes it the right hook.

Even then, the ORM layer cannot be the only control. Raw text() queries, session.execute() on Core statements, bulk operations, and anything running through a different session bypass it entirely. That is not a reason to skip the ORM layer — it removes the overwhelming majority of mistakes — but it is the reason Row-Level Security has to sit underneath, so a bypass produces zero rows rather than everyone's.

The third piece is the tenant itself. It must come from a context variable that the request middleware sets, never from a function argument, so that a lazy load triggered deep in a serializer still sees the right tenant.

Bar width is coverage. The ORM event removes almost all mistakes; the database layer is what makes the remainder safe.

Step-by-Step Guide

1. Hold the tenant in a context variable

# tenant_context.py
from contextvars import ContextVar
from contextlib import contextmanager
from uuid import UUID

_current_tenant: ContextVar[UUID | None] = ContextVar("current_tenant", default=None)

@contextmanager
def tenant_scope(tenant_id: UUID):
    token = _current_tenant.set(tenant_id)
    try:
        yield
    finally:
        _current_tenant.reset(token)

def current_tenant() -> UUID:
    tid = _current_tenant.get()
    if tid is None:
        raise RuntimeError("no tenant bound to this execution context")
    return tid

A ContextVar is the right primitive because it follows async execution correctly and is isolated per task — a module-level global would be shared across concurrent requests on the same worker, which is precisely the shared-state bug this is meant to avoid.

2. Mark which models are tenant-scoped

# models.py
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import UUID as SAUUID

class Base(DeclarativeBase):
    pass

class TenantScoped:
    """Marker mixin: presence of this class is what activates automatic filtering."""
    tenant_id: Mapped[UUID] = mapped_column(SAUUID(as_uuid=True), nullable=False, index=True)

class Invoice(Base, TenantScoped):
    __tablename__ = "invoice"
    id: Mapped[UUID] = mapped_column(SAUUID(as_uuid=True), primary_key=True)
    amount_cents: Mapped[int]

class Plan(Base):                      # deliberately global: shared reference data
    __tablename__ = "plan"
    key: Mapped[str] = mapped_column(primary_key=True)

An explicit marker beats "has a tenant_id column" because it makes the decision visible in the model, and it lets genuinely global tables opt out without a special case in the event handler.

3. Filter automatically with `do_orm_execute`

# scoping.py
from sqlalchemy import event, orm
from sqlalchemy.orm import Session, with_loader_criteria

@event.listens_for(Session, "do_orm_execute")
def _apply_tenant_filter(execute_state: orm.ORMExecuteState) -> None:
    if execute_state.is_column_load or execute_state.is_relationship_load:
        return                                     # already scoped by the parent object
    if execute_state.execution_options.get("skip_tenant_filter", False):
        return                                     # explicit, greppable escape hatch

    tid = current_tenant()
    execute_state.statement = execute_state.statement.options(
        with_loader_criteria(
            TenantScoped,
            lambda cls: cls.tenant_id == tid,
            include_aliases=True,                  # applies to joined and aliased entities too
        )
    )

with_loader_criteria against the mixin applies to every subclass at once, so adding a new tenant-scoped model requires nothing beyond inheriting the mixin. include_aliases=True is what covers joins and eager loads, which is where hand-written filters usually fall short.

4. Set the database-side scope on every connection checkout

# rls_binding.py
from sqlalchemy import event, text
from sqlalchemy.engine import Engine

@event.listens_for(Engine, "begin")
def _set_tenant_guc(conn) -> None:
    tid = _current_tenant.get()
    if tid is not None:
        # Transaction-local: cleared automatically at COMMIT or ROLLBACK.
        conn.exec_driver_sql("SELECT set_config('app.current_tenant_id', %s, true)", (str(tid),))

ALTER TABLE invoice ENABLE ROW LEVEL SECURITY;
ALTER TABLE invoice FORCE ROW LEVEL SECURITY;
CREATE POLICY invoice_tenant ON invoice
  USING      (tenant_id = current_setting('app.current_tenant_id', true)::uuid)
  WITH CHECK (tenant_id = current_setting('app.current_tenant_id', true)::uuid);

Binding on begin rather than on connect matters under connection pooling: the setting must be transaction-local so a pooled connection handed to another tenant carries nothing over, exactly as described in Connection Pooling in Multi-Tenant Systems.

5. Stamp the tenant on insert, so writes cannot be unscoped either

# defaults.py
from sqlalchemy import event
from sqlalchemy.orm import Session

@event.listens_for(Session, "before_flush")
def _stamp_tenant(session: Session, flush_context, instances) -> None:
    tid = current_tenant()
    for obj in session.new:
        if isinstance(obj, TenantScoped):
            if obj.tenant_id is None:
                obj.tenant_id = tid
            elif obj.tenant_id != tid:
                raise PermissionError(f"cannot write tenant {obj.tenant_id} in scope {tid}")

Raising on a mismatched explicit value is the important half. It turns a copy-paste that carries another tenant's identifier into an exception at flush time instead of a cross-tenant write that the WITH CHECK policy would later reject with a much less helpful message.

Five hooks, all in infrastructure code. Application code stays tenant-unaware, which is what makes the guarantee hold as the team grows.

Verification

# test_scoping.py
import pytest
from sqlalchemy import text

def test_query_returns_only_current_tenant(session, seed_two_tenants):
    with tenant_scope(TENANT_A):
        rows = session.query(Invoice).all()
    assert {r.tenant_id for r in rows} == {TENANT_A}

def test_lazy_load_is_scoped(session, seed_two_tenants):
    with tenant_scope(TENANT_A):
        org = session.get(Organisation, ORG_A)
        assert all(i.tenant_id == TENANT_A for i in org.invoices)   # relationship traversal

def test_write_with_foreign_tenant_is_rejected(session):
    with tenant_scope(TENANT_A):
        session.add(Invoice(id=uuid4(), tenant_id=TENANT_B, amount_cents=100))
        with pytest.raises(PermissionError):
            session.flush()

def test_raw_sql_is_still_protected_by_rls(session, seed_two_tenants):
    with tenant_scope(TENANT_A):
        rows = session.execute(text("SELECT tenant_id FROM invoice")).all()
    assert {r.tenant_id for r in rows} == {TENANT_A}     # the ORM event did not fire; RLS did

def test_missing_context_raises_rather_than_returning_everything(session):
    with pytest.raises(RuntimeError):
        session.query(Invoice).all()

The fourth test is the one that proves the backstop works, and the fifth is the one that proves the system fails closed. Both should run as the unprivileged application role — connecting as the owner silently disables Row-Level Security and makes the fourth test pass for the wrong reason.

The escape hatch is necessary — billing runs genuinely span tenants — and its value depends entirely on staying small and reviewable.

Failure Modes & Gotchas

Symptom: queries return everything in a background task. Cause: the task runs outside the request, so the context variable is unset. Fix: make current_tenant() raise rather than default, and wrap every task entry point in tenant_scope() from the job payload.
Symptom: eager-loaded relationships include other tenants' rows. Cause: loader criteria applied without include_aliases=True, so joined and aliased entities are missed. Fix: set it, and add a test that traverses a relationship rather than only querying a root entity.
Symptom: a pooled connection carries the previous tenant's setting. Cause: the scope was set with a plain SET on connect instead of transaction-locally on begin. Fix: use set_config(..., true) in the begin event so it is discarded at commit.
Symptom: concurrent async requests see each other's tenant. Cause: the tenant is stored in a module-level global rather than a ContextVar. Fix: use ContextVar, and test with interleaved concurrent requests rather than sequentially.

FAQ

Does the ORM event slow queries down? The overhead is building one additional criteria option per execution — microseconds, and no extra round trip. The set_config call on transaction begin is one small statement per transaction, which is measurable but tiny next to the queries it precedes. The database-side cost of the policy predicate itself is the same as a hand-written WHERE tenant_id = ..., provided the indexes are right; the tuning is covered in RLS performance tuning with partial indexes.

Is Row-Level Security still needed if the ORM event is in place? Yes, and the fourth test above is why: text() queries, Core statements, bulk inserts and any code path using a different session bypass the ORM event entirely. Those paths exist in every codebase of a reasonable age. The ORM layer is what prevents the mistakes; the database layer is what makes the mistakes survivable.

How should genuinely cross-tenant queries be written? Through the explicit execution option, in a small number of audited places: session.query(Invoice).execution_options(skip_tenant_filter=True). Make it grep-able, require a comment naming the reason, run it under a database role permitted to bypass the policy, and count the call sites in CI so growth is visible. An escape hatch that is easy to find is far safer than one that is impossible to use.

How should this work with async SQLAlchemy? Essentially unchanged, which is the main argument for ContextVar over any other storage. The event hooks are the same, AsyncSession dispatches do_orm_execute identically, and the context follows the task correctly across awaits. The one difference worth testing is concurrency: run the interleaved test with two concurrent tasks sharing an engine, because a mistake that a synchronous test cannot expose — a context captured at module import, for instance — will surface immediately there.

Does this interfere with Alembic migrations? It should not, and if it does the wiring is in the wrong place. Migrations run with their own engine and their own role, outside any request context, so the session event never fires for them. Registering the listener at application startup rather than at model-module import keeps that separation clean; registering it as a side effect of importing the models means the migration environment picks it up and then fails with a missing tenant context, which is a confusing way to learn where the registration belongs.