FastAPI + PostgreSQL RLS: Multi-Tenant do Jeito Certo

Quando comecei a construir o LucraHub — uma plataforma de gestão para vendedores da Amazon Brasil — a primeira decisão arquitetural que travou minha cabeça foi: como isolar dados entre sellers sem duplicar infra?

O cenário clássico do SaaS multi-tenant: dezenas (ou centenas) de empresas usando o mesmo sistema, mas cada uma vendo apenas os próprios dados. Existem basicamente três abordagens:

Estratégia	Isolamento	Custo de infra	Complexidade
Database por tenant	Total	Alto	Alta operacional
Schema por tenant	Alto	Médio	Média
Shared table + RLS ✅	Alto	Baixo	Baixa (após setup)

Escolhemos Row Level Security (RLS) no PostgreSQL com tabelas compartilhadas. Neste post vou mostrar exatamente como implementamos isso no LucraHub do zero, com código real.

Contexto: O LucraHub tem uma arquitetura de microsserviços — cada serviço tem seu próprio banco PostgreSQL. O RLS é aplicado por serviço individualmente, não existe um “banco central de tenants”. Isso escala bem e evita SPOF.

O que é Row Level Security?

Row Level Security é um recurso nativo do PostgreSQL que permite definir políticas de acesso por linha diretamente no banco. Diferente de filtros na aplicação, o banco recusa fisicamente retornar linhas que não pertencem ao tenant ativo — mesmo que o desenvolvedor esqueça de filtrar.

A ideia central é simples: cada linha tem um tenant_id, e o PostgreSQL compara esse campo com uma variável de sessão que configuramos antes de cada query. Sem match — a linha é invisível.

Setup inicial do banco

1. Estrutura da tabela

Todo modelo do LucraHub herda de um mixin com tenant_id obrigatório:

-- Exemplo: tabela de produtos
CREATE TABLE products (
    id          UUID        PRIMARY KEY DEFAULT gen_random_uuid(),
    tenant_id   UUID        NOT NULL,
    asin        TEXT        NOT NULL,
    title       TEXT,
    price       NUMERIC(10, 2),
    created_at  TIMESTAMPTZ DEFAULT now()
);

-- Índice composto: tenant + queries mais comuns
CREATE INDEX idx_products_tenant
    ON products (tenant_id, created_at DESC);

2. Ativando RLS e criando a política

-- Ativar RLS na tabela
ALTER TABLE products ENABLE ROW LEVEL SECURITY;

-- Política de acesso: só vê linhas do próprio tenant
CREATE POLICY tenant_isolation ON products
    USING (
        tenant_id = current_setting('app.current_tenant')::UUID
    );

-- Política de INSERT: garante que novos registros pertencem ao tenant certo
CREATE POLICY tenant_insert ON products
    FOR INSERT WITH CHECK (
        tenant_id = current_setting('app.current_tenant')::UUID
    );

-- Usuário da aplicação NÃO pode bypassar RLS
-- (só superuser/BYPASSRLS consegue)
ALTER TABLE products FORCE ROW LEVEL SECURITY;

Detalhe importante: O FORCE ROW LEVEL SECURITY é essencial. Sem ele, o dono da tabela (geralmente o mesmo role da aplicação) consegue bypassar as políticas. Sempre use.

Integrando com FastAPI

O fluxo é: request chega → middleware extrai o tenant do JWT → antes de cada query, injetamos SET LOCAL app.current_tenant = '...' na sessão PostgreSQL. O RLS faz o resto.

Request (JWT) → Auth Middleware → get_db() → SET LOCAL app.current_tenant → Query (RLS filtra)

Modelo SQLAlchemy com mixin de tenant

import uuid
from sqlalchemy import Column, String
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


class TenantMixin:
    """Mixin aplicado em todos os models do LucraHub."""

    tenant_id: UUID = Column(
        UUID(as_uuid=True),
        nullable=False,
        index=True,
        # não tem default: o RLS garante via SET LOCAL
    )


class Product(TenantMixin, Base):
    __tablename__ = "products"

    id       = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    asin     = Column(String, nullable=False)
    title    = Column(String)
    price    = Column(Numeric(10, 2))

Dependency: injetando o tenant na sessão

from contextlib import asynccontextmanager
from typing import AsyncGenerator
from uuid import UUID

from sqlalchemy.ext.asyncio import (
    AsyncSession, async_sessionmaker, create_async_engine
)
from sqlalchemy import text

from app.config import settings

engine = create_async_engine(
    settings.DATABASE_URL,
    pool_size=10,
    max_overflow=5,
    pool_pre_ping=True,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    expire_on_commit=False,
    class_=AsyncSession,
)


async def get_db(
    tenant_id: UUID,           # vem do JWT via Depends()
) -> AsyncGenerator[AsyncSession, None]:
    """
    Dependency que entrega uma sessão já configurada com
    o tenant correto. O RLS do Postgres cuida do resto.
    """
    async with AsyncSessionLocal() as session:
        # SET LOCAL: válido apenas nesta transação
        await session.execute(
            text("SET LOCAL app.current_tenant = :tid"),
            {"tid": str(tenant_id)},
        )
        yield session

Por que SET LOCAL e não SET? SET LOCAL tem escopo de transação. Quando a transação termina, a variável some. Com connection pooling isso é crítico — garante que um tenant nunca “vaza” para a próxima query de outro usuário que reutilizar a mesma conexão.

Extraindo o tenant do JWT

from uuid import UUID
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
import jwt

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")


async def get_current_tenant(
    token: str = Depends(oauth2_scheme),
) -> UUID:
    try:
        payload = jwt.decode(
            token,
            settings.JWT_SECRET,
            algorithms=["HS256"],
        )
        tenant_id = payload.get("tenant_id")
        if not tenant_id:
            raise ValueError
        return UUID(tenant_id)

    except (jwt.InvalidTokenError, ValueError):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Token inválido ou expirado",
        )


# Dependency combinada: tenant_id + sessão configurada
async def get_tenant_db(
    tenant_id: UUID = Depends(get_current_tenant),
) -> AsyncGenerator[AsyncSession, None]:
    async for db in get_db(tenant_id):
        yield db

Usando nos endpoints

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select

from app.auth.dependencies import get_tenant_db
from app.models import Product

router = APIRouter(prefix="/products")


@router.get("/")
async def list_products(
    db: AsyncSession = Depends(get_tenant_db),
):
    # Zero filtro manual: o RLS garante isolamento
    result = await db.execute(select(Product))
    return result.scalars().all()


@router.post("/")
async def create_product(
    payload: ProductCreate,
    db: AsyncSession = Depends(get_tenant_db),
    tenant_id: UUID = Depends(get_current_tenant),
):
    product = Product(
        **payload.model_dump(),
        tenant_id=tenant_id,  # explícito no insert
    )
    db.add(product)
    await db.commit()
    return product

---

Pontos de atenção que nos pegaram

Background tasks e workers

Celery, ARQ, jobs assíncronos — qualquer processamento fora do request/response cycle não tem JWT. Nesses casos, o tenant_id precisa vir do payload da task, e você injeta manualmente com SET LOCAL antes das queries. Nunca delegue isso ao RLS “automaticamente” em workers.

Migrations com Alembic

Alembic usa um role diferente (geralmente superuser ou BYPASSRLS). Isso é bom para rodar migrations, mas significa que seu script de seed/fixtures pode silenciosamente ignorar o RLS se você usar o mesmo connection string. Use roles separados por responsabilidade.

Queries administrativas

Operações internas — como relatórios consolidados, billing, suporte — precisam ver dados de todos os tenants. Para isso, crie um role separado com BYPASSRLS e nunca exponha esse role na aplicação principal.

-- Role da aplicação: sujeito ao RLS
CREATE ROLE lucrahub_app LOGIN;
GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO lucrahub_app;

-- Role administrativo: bypass RLS
CREATE ROLE lucrahub_admin LOGIN BYPASSRLS;
GRANT ALL ON ALL TABLES IN SCHEMA public TO lucrahub_admin;

Testando o isolamento

Tão importante quanto implementar é ter testes que provam que o isolamento funciona. No LucraHub temos uma suite específica pra isso:

import pytest
from uuid import uuid4


@pytest.mark.asyncio
async def test_tenant_cannot_see_other_tenant_data(db_factory):
    tenant_a = uuid4()
    tenant_b = uuid4()

    # Cria produto no tenant A
    async with db_factory(tenant_a) as db:
        product = Product(
            asin="B08XYZ",
            title="Produto do Tenant A",
            tenant_id=tenant_a,
        )
        db.add(product)
        await db.commit()

    # Tenant B não deve ver o produto de A
    async with db_factory(tenant_b) as db:
        result = await db.execute(select(Product))
        products = result.scalars().all()

    assert len(products) == 0, "Tenant B não deveria ver dados do Tenant A!"

Performance: o RLS cobra algum custo?

A resposta curta: desprezível, se os índices estiverem certos.

O PostgreSQL avalia a política RLS como um predicado adicional na query — é equivalente a você mesmo adicionar WHERE tenant_id = $1. Com índice em (tenant_id, campo_de_ordenação), o planner usa index scan normalmente.

No LucraHub, após adicionar os índices compostos corretos, o overhead observado nas queries mais frequentes foi menor que 2ms comparado ao cenário sem RLS.

Conclusão

Row Level Security é uma das features mais subestimadas do PostgreSQL. Quando bem implementada, ela transforma o banco em um guardião ativo dos dados — eliminando uma classe inteira de bugs onde um dev esquece de filtrar por tenant.

No LucraHub, o combo FastAPI + RLS nos deu:

• ✅ Isolamento garantido pelo banco, não pela aplicação
• ✅ Código de endpoint limpo, sem WHERE tenant_id = espalhado por todo lado
• ✅ Performance equivalente a queries normais com índices corretos
• ✅ Separação clara de responsabilidades: autenticação no FastAPI, autorização de dados no Postgres

O custo de setup é real — migrations, roles, testes de isolamento — mas paga dividendos conforme o produto cresce. Vale muito.