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["Requisição HTTP (FastAPI)"] --> B["Injeção de Dependência: get_session()"]
    B --> C["Abertura de Transação (AsyncSession)"]
    C --> D["Operação CRUD: select() / session.exec()"]
    D --> E{"Resultado da Operação"}
    E -->|"Sucesso"| F["await session.commit()"]
    E -->|"Erro / Exceção"| G["await session.rollback()"]
    F --> H["await session.refresh(instancia)"]
    H --> I["Fechamento Automático do Context Manager (finally)"]
    G --> I
    I --> J["Devolução da Conexão ao Pool"]

    subgraph ASYNC_POOL["AsyncEngine (Driver: asyncpg / aiosqlite)"]
        K["Pool de Sockets Assíncronos sem Bloqueio de Thread"]
    end

    C -.-> K

    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:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style G fill:#ffebee,stroke:#e53935,stroke-width:2px
    style H fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
    style I fill:#f5f5f5,stroke:#9e9e9e,stroke-width:2px
    style J fill:#e0f2f1,stroke:#009688,stroke-width:2px
    style K fill:#fffde7,stroke:#fbc02d,stroke-width:2px

🏛️ 1. Drivers Assíncronos vs Síncronos: Desvendando o Throughput

Ao construir APIs de alta demanda com FastAPI, a escolha do driver de banco de dados é o fator mais determinante para a vazão do sistema. Um driver síncrono clássico (como psycopg2 ou sqlite3) executa chamadas de rede no nível de socket do sistema operacional bloqueando a thread inteira do Python. Quando o banco leva 20ms para processar uma consulta pesada, a thread que chamou o driver fica completamente congelada, impossibilitada de atender qualquer outra requisição.

Com um driver assíncrono nativo (como asyncpg para PostgreSQL ou aiosqlite para SQLite) e o AsyncSession do SQLAlchemy/SQLModel:

  • Cada operação de I/O emite um await que devolve o controle instantaneamente para o Event Loop do asyncio.
  • Enquanto o PostgreSQL calcula a query no servidor de banco de dados, o processo do FastAPI continua servindo centenas de outras requisições HTTP na mesma CPU.
  • O ganho real não é a redução do tempo de resposta individual de uma requisição, mas sim o aumento drástico na concorrência e taxa de transferência total (RPS - Requests Per Second).

⚙️ 2. Configuração de AsyncEngine e Injeção de AsyncSession

No ecossistema assíncrono do SQLAlchemy 2.0 e SQLModel, utilizamos create_async_engine e async_sessionmaker. A injeção de dependências do FastAPI gerencia o ciclo de vida de cada sessão por requisição HTTP, garantindo que conexões nunca vazem:

from typing import AsyncGenerator, Annotated
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlmodel import SQLModel, select
from sqlmodel.ext.asyncio.session import AsyncSession

# 1. Configuração do Engine Assíncrono com pool de sockets
DATABASE_URL = "sqlite+aiosqlite:///./producao.db"

async_engine = create_async_engine(
    DATABASE_URL,
    echo=False,
    future=True
)

# 2. Fábrica de Sessões Assíncronas (Sem expirar objetos após commit)
async_session_maker = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autocommit=False,
    autoflush=False
)

# 3. Gerenciador de Dependência por Requisição com Teardown Seguro
async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    async with async_session_maker() as session:
        try:
            yield session
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

# Alias de anotação moderna (FastAPI 0.95+)
SessionDep = Annotated[AsyncSession, Depends(get_db_session)]

📝 3. Operações CRUD Assíncronas Idiomáticas

A sintaxe declarativa 2.0 substitui a antiga API session.query() pelo objeto select(), executado assincronamente através de await session.exec(statement):

Leitura com Filtro e Paginação

from sqlmodel import select
from typing import Sequence

async def listar_produtos_paginados(
    session: AsyncSession,
    categoria: str,
    skip: int = 0,
    limit: int = 20
) -> Sequence["Produto"]:
    # Construção declarativa da consulta
    stmt = (
        select(Produto)
        .where(Produto.categoria == categoria)
        .where(Produto.ativo == True)
        .order_by(Produto.id.desc())
        .offset(skip)
        .limit(limit)
    )
    resultado = await session.exec(stmt)
    return resultado.all()

Criação Atômica com commit e refresh

async def criar_produto_db(session: AsyncSession, dados: ProdutoCreate) -> Produto:
    # Converte DTO de entrada em entidade persistente
    novo_produto = Produto.model_validate(dados)
    
    session.add(novo_produto)
    await session.commit()
    # O refresh é indispensável para recarregar IDs, defaults gerados pelo banco e timestamps
    await session.refresh(novo_produto)
    return novo_produto

Atualização Parcial com PATCH (exclude_unset=True)

O método model_dump(exclude_unset=True) do Pydantic v2 é fundamental em endpoints PATCH: ele diferencia um campo que o cliente não enviou (não deve ser alterado) de um campo enviado explicitamente como None (deve ser limpo no banco):

async def atualizar_produto_patch(
    session: AsyncSession,
    produto_id: int,
    dados_patch: ProdutoUpdate
) -> Produto:
    stmt = select(Produto).where(Produto.id == produto_id)
    resultado = await session.exec(stmt)
    produto_existente = resultado.first()

    if not produto_existente:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Produto não encontrado")

    # Extrai estritamente os campos presentes na requisição HTTP
    alteracoes = dados_patch.model_dump(exclude_unset=True)
    for chave, valor in alteracoes.items():
        setattr(produto_existente, chave, valor)

    session.add(produto_existente)
    await session.commit()
    await session.refresh(produto_existente)
    return produto_existente

🛡️ 4. Boas Práticas: Soft Delete e expire_on_commit=False

  • O Parâmetro expire_on_commit=False: Por padrão no SQLAlchemy síncrono, após um commit(), todos os objetos na sessão são marcados como expirados. No código assíncrono, acessar um atributo (produto.nome) após o commit dispararia um recarregamento preguiçoso síncrono em segundo plano, disparando o erro MissingGreenlet: greenlet_spawn has not been called. Definir expire_on_commit=False no async_sessionmaker evita esse problema mantendo os dados em memória.
  • Soft Delete (Exclusão Lógica): Em vez de emitir await session.delete(item), em sistemas empresariais atualiza-se uma coluna de controle is_deleted = True e deleted_at = datetime.now(timezone.utc). Isso preserva o histórico fiscal, mantém integridade referencial com tabelas dependentes e permite recuperação de dados em caso de exclusão acidental.