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: Projeto Corporativo: API de Gestão de Pedidos com Auditoria JSON"]
        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. O projeto integrador: onde os 19 capítulos anteriores convergem

Este capítulo não introduz um mecanismo novo isolado — ele monta uma API real de gestão de pedidos e faturamento que exige, simultaneamente, quase tudo o que foi construído ao longo da especialização: modelagem relacional com SQLModel (Cap. 6-9), autenticação OAuth2/JWT e RBAC (Cap. 10-11), tratamento centralizado de exceções via exception_handlers (Cap. 12), tarefas em background para notificação assíncrona (Cap. 13), cache distribuído para consultas de leitura pesada (Cap. 15), testes automatizados (Cap. 16) e documentação OpenAPI rica dos fluxos de erro (Cap. 18). A dificuldade real de um projeto corporativo nunca está em um mecanismo isolado — está em fazê-los cooperar sem que a base de código vire um emaranhado.

Modelagem relacional: Pedido e ItemPedido com SQLModel

Um pedido não é uma linha isolada; é uma entidade com itens filhos e um total derivado. SQLModel (que unifica Pydantic e SQLAlchemy) modela essa relação com Relationship, e o total é calculado a partir dos itens em vez de armazenado de forma redundante e sujeita a desincronização:

from enum import Enum
from sqlmodel import Field, Relationship, SQLModel

class StatusPedido(str, Enum):
    CRIADO = "CRIADO"
    PAGO = "PAGO"
    FATURADO = "FATURADO"
    CANCELADO = "CANCELADO"

class ItemPedido(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    pedido_id: int = Field(foreign_key="pedido.id")
    sku: str
    quantidade: int
    preco_unitario: float
    pedido: "Pedido" = Relationship(back_populates="itens")

class Pedido(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    cliente_id: int = Field(foreign_key="usuario.id")
    status: StatusPedido = StatusPedido.CRIADO
    itens: list[ItemPedido] = Relationship(back_populates="pedido")

    @property
    def total(self) -> float:
        return sum(i.quantidade * i.preco_unitario for i in self.itens)

Camadas: Router, Service e Repository desacoplados

Colocar consulta SQL, regra de negócio e serialização HTTP na mesma função é o caminho mais rápido para um main.py de milhares de linhas intestável. A separação em camadas isola cada responsabilidade: o Router só traduz HTTP ↔ chamadas de função; o Service concentra as regras de negócio (transições de status válidas, cálculo de faturamento); o Repository encapsula o acesso a dados, permitindo trocar a implementação (ou mocá-la em testes unitários) sem tocar na lógica de negócio:

class PedidoService:
    def __init__(self, repo: "PedidoRepository") -> None:
        self._repo = repo

    async def faturar(self, pedido_id: int) -> Pedido:
        pedido = await self._repo.obter(pedido_id)
        if pedido is None:
            raise PedidoNaoEncontrado(pedido_id)
        if pedido.status != StatusPedido.PAGO:
            raise TransicaoInvalida(pedido.status, StatusPedido.FATURADO)
        pedido.status = StatusPedido.FATURADO
        return await self._repo.salvar(pedido)

Atomicidade: faturamento e baixa de estoque não podem divergir

Faturar um pedido e debitar o estoque correspondente são duas escritas que precisam ter sucesso — ou falhar — juntas. Se a baixa de estoque falhar depois que o pedido já foi marcado como FATURADO, o sistema fica em um estado contábil inconsistente (nota fiscal emitida para um item sem estoque real). A transação explícita do SQLAlchemy/SQLModel garante que ambas as escritas sejam confirmadas atomicamente ou revertidas juntas:

async def faturar_com_baixa_estoque(session, pedido_id: int, item_sku: str, qtd: int) -> None:
    async with session.begin():
        pedido = await session.get(Pedido, pedido_id)
        estoque = await session.get(Estoque, item_sku)
        if estoque.saldo < qtd:
            raise ValueError("Estoque insuficiente")  # dispara ROLLBACK automático
        estoque.saldo -= qtd
        pedido.status = StatusPedido.FATURADO
        # COMMIT único ao sair do bloco sem exceção

Documentando os erros de negócio no contrato OpenAPI

Retomando o Capítulo 18: os erros específicos do domínio — pedido inexistente, transição de status inválida, estoque insuficiente — devem ser declarados via responses= na rota, não deixados implícitos, para que o schema publicado seja o contrato real da API:

@app.post(
    "/pedidos/{pedido_id}/faturar",
    response_model=PedidoResponse,
    responses={
        404: {"model": ErroSchema, "description": "Pedido não encontrado"},
        409: {"model": ErroSchema, "description": "Transição de status inválida ou estoque insuficiente"},
    },
    tags=["Faturamento"],
)
async def faturar_pedido(pedido_id: int, service: PedidoService = Depends(get_pedido_service)):
    return await service.faturar(pedido_id)

Trilha de auditoria como tarefa em background

Registrar quem fez o quê (usuário, ação, payload alterado, timestamp) é obrigatório em sistemas financeiros, mas não deve bloquear a resposta ao cliente. A gravação do log de auditoria é um candidato natural para BackgroundTasks (Cap. 13): a resposta HTTP do faturamento não precisa esperar a escrita do registro de auditoria em disco/banco de logs para ser enviada.

Cache nas consultas de leitura, invalidação nas escritas

Consultas de listagem de pedidos por cliente são lidas com muito mais frequência do que pedidos são criados — um candidato natural para cache distribuído (Cap. 15) com TTL curto, invalidado explicitamente sempre que o status de um pedido daquele cliente muda, evitando que o cliente veja um status desatualizado (PAGO quando já foi FATURADO) por estar servindo de um cache obsoleto.