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

💻 Exemplo 1 — Notificação de Confirmação Enviada em Background

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def enviar_email_confirmacao(email: str, pedido_id: int) -> None:
    # Simula uma chamada bloqueante a um provedor SMTP/SES — o cliente HTTP
    # já recebeu a resposta antes desta função começar a executar.
    print(f"[BACKGROUND] E-mail de confirmação enviado para {email} (Pedido #{pedido_id})")


@app.post("/pedidos/{pedido_id}/confirmar")
async def confirmar_pedido(pedido_id: int, background_tasks: BackgroundTasks):
    background_tasks.add_task(enviar_email_confirmacao, "cliente@email.com", pedido_id)
    return {"status": "PROCESSANDO", "pedido_id": pedido_id}

💻 Exemplo 2 — Múltiplas Tarefas Encadeadas (PDF + Métricas) na Mesma Requisição

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()


def gerar_pdf_fatura(doc_id: int) -> None:
    print(f"[BACKGROUND] PDF da fatura #{doc_id} gerado")


def atualizar_metricas(doc_id: int) -> None:
    print(f"[BACKGROUND] Métricas de processamento atualizadas para o documento #{doc_id}")


def notificar_slack(mensagem: str) -> None:
    print(f"[BACKGROUND] Slack notificado: {mensagem}")


@app.post("/documentos/{doc_id}/processar")
async def processar_documento(doc_id: int, background_tasks: BackgroundTasks):
    # Executadas sequencialmente, na ordem de registro, após a resposta ser
    # enviada — não em paralelo entre si, mas fora do caminho crítico do cliente.
    background_tasks.add_task(gerar_pdf_fatura, doc_id)
    background_tasks.add_task(atualizar_metricas, doc_id)
    background_tasks.add_task(notificar_slack, f"Documento #{doc_id} em processamento")
    return {"mensagem": "Tarefas agendadas", "documento_id": doc_id}

💻 Exemplo 3 — Tratamento de Falhas com Log de Auditoria e Régua Celery vs Nativa

import logging

from fastapi import BackgroundTasks, FastAPI

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("tarefas_background")

app = FastAPI()


def processar_pagamento_com_log(pedido_id: int, valor: float) -> None:
    try:
        if valor <= 0:
            raise ValueError(f"Valor inválido para o pedido #{pedido_id}: {valor}")
        print(f"[BACKGROUND] Pagamento de R$ {valor:.2f} processado para o pedido #{pedido_id}")
    except Exception:
        # Como a resposta HTTP já foi enviada, o erro não pode mais virar
        # um status code — só resta registrar para auditoria/alerta.
        logger.exception("Falha ao processar pagamento em background (pedido #%s)", pedido_id)


@app.post("/pedidos/{pedido_id}/pagar")
async def pagar_pedido(pedido_id: int, valor: float, background_tasks: BackgroundTasks):
    background_tasks.add_task(processar_pagamento_com_log, pedido_id, valor)
    return {"status": "PAGAMENTO_EM_PROCESSAMENTO", "pedido_id": pedido_id}


def recomendar_estrategia_tarefas(tempo_estimado_segundos: float, exige_retries: bool) -> str:
    if tempo_estimado_segundos > 5 or exige_retries:
        return "RECOMENDADO: Celery / RQ + Redis / RabbitMQ (worker dedicado, com retry e persistência)"
    return "RECOMENDADO: BackgroundTasks nativas do FastAPI"


print(recomendar_estrategia_tarefas(1, False))    # envio de e-mail simples
print(recomendar_estrategia_tarefas(60, True))    # processamento de vídeo com retry