Isaev1980 42 minutos atrás Três soluções de arquitetura para SaaS B2B multi-tenant que eu gostaria de ter conhecido antes Médio 10 min 1.6K Python * PostgreSQL * Segurança da Informação * Programação * Caso Da caixa de areia O erro mais caro do meu SaaS B2B teve exatamente uma linha
python# app/config.py TENANT_ID = "tenant-1"
Quando eu tinha um tenant, tudo funcionava corretamente. No segundo — metade das entidades de administração (médicos, serviços, tabelas de preços) começou a desaparecer da interface do cliente. Não "não salvar" — mas aparecer no banco de dados com outro tenant_id . Eu fiquei olhando para essa mística por um dia e meio, antes de perceber: 30 endpoints pegam tenant_id da closure de config , e não de user.tenant_id . Óbvio em retrospectiva. Completamente invisível durante o primeiro piloto. Esta análise é sobre três soluções de arquitetura para SaaS multi-tenant em uma indústria regulamentada, que são tomadas na primeira semana e, em seguida, por anos, economizam meses de refatoração ou acumulam silenciosamente dívida técnica. Stack: Python 3.11 / FastAPI / SQLAlchemy 2.x / PostgreSQL 16. Contexto: SaaS B2B para clínicas médicas privadas, 152-FZ, agendamento em tempo real. TL;DR
- Multi-tenancy via tenant_id column + helper scoped_select do primeiro commit — mesmo que haja apenas um tenant. A razão está dentro.
- Proteção contra double-booking via PostgreSQL EXCLUDE USING gist + tsrange — números reais: 12 solicitações concorrentes por slot → 1×200, 11×409, sem application-level locks.
- 152-FZ — são campos e helpers : data_category enum em audit_logs com autoclassificador, migração transparente de PBKDF2 para argon2id sem forced reset, patient_consents com versionamento scoped.
- Um bug com TENANT_ID hardcoded mostra por que todas as três soluções precisam ser implementadas imediatamente.
- Operational tooling : verificação de integridade somente leitura em sete categorias de anomalias + detecção de schema drift via alembic check em um PG limpo. Solução nº 1: multi-tenancy via tenant_id column Três abordagens clássicas para multi-tenancy em um banco de dados relacional: Estratégia Isolamento Backup/migrações Custo Quando se encaixa Schema-per-tenant No nível do Postgres Complexo (N esquemas × M migrações) Médio 5–50 tenants empresariais com diferentes SLA Database-per-tenant Ferro Muito complexo Alto Compliance-driven, ≤10 tenants Shared DB + tenant_id No nível do aplicativo Simples (um esquema) Baixo SMB SaaS, centenas–milhares de tenants Eu escolhi a terceira opção. Cada tabela de negócios contém tenant_id TEXT NOT NULL com um índice, todas as consultas devem filtrar por ele. O preço dessa simplicidade — um único erro no filtro se transforma em um IDOR em dados confidenciais entre tenants. Para não esquecer disso, o projeto tem um helper sutil:
python# app/tenant_scope.py from sqlalchemy import select from sqlalchemy.sql import Select def scoped_select(model, user) -> Select: """SELECT, automaticamente limitado ao tenant atual. Lança PermissionError se o usuário não tiver tenant_id — isso é intencional: é melhor um 500 nos logs do que um vazamento silencioso. """ if not user.tenant_id: raise PermissionError("User has no tenant_id; refusing to query") return select(model).where(model.tenant_id == user.tenant_id) def scoped_get(db, model, id_, user): return db.scalar( select(model).where(model.id == id_, model.tenant_id == user.tenant_id) )
O princípio — é melhor um erro explícito do que um vazamento silencioso. E na revisão de código, a regra é simples: qualquer select(BusinessModel) sem scoped_select — uma observação de bloqueio. Bug story em callout: uma linha, dois dias de debug Na primeira versão, eu usei o closure tenant_id do config em todos os write-handlers:
python> @router.post("/v1/doctors") > def create_doctor(payload, db, user): > item = Doctor( > id=f"doctor-{uuid4()}", > tenant_id=tenant_id, # ← closure do config! > **payload.model_dump(), > ) > ``` O read-path usou scoped_select(Doctor, user) — corretamente. O write-path bugou config.TENANT_ID . Enquanto houver apenas um tenant — a dessincronização é invisível. Quando conectamos a segunda clínica, seu administrador viu a lista de médicos vazia, embora tivesse acabado de adicionar três. Esses três estavam no banco de dados com tenant_id="tenant-1" . Fix: tenant_id=user.tenant_id e regressão em E2E “registrar tenant B → criar entidade como user-B → logar como user-A → certificar-se de que não é visto”. Lição: o helper é necessário desde o primeiro commit, mesmo que haja apenas um tenant. No décimo tenant, é tarde demais — você terá que revisar cada endpoint. E por que não PostgreSQL Row-Level Security? Esta é a primeira pergunta que me fazem, e ela é justa. RLS daria isolamento de tenant no nível do banco de dados, e não do aplicativo — teoricamente mais confiável. Na prática, RLS requer SET app.tenant_id = '...' no início de cada transação, quebra o stateless connection pooling (PgBouncer em modo de transação se torna mais difícil de configurar) e não funciona bem com migrações e tarefas de administração, onde você quer contornar a restrição legalmente. Para um projeto solo em estágio inicial, a sobrecarga de RLS não é justificada — scoped_select + regra de lint na revisão de código dá 95% de proteção com 5% de complexidade. Em um estágio posterior (dezenas de tenants empresariais com requisitos de conformidade), RLS se torna um complemento razoável, não uma substituição. Solução nº 2: PostgreSQL EXCLUDE USING gist contra double-booking Dois administradores reservam simultaneamente o mesmo slot com o mesmo médico. Ambos veem o slot livre. Ambos chamam POST /appointments . Ambos recebem 200 OK . Dupla gravação. A proteção “no código” ( SELECT para verificação → ) sob carga paralela não funciona. Entre duas solicitações na mesma transação, outra transação consegue inserir um slot sobreposto. Eu vi isso no primeiro teste de carga. A evolução da proteção no projeto ocorreu em três estágios. Apresento números reais de benchmarks no PostgreSQL 16, single-node: Estágio 1 — application-level check. 12 solicitações concorrentes por slot → 4×200, 8×409 . Gravações duplas visíveis. create_unique_appointments p95 = 26ms. Estágio 2 — transactional advisory locks ( pg_advisory_xact_lock por (doctor_id, time_window) antes de INSERT). 12 concorrentes → 1×200, 11×409 . Race-condition fechado, mas create_unique_appointments p95 subiu para 36.86ms devido à sobrecarga de serialização. Estágio 3 — hard DB invariant via EXCLUDE USING gist . As mesmas 12 concorrentes → 1×200, 11×409 , create_unique_appointments p95 = 31.69ms , conflict-path p95 = 11.22ms . A proteção foi movida do aplicativo para o esquema, o advisory lock não é mais necessário. Restrição DDL: ``sql -- Alembic upgrade CREATE EXTENSION IF NOT EXISTS btree_gist; ALTER TABLE appointments ADD CONSTRAINT appointments_doctor_slot_excl EXCLUDE USING gist ( doctor_id WITH =, tsrange(starts_at, ends_at, '[)') WITH && ) WHERE (status NOT IN ('cancelled', 'no_show'));
O que está acontecendo aqui:
- doctor_id WITH = — dois intervalos são considerados conflitantes apenas se tiverem o mesmo doctor_id .
- tsrange(starts_at, ends_at, '[)') — intervalo semiaberto; a entrada 10:00–10:30 e 10:30–11:00 não se sobrepõem.
- && — operador de interseção de intervalos.
- WHERE (status NOT IN ('cancelled', 'no_show')) — cancelados e não comparecimentos são excluídos, o paciente após o cancelamento pode se reinscrever no mesmo slot. btree_gist é necessário porque = para escalar doctor_id via o operador gist padrão não funciona; a extensão adiciona semântica btree dentro do índice gist. Mapeamento de erros em HTTP no FastAPI: ``python from psycopg.errors import ExclusionViolation
try: db.add(appointment) db.flush() except IntegrityError as exc: if isinstance(exc.orig, ExclusionViolation): raise HTTPException( status_code=409, detail={ "code": "APPOINTMENT_SLOT_CONFLICT", "message": "O slot está ocupado por outra entrada", }, ) raise
tsrange
ou
tstzrange
?
Eu escolhi
tsrange
(timestamp ingênuo sem fuso horário), porque neste projeto todos os timestamps são armazenados em UTC, e a conversão para o horário local da clínica é feita no nível do aplicativo via
tenant.timezone
.
tstzrange
funcionaria, mas adicionaria uma camada extra de conversões implícitas e possíveis bugs com DST. Se você tiver a política de armazenamento “timestamp with timezone em todos os lugares” — pegue
tstzrange
, a diferença é apenas em como o Postgres internamente trata os intervalos.
Armadilhas:
- EXCLUDE
cria um índice gist, que cresce mais rápido que B-tree em grandes volumes. De acordo com o [artigo no Habr sobre btree_gist benchmark](https://habr.com/ru/articles/820455/) — degradação de gravação para 2× e leitura para 20% em tabelas quentes. Para uma tabela de registros de pacientes, isso é ok (cardinalidade baixa), para uma tabela de eventos com milhões de linhas — pense nisso.
- WHERE
a cláusula partial constraint funciona apenas em / DELETE. O cancelamento deve ser
UPDATE status='cancelled'
, não
DELETE
— caso contrário, a entrada “desaparece”, e o slot em um segundo é ocupado por outra pessoa, e a entrada cancelada não pode mais ser restaurada.
- Nos testes, você precisa do mesmo dialeto (PG, não SQLite). SQLite aceita DDL sem erros, mas não cria nenhuma restrição de exclusão — o bug nos testes não pode ser rastreado.
O que eu consegui no final: com o crescimento de 1 para 10 clínicas na produção, nenhuma gravação dupla. Não precisa de Redis, não precisa de distributed lock, não precisa de application-level retry. PostgreSQL faz todo o trabalho.
Solução nº 3: 152-FZ como código
Para dados médicos na Federação Russa, 152-FZ exige: armazenamento em data centers russos, proteção de dados durante a transmissão e armazenamento, log de acesso a PII, consentimento separado para o processamento de categorias especiais, a capacidade de excluir/anonimizar mediante solicitação. O que disso se transforma em código:
Criptografia de senha: migração PBKDF2 → argon2id sem forced reset
Inicialmente, as senhas eram hashadas via
PBKDF2-HMAC-SHA256
com 120.000 iterações — o padrão OWASP até 2023. A recomendação moderna OWASP/NIST — argon2id (memory-hard, mais resistente a ataques de GPU). Uma migração acentuada via forced password reset é inaceitável em B2B (dezenas de funcionários da clínica, o registrador não poderá logar na manhã de segunda-feira), portanto, um esquema transparente funciona: novas senhas são escritas como argon2id, as antigas são verificadas em seu formato, com login bem-sucedido, são rehashadas silenciosamente em argon2id.
A chave — detecção de formato por prefixo de hash:
```python
import hashlib
import secrets
from argon2 import PasswordHasher
from argon2 import exceptions as argon2_exceptions
PASSWORD_ITERATIONS = 120_000 # legacy PBKDF2 para backward-compat
PASSWORD_PREFIX = "pbkdf2_sha256"
ARGON2_PREFIX = "$argon2"
PASSWORD_HASHER = PasswordHasher() # argon2id com parâmetros padrão
def hash_password(password: str) -> str:
"""Todas as novas senhas — argon2id."""
return PASSWORD_HASHER.hash(password)
def verify_password(password: str, encoded: str) -> bool:
if encoded.startswith(ARGON2_PREFIX):
try:
return PASSWORD_HASHER.verify(encoded, password)
except (argon2_exceptions.VerifyMismatchError,
argon2_exceptions.InvalidHashError):
return False
if encoded.startswith(f"{PASSWORD_PREFIX}$"):
_, iterations, salt, digest = encoded.split("$", 3)
computed = hashlib.pbkdf2_hmac(
"sha256", password.encode(), salt.encode(), int(iterations)
)
return secrets.compare_digest(computed.hex(), digest)
return False
def password_needs_rehash(encoded: str) -> bool:
if encoded.startswith(f"{PASSWORD_PREFIX}$"):
return True # qualquer PBKDF2 → upgrade em argon2id
if encoded.startswith(ARGON2_PREFIX):
return PASSWORD_HASHER.check_needs_rehash(encoded)
return False
Em /auth/login após a verificação bem-sucedida:
pythonif password_needs_rehash(user.password_hash): user.password_hash = hash_password(plain_password) db.commit()
Parâmetros argon2id e trade-off. Eu uso os padrões argon2-cffi : memory_cost=65536 (64 MiB), time_cost=3 , parallelism=4 . No meu VPS, isso dá uma latência de login de cerca de 80ms — aceitável para o usuário, caro para o atacante (força bruta na GPU se torna impraticável devido à natureza memory-bound do algoritmo). Se as clínicas piloto forem cerca de 100 logando simultaneamente, você pode reduzir memory_cost para 32 MiB, mas até agora eu não cheguei nem perto dessa restrição. Por que argon2id, e não bcrypt ou scrypt? bcrypt é estável, mas não memory-hard e perde para as fazendas de GPU. scrypt memory-hard, mas menos ajustável entre os trade-offs de latência/memória. argon2id — o vencedor da Password Hashing Competition 2015, recomendação OWASP desde 2023, e sua abordagem “híbrida” combina a resistência do argon2i a ataques de canal lateral com a resistência a GPU do argon2d. Para B2B regulamentado, este é um default razoável. Audit log com classificação de categorias de PII Cada operação com dados do paciente é escrita em audit_logs com o campo data_category . Este enum é necessário para que, mediante solicitação do sujeito (ou Roskomnadzor), você possa baixar apenas as categorias relevantes com um SELECT, e não analisar entity_type / action por linhas:
pythonclass AuditLog(Base): __tablename__ = "audit_logs" id: Mapped[str] = mapped_column(String, primary_key=True) tenant_id: Mapped[str] = mapped_column(String, index=True) actor_user_id: Mapped[Optional[str]] = mapped_column(String, nullable=True) entity_type: Mapped[str] = mapped_column(String) entity_id: Mapped[str] = mapped_column(String) action: Mapped[str] = mapped_column(String) # 152-FZ art. 10 p. 2 — marcação separada de categorias especiais de PII. data_category: Mapped[str] = mapped_column(String, default="general", index=True) before_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) after_json: Mapped[Optional[dict]] = mapped_column(JSON, nullable=True) ip: Mapped[Optional[str]] = mapped_column(String, nullable=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=utcnow) __table_args__ = ( Index("ix_audit_logs_tenant_created", "tenant_id", text("created_at DESC")), )
data_category — enum de três valores, que não é definido manualmente, mas por um classificador automático por entity_type:
python_AUDIT_SPECIAL_HEALTH_ENTITIES = frozenset({ "patient", "patient_consent", "appointment", "encounter", "notification", "compliance_retention", }) _AUDIT_IDENTITY_ENTITIES = frozenset({"user", "user_group"}) def _classify_audit_entity(entity_type: str) -> str: if entity_type in _AUDIT_SPECIAL_HEALTH_ENTITIES: return "special_health" if entity_type in _AUDIT_IDENTITY_ENTITIES: return "identity" return "general"
special_health inclui não apenas diagnósticos — lá estão appointment e notification`, porque o próprio fato de se inscrever em um médico específico + o motivo do tratamento já são considerados uma categoria especial de PII em uma interpretação estrita da 152-FZ art. 10. Um classificador conservador é melhor do que uma subclassificação: se o regulador perguntar “quais dados de saúde você coleta” — é melhor errar para o lado de “mais” do que “menos”. Como 152-FZ difere do GDPR. O modelo não é transferido um para um: 152-FZ está mais próximo do HIPAA americano, pois destaca separadamente categorias especiais com requisitos adicionais para consentimento e armazenamento, além do requisito de residência (apenas data centers da Federação Russa) e registro obrigatório no registro de operadores PII no RKN. A estrutura do GDPR controller/processor/lawful-basis é semelhante em forma, mas o regulador e os documentos são diferentes. Operational tooling Duas ferramentas sem as quais eu não confiaria no banco de dados do projeto piloto.
- Read-only integrity check. Um script em sete estratégias, cada uma pegando uma classe de anomalias que a SQL-constraint não pega: Estratégia O que pega orphan-fk
tenant-scope Registros para pais inexistentes e vazamentos de FK entre tenants state-machine + date-invariants status='finalized' AND finalized_at IS NULL , ends_at <= starts_at logical-duplicates + soft-delete + audit-log Duplicatas de telefone/passaporte por tenant, anonimizadas com PII, data_category fora do enum É executado contra um restore local do dump de produção (não contra a própria produção — uma varredura de longa duração em tabelas quentes bloqueia os writers). Exit code 1 se houver descobertas — adequado como um CI-gate antes de uma grande migração. 2. Schema drift detection. Antes de cada lançamento — ciclo completo em um PG limpo:
bashmake postgres-up DATABASE_URL=... alembic upgrade head DATABASE_URL=... alembic check # autogen diff contra modelos
alembic check não é ideal: não vê a dessincronização JSONB→JSON entre Postgres e SQLite, não distingue Index(unique=True) vs UniqueConstraint . Mas os drifts básicos entre models.py e o esquema real pega de forma confiável. Em um banco de dados de desenvolvimento local com lixo acumulado, este fluxo de trabalho não funciona — o drift é mascarado por objetos existentes. Cinco conclusões 1. Multi-tenant do primeiro commit via scoped_select helper e user.tenant_id — mesmo que haja apenas um tenant. Hardcoded config.TENANT_ID — esta é uma dívida técnica que dispara no segundo tenant e requer uma revisão de cada endpoint. 2. A proteção contra race-condition de reservas deve ser feita no nível do banco de dados via EXCLUDE USING gist + tsrange — mais barato e mais confiável que o application-level locking. Números reais: 12 solicitações concorrentes → 1×200, 11×409, p95 31.69ms. Não requer Redis ou coordenação distribuída. 3. Conformidade com 152-FZ — são campos e helpers , não uma marca de marketing. data_category enum em audit_logs com autoclassificador por entity_type , migração transparente PBKDF2 → argon2id via password_needs_rehash , separado patient_consents com versionamento scoped. 4. Verificação de integridade do banco de dados como um artefato operacional : sete categorias de anomalias que as restrições SQL não pegam. Execução contra um restore local do dump de produção antes de um grande lançamento — seguro de baixo custo. 5. Schema drift detection em CI — alembic upgrade head + alembic check em um PG limpo. Não funciona em um banco de dados de desenvolvimento com lixo acumulado: o drift é mascarado por objetos existentes. Tags: multi-tenant fastapi postgresql sqlalchemy exclude using gist argon2id pbkdf2 152-фз audit log b2b saas Hubs: Python PostgreSQL Segurança da Informação Programação
