Skip to main content

Quick start

This walks through a minimal multi-tenant app: declare a policy, apply it to the database, and scope every request to the caller's tenant.

1. Declare policies on your models​

Add RLSMixin to a model and list its policies in __rls_policies__. The mixin derives the table name and registers each policy for you.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from fastapi_rls import TenantPolicy
from fastapi_rls.adapters.sqlalchemy import RLSMixin


class Base(DeclarativeBase):
    pass


class Document(Base, RLSMixin):
    __tablename__ = "documents"
    __rls_policies__ = [
        TenantPolicy("tenant_isolation", column="tenant_id"),
    ]

    id: Mapped[int] = mapped_column(primary_key=True)
    tenant_id: Mapped[int] = mapped_column(index=True)
    title: Mapped[str]
tip

If your declarative base doesn't propagate __init_subclass__ (some custom base setups don't), call collect_policies(Base) once at startup to register every model's policies explicitly. See the SQLAlchemy guide.

2. Apply the policies to the database​

Build one RLS facade and let it reconcile your registry with the database. sync() enables and forces RLS on each table and creates every registered policy, in a single transaction.

from fastapi_rls import RLS

rls = RLS(engine=engine)   # your SQLAlchemy Engine
rls.sync()                 # ENABLE + FORCE RLS and CREATE every registered policy

Prefer migrations or a CLI? The same reconciliation is available:

# Standalone CLI, no Alembic required:
fastapi-rls sync --url "$DATABASE_URL" --policies myapp.models
# Inside an Alembic migration:
import fastapi_rls.alembic_ops  # noqa: F401 — registers op.enable_rls / op.create_policy

def upgrade():
    op.enable_rls("documents")
    op.create_policy(
        TenantPolicy("tenant_isolation", column="tenant_id", table="documents")
    )

3. Wire the request context​

The session dependency owns the transaction and applies your identity with SET LOCAL. You provide an identity callable that returns the context mapping — fastapi-rls never authenticates; it propagates the principal you resolve.

from fastapi import Depends, FastAPI
from sqlalchemy import select
from sqlalchemy.orm import Session
from fastapi_rls import RLS

rls = RLS(engine=engine)


def identity(user=Depends(get_current_user)) -> dict:
    return {"tenant_id": user.tenant_id}


get_session = rls.session_dependency(identity=identity)

app = FastAPI()


@app.get("/documents")
def list_documents(session: Session = Depends(get_session)):
    # PostgreSQL filters by tenant. No WHERE clause required.
    return session.scalars(select(Document)).all()

That's the whole integration. Every query on that session is transparently scoped to the caller's tenant. A request with no context sees no rows — never everyone's.

4. Async is identical​

from sqlalchemy.ext.asyncio import AsyncSession

rls = RLS(async_engine=async_engine)
get_session = rls.async_session_dependency(identity=identity)


@app.get("/documents")
async def list_documents(session: AsyncSession = Depends(get_session)):
    result = await session.scalars(select(Document))
    return result.all()

What just happened​

  1. TenantPolicy compiled to a predicate like tenant_id = (SELECT NULLIF(current_setting('rls.tenant_id', true), '')::integer).
  2. rls.sync() ran ENABLE/FORCE ROW LEVEL SECURITY and CREATE POLICY.
  3. The session dependency opened a transaction and ran SET LOCAL rls.tenant_id = … from your identity.
  4. PostgreSQL enforced the predicate on every statement; committing the transaction cleared the setting.

Read how policies work and the request context model next — or jump straight to the security model before shipping.