fastapi-rls 0.1.0

PostgreSQL Row Level Security (RLS) for FastAPI and SQLAlchemy applications

FastAPI RLS

PostgreSQL Row Level Security (RLS) for FastAPI and SQLAlchemy applications — enforce multi-tenant and per-user data isolation at the database level, where it cannot be bypassed by a forgotten WHERE clause.

This is the FastAPI/SQLAlchemy sibling of django-rls. The policy model, the session-variable context, and the generated DDL are intentionally the same; only the framework integration differs.

📖 Documentation: https://fastapi-rls.com (source in website/)

Features

• 🔒 Database-level Row Level Security using native PostgreSQL RLS
• 🏢 Tenant-based and user-based policies, plus raw CustomPolicy
• 🧩 ORM-agnostic core — policies, context, and DDL have zero framework dependencies; SQLAlchemy, FastAPI, and Alembic are optional adapters
• ⚡ Sync and async — works with Session and AsyncSession
• 🛡️ Leak-proof by construction — identity is applied with SET LOCAL inside a per-request transaction, with a pool-checkout scrub as a defense-in-depth backstop
• 🔑 Immutable identity — user_id/tenant_id cannot be silently overridden mid-request
• 🚀 FastAPI dependency + optional identity middleware
• 🧰 Alembic operation directives and a standalone fastapi-rls CLI
• 🧪 Extensively tested against real PostgreSQL, including cross-tenant leak tests

Installation

pip install "fastapi-rls[all]"          # everything
pip install "fastapi-rls[fastapi]"      # FastAPI + SQLAlchemy
pip install "fastapi-rls[sqlalchemy]"   # SQLAlchemy only
pip install "fastapi-rls[alembic,cli]"  # migrations + CLI

Requirements: Python 3.10–3.13, PostgreSQL 13–17, SQLAlchemy 2.0, FastAPI ≥0.100. Every one of those versions is exercised in CI (see the test matrix).

Quick start

1. Declare policies on your models

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"),
    ]
    id: Mapped[int] = mapped_column(primary_key=True)
    tenant_id: Mapped[int] = mapped_column(index=True)
    title: Mapped[str]

RLSMixin binds each policy to the table and registers it. (Call collect_policies(Base) at startup if your declarative base doesn't propagate __init_subclass__.)

2. Apply the policies to the database

from fastapi_rls import RLS

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

or from the CLI (no Alembic required):

fastapi-rls sync --url "$DATABASE_URL" --policies myapp.models

or 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

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

rls = RLS(engine=engine)

def identity(user = Depends(get_current_user)) -> dict:
    # fastapi-rls never authenticates — you resolve the principal, it propagates it.
    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)):
    return session.scalars(select(Document)).all()   # PostgreSQL filters by tenant

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

Async is identical

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()

How it works

1. A policy compiles to a PostgreSQL predicate like tenant_id = (SELECT NULLIF(current_setting('rls.tenant_id', true), '')::integer) (the current_setting read is wrapped in a scalar subquery so the planner evaluates it once per statement — the documented RLS performance pattern).
2. The session dependency opens a transaction and runs SET LOCAL rls.tenant_id = … from the identity you resolved.
3. PostgreSQL enforces the predicate on every SELECT/INSERT/UPDATE/DELETE.
4. Committing the transaction clears the LOCAL setting; a pool-checkout scrub is a second line of defense. Identity can never survive into another request.

Policy types

Policy	Use
TenantPolicy(name, column="tenant_id")	Row visible when the tenant column matches rls.tenant_id
UserPolicy(name, column="owner_id")	Row visible when the user column matches rls.user_id
CustomPolicy(name, using=..., check=...)	A raw predicate (injection-screened) when the above don't fit

All support operation (ALL/SELECT/INSERT/UPDATE/DELETE), permissive=False for RESTRICTIVE policies, roles=, and context_type (integer/bigint/uuid/text).

Configuration

Pass an RLSConfig to RLS(config=...) (or configure(...) the global one):

Option	Default	Meaning
require_context	False	Raise if no identity is set before a query
transaction_per_request	True	Use SET LOCAL inside a per-request transaction
reset_context_on_checkout	True	Install the pool-checkout scrub backstop
audit_log	False	Emit structured logs on context set/clear
default_roles	"public"	Default TO clause for policies
key_prefix	"rls"	Session-variable namespace
registered_context_keys	()	Extra keys to scrub during connection hygiene

CLI

fastapi-rls sync   --url "$DATABASE_URL" --policies myapp.models   # reconcile
fastapi-rls plan   --url "$DATABASE_URL" --policies myapp.models   # dry run
fastapi-rls audit  --url "$DATABASE_URL" --policies myapp.models   # report drift
fastapi-rls enable --url "$DATABASE_URL" documents                 # enable + force

Security notes

• Connect as a non-superuser. PostgreSQL superusers and BYPASSRLS roles ignore RLS entirely. Your application role must be neither.
• FORCE ROW LEVEL SECURITY is applied by default so the table owner is constrained too.
• Table, column, role, and operation names are validated against a strict identifier allowlist before reaching any DDL; CustomPolicy expressions are screened for statement terminators, comments, and DML/DDL keywords.

License

BSD 3-Clause License — see LICENSE.