Especialização em Backend com Python & FastAPI • FastAPI & Python 3.12+ • Pydantic v2, SQLModel, Async/Await e Microsserviços

🗺️ Mapa Conceitual do Tópico

flowchart TD
    A["Cliente HTTP / Frontend"] --> B["FastAPI Endpoint"]
    B --> C["DTO de Entrada (UsuarioCreate)"]
    C --> D["Validação Pydantic v2"]
    D --> E["Conversão para Entidade ORM (Usuario table=True)"]
    E --> F["Engine SQLAlchemy 2.0 (Pool de Conexões)"]
    F --> G[("Banco Relacional (PostgreSQL / SQLite)")]
    
    subgraph ARQ_MODELS["Hierarquia Declarativa de Schemas"]
        H["UsuarioBase (Campos Compartilhados)"]
        H --> C
        H --> I["UsuarioRead (DTO de Resposta sem Senha)"]
        H --> J["UsuarioUpdate (Campos Opcionais para PATCH)"]
        H --> E
    end

    G --> E
    E --> I
    I --> K["Serialização JSON Segura para Cliente"]

    style A fill:#e1f5fe,stroke:#03a9f4,stroke-width:2px
    style B fill:#fff3e0,stroke:#ff9800,stroke-width:2px
    style C fill:#ede7f6,stroke:#7e57c2,stroke-width:2px
    style D fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style E fill:#fce4ec,stroke:#e91e63,stroke-width:2px
    style F fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
    style G fill:#e0f2f1,stroke:#009688,stroke-width:2px

🏛️ 1. O Desafio da Duplicação e a Solução SQLModel

No desenvolvimento tradicional de APIs REST em Python utilizando SQLAlchemy e Pydantic separadamente, os engenheiros enfrentavam um problema crônico de violação do princípio DRY (Don’t Repeat Yourself):

  1. Escrevia-se uma classe declarativa SQLAlchemy com colunas de banco de dados (Column, Integer, String).
  2. Escrevia-se uma classe Pydantic quase idêntica com anotações de tipo para validar o payload JSON recebido.
  3. Escrevia-se uma terceira classe Pydantic para definir a resposta da API com orm_mode = True (atual from_attributes = True).

Qualquer adição ou renomeação de coluna no banco exigia sincronizar manualmente 3 a 5 classes em arquivos distintos, abrindo brechas para inconsistências de tipagem em tempo de execução.

O SQLModel, projetado por Sebastián Ramírez (criador do FastAPI), unifica o Pydantic v2 e o SQLAlchemy 2.0 em uma única biblioteca. Uma classe que herda de SQLModel e declara table=True é simultaneamente:

  • Um modelo de validação e serialização do Pydantic (com suporte completo a OpenAPI/Swagger).
  • Uma tabela relacional mapeada pelo SQLAlchemy Core e ORM (com metadados DDL, constraints e índices).

📐 2. O Padrão Arquitetural de Múltiplos Modelos (DTOs)

Embora o SQLModel permita usar a mesma classe com table=True para entrada, saída e banco de dados, expor a tabela do banco diretamente aos endpoints é um grave antipadrão de segurança. Essa prática expõe colunas sensíveis (como hashes de senhas, tokens internos, flags de admin ou dados confidenciais) e pode permitir ataques de Mass Assignment (quando um usuário malicioso injeta is_admin=true no JSON de cadastro).

A arquitetura profissional canônica adota a Segregação de Responsabilidades via Herança de Schemas:

Classe do Padrão table=True? Finalidade Arquitetural
UsuarioBase ❌ Não Schema abstrato com campos comuns públicos (nome, email, username).
UsuarioCreate ❌ Não DTO de entrada. Inclui senha_plana para cadastro e validações estritas.
UsuarioRead ❌ Não DTO de resposta da API (response_model). Inclui id e criado_em, mas nunca a senha.
UsuarioUpdate ❌ Não DTO para operações PATCH. Todos os campos são opcionais (None).
Usuario (DB) ✅ Sim Entidade de banco de dados física. Possui primary_key, senha_hash, constraints e índices. 
from datetime import datetime, timezone
from typing import Optional
from sqlmodel import SQLModel, Field
from pydantic import EmailStr

# 1. Base compartilhada por todos os schemas
class UsuarioBase(SQLModel):
    nome: str = Field(min_length=2, max_length=120, index=True)
    email: EmailStr = Field(unique=True, index=True)
    ativo: bool = Field(default=True)

# 2. Contrato de Entrada (Recebe a senha em texto claro para hashing)
class UsuarioCreate(UsuarioBase):
    senha_plana: str = Field(min_length=8, max_length=64, description="Senha em texto claro")

# 3. Contrato de Saída (Filtra dados sensíveis com segurança absoluta)
class UsuarioRead(UsuarioBase):
    id: int
    criado_em: datetime

# 4. Contrato de Atualização Parcial (PATCH)
class UsuarioUpdate(SQLModel):
    nome: Optional[str] = Field(default=None, min_length=2, max_length=120)
    email: Optional[EmailStr] = None
    ativo: Optional[bool] = None

# 5. Entidade Persistente Mapeada no Banco de Dados
class Usuario(UsuarioBase, table=True):
    __tablename__ = "usuarios"

    id: Optional[int] = Field(default=None, primary_key=True)
    senha_hash: str = Field(nullable=False, description="Hash seguro Argon2/Bcrypt")
    criado_em: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc),
        nullable=False
    )

⚙️ 3. Engine e Configuração de Pool de Conexões em Produção

O Engine é a fábrica central de conexões com o banco de dados. Em aplicações FastAPI de alto rendimento, o Engine deve ser instanciado como um singleton (uma única instância global reutilizada por todo o ciclo de vida do processo):

from sqlmodel import create_engine
from sqlalchemy.pool import QueuePool

DATABASE_URL = "postgresql+psycopg://app_user:segredo123@db.empresa.internal:5432/app_db"

engine = create_engine(
    DATABASE_URL,
    echo=False,              # Desativado em produção por overhead de I/O de console
    poolclass=QueuePool,     # Gerenciamento em fila com bloqueio inteligente
    pool_size=20,            # Conexões simultâneas permanentes no pool
    max_overflow=30,         # Conexões adicionais temporárias para picos de tráfego
    pool_timeout=15,         # Segundos de espera por uma conexão livre antes de TimeoutError
    pool_recycle=1800,       # Recicla conexões a cada 30 minutos (evita drop por firewalls)
    pool_pre_ping=True,      # Executa um 'SELECT 1' transparente antes de entregar a conexão
)

Importância Vital do pool_pre_ping=True

Em arquiteturas de nuvem modernas (AWS RDS, Aurora, GCP Cloud SQL, Supabase) ou através de proxies como PgBouncer, conexões inativas por mais de 5 a 15 minutos são fechadas silenciosamente pelo servidor. Sem pool_pre_ping=True, o SQLAlchemy entregará uma conexão morta da sua fila para uma requisição de usuário, resultando em erros fatais do tipo OperationalError: SSL connection has been closed unexpectedly. Com o pre-ping ativado, o SQLAlchemy detecta a conexão morta de forma transparente, descarta-a e abre uma nova sem que o usuário perceba qualquer falha.