litestar-tenants

Multi-tenancy for Litestar apps, with two isolation modes: Postgres Row-Level Security (rls), or database-agnostic ORM-level scoping via SQLAlchemy events (orm-scoped).

These are not equivalent security guarantees. Under rls, every tenant gets its own Postgres role, and a raw psql session logged in as that role cannot see or write another tenant's rows, regardless of what SQL it runs - isolation is enforced by the database. Under orm-scoped, isolation lives in Python: a bug in this library, a raw SQL query, or a bypass of the ORM can leak across tenants. rls is the stronger default whenever you're on Postgres; orm-scoped exists for everyone else (or for tests against SQLite). Pick per your threat model, not by default.

Features

• Two isolation modes, one interface. Which mode is active is determined entirely by the provider instance you pass TenantsPlugin - OrmScopedSessionProvider or RLSSessionProvider - not by a separate mode flag.
• Postgres RLS with role management included. PostgresManager provisions a Postgres role per tenant, scopes sessions via SET SESSION ROLE, and setup_rls_schema sets up the enforcing policies - no Alembic dependency, no manual SQL.
• ORM-level scoping for any SQLAlchemy backend. TenantMixin + register_tenant_scoping() transparently filter every query and force-pin every write to the session's tenant, on SQLite, MySQL, or anywhere else Postgres isn't an option.
• Pluggable tenant resolution. HeaderTenantResolver, SubdomainTenantResolver, ClaimTenantResolver built in; TenantResolver is a one-method Protocol for anything else.
• Fail-closed. A resolver that can't determine a tenant raises TenantNotResolvedError rather than falling back to an unscoped session.
• One plugin. TenantsPlugin wires the provider into Litestar's DI as a single dependency; for orm-scoped it also calls register_tenant_scoping() so you don't have to remember to.

Install

pip install litestar-tenants

greenlet and typing-extensions come along as runtime dependencies automatically; no extras are required for either isolation mode. rls mode additionally needs a Postgres driver (e.g. asyncpg) at runtime - bring your own, it isn't bundled.

Quick start

orm-scoped mode needs no external database - mix in TenantMixin, register the event listeners, and wire a provider:

from litestar import Litestar, get
from litestar.di import NamedDependency
from sqlalchemy import Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from litestar_tenants import (
    HeaderTenantResolver,
    OrmScopedSessionProvider,
    TenantMixin,
    TenantsPlugin,
)


class Base(DeclarativeBase):
    pass


class Item(Base, TenantMixin):
    __tablename__ = "item"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]


engine = create_async_engine("sqlite+aiosqlite:///app.db")
session_maker = async_sessionmaker(engine)

plugin = TenantsPlugin(
    provider=OrmScopedSessionProvider(
        session_maker=session_maker,
        resolver=HeaderTenantResolver(),  # reads X-Tenant-ID by default
    )
)


@get("/items")
async def list_items(db_session: NamedDependency[AsyncSession]) -> list[str]:
    rows = (await db_session.execute(...)).scalars().all()
    return [row.name for row in rows]


app = Litestar(route_handlers=[list_items], plugins=[plugin])

Every query issued through db_session is transparently filtered to the resolved tenant, and every insert/update is force-pinned to it - a handler cannot accidentally (or maliciously, via a crafted request body) read or write another tenant's rows through the ORM.

Usage

Postgres RLS mode

Mark models with @with_rls, provision the RLS function/policies once with setup_rls_schema, and wire an RLSSessionProvider:

from litestar import Litestar, get
from litestar.di import NamedDependency
from sqlalchemy import Mapped, mapped_column
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase

from litestar_tenants import HeaderTenantResolver, TenantsPlugin
from litestar_tenants.rls import PostgresManager, RLSSessionProvider, setup_rls_schema, with_rls


class Base(DeclarativeBase):
    pass


@with_rls
class Item(Base):
    __tablename__ = "item"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    tenant: Mapped[str]


engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/app")

async def on_startup() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
        await setup_rls_schema(conn, Base.metadata)  # idempotent, safe to call every startup

manager = PostgresManager.from_engine(engine, schema_name="public")

plugin = TenantsPlugin(
    provider=RLSSessionProvider(
        manager=manager,
        resolver=HeaderTenantResolver(),
        create_if_missing=False,  # see "Tenant provisioning" below
    )
)


@get("/items")
async def list_items(db_session: NamedDependency[AsyncSession]) -> list[str]:
    rows = (await db_session.execute(...)).scalars().all()
    return [row.name for row in rows]


app = Litestar(route_handlers=[list_items], plugins=[plugin])

Provision and remove tenants directly through the manager:

await manager.create_tenant("acme")
await manager.list_tenants()    # {"acme", ...}
await manager.delete_tenant("acme")

Tenant provisioning

create_if_missing defaults to False on RLSSessionProvider, overriding PostgresManager's own default of True. Auto-creating a Postgres role from a request-resolved tenant identifier (a header, a JWT claim) means an attacker-controlled value could trigger role creation. Provision tenants explicitly via manager.create_tenant() and only opt into create_if_missing=True if that risk is acceptable for your resolver - e.g. one backed by a trusted, pre-validated tenant list.

Tenant resolvers

A resolver extracts a tenant identifier from the incoming connection; both isolation modes take one via their provider. Three are built in:

Implement TenantResolver yourself for anything else - it's a one-method Protocol:

class TenantResolver(Protocol):
    def resolve(self, connection: ASGIConnection[Any, Any, Any, Any]) -> TenantIdentifier: ...

A resolver must raise TenantNotResolvedError rather than returning a fallback value - both providers fail closed on an unresolved tenant instead of falling back to an unscoped session.

Configuration

TenantsPlugin(provider=..., dependency_key="db_session")

dependency_key controls the name the session is injected under (default db_session, matching the examples above).

Design

rls and orm-scoped are not interchangeable. One is a database guarantee, the other an application-level one - see the intro paragraph above. Choosing between them is a threat-model decision this library deliberately doesn't make for you.

orm-scoped force-pins writes, not just filters reads. SQLAlchemy's with_loader_criteria only restricts which rows a SELECT/bulk UPDATE matches, and has no effect on INSERT at all. Without also force-pinning the tenant column on every UPDATE/INSERT, a caller could still write an arbitrary tenant value through the ORM.

Known gaps in that force-pinning: multi-row bulk inserts (insert(Model).values([{...}, {...}])) and executemany-style calls (session.execute(stmt, [{...}, {...}])) bypass per-statement rewriting entirely and can still write an arbitrary tenant. If your workload relies on either, prefer rls mode or add your own guard around those call sites.

create_if_missing defaults to False for rls, overriding PostgresManager's own default of True - see Tenant provisioning.

Resolvers fail closed. TenantNotResolvedError beats a fallback to an unscoped session; an unresolved tenant should be a 4xx, never a query that silently sees everything.

Development

uv sync

uv run ruff check
uv run ruff format --check
uv run pyrefly check

uv run pytest                 # unit tests, 100% coverage required, no external services
uv run pytest -m integration  # real-Postgres tests for `rls` mode, needs Docker (via pytest-databases)

The unit suite runs against SQLite/mocked sessions and covers orm-scoped fully plus all of rls's Python-level logic. The integration suite is what actually proves Postgres enforces rls isolation at the database level - reading/writing through raw SQL, not just through the ORM.

Acknowledgments

The rls isolation mode (src/litestar_tenants/rls/) is adapted from sqlalchemy-tenants by Michele Zanotti, licensed under the MIT License. Adapted, not vendored unmodified.

License

litestar-tenants is licensed under BSD-3-Clause - see LICENSE.

The rls module additionally carries the following notice, as required by the MIT License of the code it was adapted from (see Acknowledgments above):

MIT License

Copyright (c) 2025 Michele Zanotti

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.