Skip to main content

Complete Guide: Multi-Tenant Flask with PostgreSQL RLS

This is the full walkthrough — from an empty project to a Flask application whose tenant isolation is enforced by PostgreSQL and proven by tests. The Quick Start compresses this into four steps; this guide explains each decision along the way.

What you'll build: a minimal invoicing API where every tenant sees only its own invoices, enforced by Row-Level Security — meaning even a raw SELECT * FROM invoices returns only the current tenant's rows.

1. Project setup​

mkdir rls-demo && cd rls-demo
python -m venv .venv && source .venv/bin/activate
pip install flask flask-sqlalchemy "flask-rls[alembic]" psycopg alembic

You need a running PostgreSQL 12+ (any recent version; the library is tested against 12 through 17). For local work:

docker run -d --name rls-pg -e POSTGRES_PASSWORD=dev -p 5432:5432 postgres:16-alpine

2. Database roles: the step everyone skips​

RLS has one rule that surprises people: a table's owner bypasses its policies unless the table is marked FORCE ROW LEVEL SECURITY — and superusers bypass RLS always. If your app connects as the same role that ran CREATE TABLE, your policies can silently do nothing.

So use two roles: one that owns the schema (runs migrations), one that the application connects as:

-- as postgres/superuser:
CREATE ROLE migrator LOGIN PASSWORD 'migrate-secret';
CREATE ROLE app_user LOGIN PASSWORD 'app-secret';
CREATE DATABASE invoicing OWNER migrator;

\c invoicing
GRANT USAGE ON SCHEMA public TO app_user;
ALTER DEFAULT PRIVILEGES FOR ROLE migrator IN SCHEMA public
  GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO app_user;

Flask RLS emits FORCE ROW LEVEL SECURITY for every registered table as a second line of defense, but the dedicated non-owner app_user is what makes the security model clean. The security model page covers this in depth.

3. The application​

# app.py
from flask import Flask, g, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from flask_rls import RLS

db = SQLAlchemy()
rls = RLS()

def create_app():
    app = Flask(__name__)
    app.config["SQLALCHEMY_DATABASE_URI"] = (
        "postgresql+psycopg://app_user:app-secret@localhost/invoicing"
    )
    db.init_app(app)

    with app.app_context():
        rls.init_app(app, engine=db.engine)

    @app.before_request
    def set_rls_context():
        # Resolve the tenant however your app does: subdomain, JWT claim,
        # API-key lookup. For the demo, a header.
        g.tenant_id = request.headers.get("X-Tenant-ID")

    @app.get("/invoices")
    def list_invoices():
        invoices = Invoice.query.all()   # no tenant filter — RLS supplies it
        return jsonify([{"id": i.id, "amount": i.amount} for i in invoices])

    return app

class Invoice(db.Model):
    __tablename__ = "invoices"
    id = db.Column(db.Integer, primary_key=True)
    tenant_id = db.Column(db.Text, nullable=False, index=True)
    amount = db.Column(db.Numeric, nullable=False)

Two things to notice:

  • The route handler has no tenant filter. That's the point — after this guide, the database adds it for you.
  • tenant_id is indexed. The RLS predicate is tenant_id = <current tenant> on every query, so this index is what keeps policies cheap.

On each transaction begin, Flask RLS reads g.tenant_id and runs SELECT set_config('rls.tenant_id', :value, true). The true makes it transaction-scoped — PostgreSQL discards it on COMMIT/ROLLBACK, which is why pooled connections can't leak context between requests (details).

4. Policies as a migration​

Initialize Alembic and register the RLS operations in migrations/env.py:

# migrations/env.py — add near the top
from flask_rls.alembic import *  # noqa: F401,F403  registers op.enable_rls, ...

Write the migration (run migrations as migrator, the owner role):

# migrations/versions/xxxx_enable_rls_on_invoices.py
from alembic import op
from flask_rls import TenantPolicy

def upgrade():
    op.enable_rls("invoices")
    op.force_rls("invoices")
    op.create_policy("invoices", TenantPolicy("tenant_isolation", "tenant_id"))

def downgrade():
    op.drop_policy(
        "invoices", "tenant_isolation",
        policy=TenantPolicy("tenant_isolation", "tenant_id"),
    )
    op.no_force_rls("invoices")
    op.disable_rls("invoices")

Prefer to see the SQL before trusting a library with your security? Register the policy on the extension and dump it:

flask rls sql --table invoices
ALTER TABLE "invoices" ENABLE ROW LEVEL SECURITY;
ALTER TABLE "invoices" FORCE ROW LEVEL SECURITY;
CREATE POLICY "tenant_isolation" ON "invoices" AS PERMISSIVE FOR ALL TO PUBLIC
  USING (tenant_id = (SELECT NULLIF(current_setting('rls.tenant_id', true), '')::text))
  WITH CHECK (tenant_id = (SELECT NULLIF(current_setting('rls.tenant_id', true), '')::text));

The predicate wraps current_setting in a scalar subquery so PostgreSQL evaluates it once per statement instead of once per row — the InitPlan pattern.

5. See it work​

Seed two tenants and query as each:

curl -s -X GET localhost:5000/invoices -H "X-Tenant-ID: acme"   # only acme rows
curl -s -X GET localhost:5000/invoices -H "X-Tenant-ID: globex" # only globex rows
curl -s -X GET localhost:5000/invoices                          # [] — fail closed

The third call is the important one: no tenant header, zero rows. Missing context in Flask RLS fails closed — the GUC is NULL, the predicate matches nothing. During development you can make this loud instead of silent:

app.config["RLS_REQUIRE_CONTEXT"] = True   # missing context raises RLSContextRequiredError

6. Prove it with tests​

Trust nothing you haven't asserted. The essential suite (full guidance):

from sqlalchemy import text

def test_isolation(engine, rls):
    with rls.override(tenant_id="acme"):
        with engine.connect() as conn:
            rows = conn.execute(text("SELECT tenant_id FROM invoices")).all()
    assert {r.tenant_id for r in rows} == {"acme"}

def test_fail_closed(engine, rls):
    with rls.bypass():   # emits no context
        with engine.connect() as conn:
            rows = conn.execute(text("SELECT * FROM invoices")).all()
    assert rows == []

def test_with_check_blocks_cross_tenant_writes(engine, rls):
    with rls.override(tenant_id="acme"):
        with engine.begin() as conn, pytest.raises(Exception):
            conn.execute(text(
                "INSERT INTO invoices (tenant_id, amount) VALUES ('globex', 10)"
            ))

Note the raw SQL: these tests deliberately bypass the ORM to prove the database enforces isolation. Run them as app_user, not the owner — the Flask RLS test suite itself does this against real PostgreSQL via Testcontainers.

7. Background jobs, CLI commands, admin​

Outside a request there's no g, so context fails closed. Be explicit:

with rls.override(tenant_id="acme"):      # act as one tenant
    generate_monthly_statement()

with rls.bypass():                        # no context — non-RLS tables only
    create_tenant_record()

bypass() is not god-mode: on RLS-protected tables it sees zero rows. True cross-tenant access (admin dashboards, analytics) should go through a separate engine bound to a PostgreSQL role with BYPASSRLS — a deliberate, auditable choice rather than a default.

Common pitfalls​

SymptomCauseFix
Policies exist but every row is visibleApp connects as table owner without FORCEConnect as non-owner role; Flask RLS emits FORCE — verify with \d invoices
Every query returns zero rowsContext never setSet g.tenant_id in before_request; use RLS_REQUIRE_CONTEXT to make it loud
Slow queries after enabling RLSMissing index on tenant_idIndex every policy column
Works in tests, empty in productionTests ran as superuserRun tests as the app's non-owner role

Where to next​