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["API Gateway / Router"]
    B --> C["Controller / Handler"]
    C --> D["Service Layer (Regras de Negócio)"]
    D --> E["Repository / ORM (Persistência)"]
    E --> F["Banco de Dados / Cache"]

    subgraph ARQ["Arquitetura do Capítulo"]
        G["Conceito: Sessões Assíncronas (AsyncSession) e Operações CRUD"]
        H["Segurança, Validação e Resiliência"]
        I["Alta Performance e Escalabilidade"]
    end

    D --> ARQ

    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

🏛️ 1. Por que AsyncSession em vez de Session

Uma API FastAPI síncrona que usa psycopg2 bloqueia a thread do worker inteira enquanto espera a resposta do banco — nenhuma outra requisição é processada por aquele worker durante esse tempo. Com um driver assíncrono (asyncpg para PostgreSQL, aiosqlite para SQLite) e AsyncSession, o await na chamada de I/O devolve o controle à event loop do asyncio, que aproveita esse intervalo para atender outras requisições concorrentes na mesma thread. O ganho não é velocidade por requisição individual — é vazão (throughput) sob carga concorrente de I/O.

from sqlmodel import SQLModel, Field
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine
from typing import Optional

class Produto(SQLModel, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    nome: str
    preco: float

engine: AsyncEngine = create_async_engine("sqlite+aiosqlite:///./app.db")

async def criar_tabelas() -> None:
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

Note o conn.run_sync(...): create_all é uma operação síncrona do SQLAlchemy Core que não tem equivalente assíncrono nativo, então ela é despachada para rodar dentro da conexão assíncrona via essa ponte — um padrão recorrente sempre que uma API legada do SQLAlchemy precisa ser chamada a partir de código async.

CRUD com select() e session.exec()

Desde o SQLAlchemy 2.0 (que o SQLModel adota), consultas usam a construção declarativa select() em vez do antigo session.query(). Com AsyncSession, cada chamada que toca o banco precisa de await:

from sqlmodel import select

async def buscar_caros(session: AsyncSession, limite: float) -> list[Produto]:
    stmt = select(Produto).where(Produto.preco > limite).order_by(Produto.preco)
    resultado = await session.exec(stmt)
    return list(resultado.all())

async def criar_produto(session: AsyncSession, nome: str, preco: float) -> Produto:
    produto = Produto(nome=nome, preco=preco)
    session.add(produto)
    await session.commit()
    await session.refresh(produto)   # recarrega o ID gerado pelo banco
    return produto

session.add() não grava nada — apenas marca o objeto como pendente na sessão. Só await session.commit() executa o INSERT/UPDATE e finaliza a transação; session.refresh() é necessário depois porque, após o commit, o SQLAlchemy expira os atributos do objeto em memória por padrão, e um acesso a produto.id sem refresh dispararia uma nova consulta implícita (ou falharia, fora do contexto assíncrono correto).

Atualização parcial e soft delete

Para PATCH, o padrão idiomático usa model_dump(exclude_unset=True), que retorna apenas os campos explicitamente enviados no payload — distinguindo “campo omitido” de “campo enviado como None”:

class ProdutoUpdate(SQLModel):
    nome: Optional[str] = None
    preco: Optional[float] = None

async def aplicar_patch(session: AsyncSession, produto: Produto, dados: ProdutoUpdate) -> Produto:
    for campo, valor in dados.model_dump(exclude_unset=True).items():
        setattr(produto, campo, valor)
    session.add(produto)
    await session.commit()
    await session.refresh(produto)
    return produto

Exclusão física (DELETE) quebra integridade referencial e histórico de auditoria em sistemas corporativos; por isso muitas tabelas preferem soft delete — uma flag is_deleted e um deleted_at, filtrados nas consultas normais, preservando o registro para relatórios e rastreabilidade.

🔗 Recursos Pedagógicos do Capítulo 06

Recurso Didático Finalidade Link de Acesso
📊 Slides de Aula Apresentação visual interativa com Dark Mode e suporte a teclado Ver Slides
🧠 Quiz Formativo Teste interativo de fixação com feedback imediato por alternativa Fazer Quiz
💻 Exemplos de Código Demonstrações funcionais com código executável Ver Exemplos
🧩 Exercícios em 4 Níveis Lista progressiva de fixação com gabarito em bloco colapsável Resolver Exercícios
⬅️ Capítulo Anterior 📚 Sumário de Tópicos Próximo Capítulo ➡️