SQLAlchemy models

The SQLAlchemy adapter is the sugar layer over the ORM-agnostic core. Its job is to derive table names, register policies, and run the set_config round-trip on a session.

Declaring policies with RLSMixin

Mix RLSMixin into a model and list its policies. On class creation the mixin binds each policy to __tablename__ and registers it into the default registry.

from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from fastapi_rls import TenantPolicy, UserPolicy
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"),
        UserPolicy("owner_can_write", column="owner_id", operation="UPDATE"),
    ]

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

Because the policies are bound to the table by the mixin, you do not pass table= here.

Using a specific registry

By default policies register into the global default_registry. To keep a model's policies in a separate registry, set __rls_registry__ on the class:

from fastapi_rls import PolicyRegistry

reporting_registry = PolicyRegistry()


class Report(Base, RLSMixin):
    __tablename__ = "reports"
    __rls_registry__ = reporting_registry
    __rls_policies__ = [TenantPolicy("tenant_isolation", column="tenant_id")]

Then point the facade at it: RLS(engine=engine, registry=reporting_registry).

collect_policies — the version-robust path

Some custom declarative bases don't propagate __init_subclass__ the way the mixin relies on. If your models aren't registering, call collect_policies once at startup to walk the mapper registry and register every model's policies explicitly:

from fastapi_rls.adapters.sqlalchemy import collect_policies

collect_policies(Base)                      # into default_registry
collect_policies(Base, reporting_registry)  # into a specific registry

This is idempotent and safe to call at import time.

Executors

The core context API talks to the database through an RLSExecutor protocol. The adapter provides two implementations that wrap a session:

• SessionExecutor(session) — runs SELECT set_config(:key, :value, :is_local) and SELECT current_setting(:key, true) on a sync Session.
• AsyncSessionExecutor(session) — the async equivalent for AsyncSession.

You rarely construct these yourself — the FastAPI dependencies do it for you — but they're public for advanced or non-FastAPI use.

Pool hygiene

install_pool_hygiene(engine) registers a checkin listener that scrubs rls.* session variables (or drops the connection) when it returns to the pool. The RLS facade installs it automatically for a sync engine when reset_context_on_checkout is enabled. It is a defense-in-depth backstop; the primary guarantee is the SET LOCAL transaction (see request context).

Applying policies to the database

Once models are registered, reconcile with the database via the facade:

from fastapi_rls import RLS

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

Or via the CLI / Alembic. Next: wire it into FastAPI requests.