Especialização em Backend com Python & FastAPI • Demonstrações Práticas Compiláveis

💻 Exemplo 1 — Modelagem SQLModel: Pedido, ItemPedido e Total Derivado

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
    status: StatusPedido = StatusPedido.CRIADO

    itens: list[ItemPedido] = Relationship(back_populates="pedido")

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


# Demonstração isolada (sem banco), só para validar o cálculo do total:
pedido = Pedido(id=1, cliente_id=42)
pedido.itens = [
    ItemPedido(pedido_id=1, sku="SKU-MOUSE", quantidade=2, preco_unitario=199.90),
    ItemPedido(pedido_id=1, sku="SKU-TECLADO", quantidade=1, preco_unitario=349.00),
]
print(f"Pedido #{pedido.id} — Total: R$ {pedido.total:.2f}")

💻 Exemplo 2 — Camadas Service/Repository com Transação Atômica de Faturamento

from dataclasses import dataclass


class PedidoNaoEncontrado(Exception):
    def __init__(self, pedido_id: int) -> None:
        super().__init__(f"Pedido #{pedido_id} não encontrado")


class TransicaoInvalida(Exception):
    def __init__(self, atual: str, esperado: str) -> None:
        super().__init__(f"Não é possível ir de '{atual}' para '{esperado}'")


class EstoqueInsuficiente(Exception):
    def __init__(self, sku: str) -> None:
        super().__init__(f"Estoque insuficiente para o SKU '{sku}'")


@dataclass
class RegistroEstoque:
    sku: str
    saldo: int


class PedidoRepositoryFake:
    """Repositório em memória — em produção, encapsularia sessões SQLModel/SQLAlchemy."""

    def __init__(self) -> None:
        self._pedidos: dict[int, dict] = {
            1: {"status": "PAGO", "itens": [{"sku": "SKU-MOUSE", "quantidade": 2}]}
        }
        self._estoque: dict[str, RegistroEstoque] = {"SKU-MOUSE": RegistroEstoque("SKU-MOUSE", 10)}

    def obter(self, pedido_id: int) -> dict:
        pedido = self._pedidos.get(pedido_id)
        if pedido is None:
            raise PedidoNaoEncontrado(pedido_id)
        return pedido

    def faturar_com_baixa_estoque(self, pedido_id: int) -> dict:
        # Simula "async with session.begin(): ..." — todas as escritas abaixo
        # são tratadas como uma unidade atômica: ou todas persistem, ou nenhuma.
        pedido = self.obter(pedido_id)
        if pedido["status"] != "PAGO":
            raise TransicaoInvalida(pedido["status"], "FATURADO")

        for item in pedido["itens"]:
            estoque = self._estoque[item["sku"]]
            if estoque.saldo < item["quantidade"]:
                raise EstoqueInsuficiente(item["sku"])  # equivale a um ROLLBACK

        for item in pedido["itens"]:
            self._estoque[item["sku"]].saldo -= item["quantidade"]
        pedido["status"] = "FATURADO"
        return pedido


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

    def faturar(self, pedido_id: int) -> dict:
        return self._repo.faturar_com_baixa_estoque(pedido_id)


service = PedidoService(PedidoRepositoryFake())
resultado = service.faturar(1)
print(f"Pedido faturado: {resultado}")

💻 Exemplo 3 — Endpoint Integrado: responses= de Erro + Auditoria em Background

import time

from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="API de Pedidos e Faturamento", version="1.0.0")


class ErroSchema(BaseModel):
    detail: str


class PedidoFaturadoResponse(BaseModel):
    pedido_id: int
    status: str
    nfe: str


def registrar_auditoria(usuario_id: int, acao: str, detalhes: dict) -> None:
    # Executa após a resposta HTTP já ter sido enviada ao cliente (Cap. 13).
    registro = {"timestamp": int(time.time()), "usuario_id": usuario_id, "acao": acao, "detalhes": detalhes}
    print(f"[AUDITORIA] {registro}")


def get_pedido_service() -> PedidoService:
    return PedidoService(PedidoRepositoryFake())


@app.post(
    "/pedidos/{pedido_id}/faturar",
    response_model=PedidoFaturadoResponse,
    responses={
        404: {"model": ErroSchema, "description": "Pedido não encontrado"},
        409: {"model": ErroSchema, "description": "Transição inválida ou estoque insuficiente"},
    },
    tags=["Faturamento"],
)
async def faturar_pedido(
    pedido_id: int,
    background_tasks: BackgroundTasks,
    service: PedidoService = Depends(get_pedido_service),
):
    try:
        resultado = service.faturar(pedido_id)
    except PedidoNaoEncontrado as exc:
        raise HTTPException(status_code=404, detail=str(exc)) from exc
    except (TransicaoInvalida, EstoqueInsuficiente) as exc:
        raise HTTPException(status_code=409, detail=str(exc)) from exc

    nfe = f"NFE-{pedido_id:06d}-{int(time.time())}"
    background_tasks.add_task(
        registrar_auditoria, usuario_id=1, acao="FATURAR_PEDIDO",
        detalhes={"pedido_id": pedido_id, "nfe": nfe},
    )
    return PedidoFaturadoResponse(pedido_id=pedido_id, status=resultado["status"], nfe=nfe)