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["Modelos SQLModel (Código Python)"] --> B["target_metadata = SQLModel.metadata"]
    C["Esquema Físico Atual do Banco de Dados"] --> D["Conexão AsyncEngine (alembic/env.py)"]
    
    B --> E["Alembic: Comparação de Metadados"]
    D --> E
    
    E --> F["alembic revision --autogenerate -m '...'"]
    F --> G["Arquivo de Migração (upgrade e downgrade)"]
    G --> H{"Revisão Manual de Código (Code Review)"}
    
    H -->|"Aprovado"| I["CI/CD Pipeline: alembic upgrade head"]
    H -->|"Problema Detectado"| J["Ajuste Manual do Script DDL"]
    J --> H

    I --> K[("Banco Atualizado com Zero-Downtime")]

    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:#fffde7,stroke:#fbc02d,stroke-width:2px
    style H fill:#e0f2f1,stroke:#009688,stroke-width:2px
    style I fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style J fill:#ffebee,stroke:#e53935,stroke-width:2px
    style K fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px

🏛️ 1. O Problema de create_all() e a Necessidade do Alembic

Em protótipos rápidos, o método SQLModel.metadata.create_all(engine) é muito utilizado para gerar as tabelas no banco de dados. Contudo, em ambientes corporativos e de produção, create_all() é completamente insuficiente:

  • Ele emite apenas comandos CREATE TABLE IF NOT EXISTS.
  • Se você adicionar um novo campo a um modelo existente, remover uma coluna ou alterar o tipo de um índice, o create_all() simplesmente ignora a alteração. O banco permanece inalterado e a aplicação falha em tempo de execução com erros de coluna ausente.
  • Ele não possui histórico de versões, impossibilitando auditoria estrutural e rollbacks em caso de incidentes operacionais.

O Alembic é o padrão oficial de migração do ecossistema SQLAlchemy/SQLModel. Ele trata o esquema relacional como uma cadeia versionada de commits (semelhante ao Git), onde cada arquivo de migração contém duas funções simétricas:

  • upgrade(): Aplica as alterações DDL incrementais para avançar a versão do banco.
  • downgrade(): Desfaz exatamente as operações do upgrade, permitindo reverter o banco para o estado anterior caso a release apresente instabilidade.

⚙️ 2. Configuração Assíncrona no env.py do Alembic

Para projetos modernos que utilizam drivers assíncronos (asyncpg, aiosqlite), o Alembic deve ser inicializado com o template assíncrono:

alembic init -t async alembic

No arquivo de configuração alembic/env.py, duas etapas são estritamente obrigatórias para que o SQLModel seja reconhecido:

  1. Importação dos Módulos de Modelos: O Python só registra classes no SQLModel.metadata quando os respectivos arquivos .py são importados pelo menos uma vez.
  2. Atribuição do target_metadata: Apontar para os metadados unificados do SQLModel.
# Trecho essencial do alembic/env.py
from logging.config import fileConfig
from sqlalchemy.ext.asyncio import async_engine_from_config
from sqlalchemy import pool
from alembic import context

# 1. Importa os metadados do SQLModel
from sqlmodel import SQLModel

# 2. Importa explicitamente todos os módulos que contêm tabelas
import app.models.usuario
import app.models.pedido
import app.models.produto

# 3. Define o target de metadados para autogenerate
target_metadata = SQLModel.metadata

def do_run_migrations(connection):
    context.configure(
        connection=connection,
        target_metadata=target_metadata,
        compare_type=True,  # Detecta alterações de tipo de coluna (ex: VARCHAR(50) -> VARCHAR(100))
    )
    with context.begin_transaction():
        context.run_migrations()

async def run_async_migrations():
    connectable = async_engine_from_config(
        config.get_section(config.config_ini_section),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    async with connectable.connect() as connection:
        await connection.run_sync(do_run_migrations)
    await connectable.dispose()

🔍 3. Autogenerate: Capacidades e Limitações Críticas

O comando de detecção automática compara a representação em memória de target_metadata com as tabelas físicas do banco conectado:

alembic revision --autogenerate -m "adiciona coluna documento_fiscal em usuarios"
O Autogenerate Detecta com Precisão O que o Autogenerate NÃO Detecta (Exige Revisão Manual)
✅ Criação de novas tabelas ⚠️ Renomeação de Coluna: Detecta como drop_column() seguido de add_column(), resultando em perda de todos os dados se não for editado manualmente para op.alter_column(new_column_name=...).
✅ Remoção de tabelas mapeadas ⚠️ Mudança de Nome de Tabela: Gera DROP e CREATE em vez de RENAME TABLE.
✅ Adição e remoção de colunas simples ⚠️ Alterações em Enum Types: Tipos enumerados nativos do PostgreSQL frequentemente exigem scripts DDL manuais via op.execute().
✅ Índices e Foreign Keys declaradas ⚠️ Valores Padrão Dinâmicos e Triggers: Procedimentos armazenados e triggers não são inspecionados.

🛡️ 4. Padrão Zero-Downtime: Adicionando Colunas NOT NULL

Em bancos de dados em produção com milhões de registros, adicionar diretamente uma coluna com constraint NOT NULL sem valor padrão falhará imediatamente (column contains null values) e causará bloqueio exclusivo da tabela (exclusive table lock), derrubando a API.

A boa prática arquitetural de engenharia adota o padrão em 3 fases (Expand & Contract) dentro da migração:

"""migracao_segura_status_usuario

Revision ID: 4f8b2c9d1e3a
Revises: 1a2b3c4d5e6f
Create Date: 2026-09-28
"""
from alembic import op
import sqlalchemy as sa

def upgrade() -> None:
    # Passo 1: Adiciona a nova coluna permitindo NULL (Sem lock prolongado)
    op.add_column("usuarios", sa.Column("status_conta", sa.String(length=20), nullable=True))
    
    # Passo 2: Executa data-backfill para todos os registros legados existentes
    op.execute("UPDATE usuarios SET status_conta = 'ATIVO' WHERE status_conta IS NULL")
    
    # Passo 3: Aplica a restrição NOT NULL agora que nenhum registro viola a regra
    op.alter_column("usuarios", "status_conta", nullable=False)

def downgrade() -> None:
    # Reversão atômica em caso de emergência operacional
    op.drop_column("usuarios", "status_conta")