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["POST /pedidos/{id}/confirmar"] --> B["Handler da rota"]
    B --> C["background_tasks.add_task(enviar_email, ...)"]
    C --> D["return response (202/200 imediato)"]
    D --> E["Cliente recebe resposta"]
    D -.->|"após enviar a resposta"| F["Tarefa executa no mesmo processo"]
    F -->|"sucesso"| G["Log de auditoria"]
    F -->|"falha"| H["Exceção tratada internamente"]

    I{"Tempo estimado > 2-5s\nou exige retry/persistência?"} -->|"sim"| J["Migrar para Celery/RQ + Redis/RabbitMQ"]
    I -->|"não"| K["BackgroundTasks nativas são suficientes"]

    style A fill:#e1f5fe,stroke:#03a9f4,stroke-width:2px
    style C fill:#fff3e0,stroke:#ff9800,stroke-width:2px
    style D fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style F fill:#ede7f6,stroke:#7e57c2,stroke-width:2px
    style J fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
    style K fill:#e8f5e9,stroke:#4caf50,stroke-width:2px

🏛️ 1. BackgroundTasks: paralelismo leve depois da resposta HTTP

BackgroundTasks resolve um problema específico: executar uma função depois que a resposta HTTP já foi enviada ao cliente, sem fazê-lo esperar. O caso canônico é enviar um e-mail de confirmação — o cliente não precisa (nem deveria) aguardar a latência de um provedor SMTP externo para receber a confirmação de que seu pedido foi criado.

Mecânica: injeção via parâmetro de rota

Diferente de outras dependências, BackgroundTasks é declarado diretamente como tipo do parâmetro da rota — o FastAPI injeta automaticamente uma instância por requisição, e qualquer tarefa agendada com .add_task() só é executada após a Response ser serializada e o socket de resposta ser fechado para o cliente:

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def enviar_email_confirmacao(email: str, pedido_id: int) -> None:
    # Chamada bloqueante para um provedor de e-mail (SMTP, SES, SendGrid).
    print(f"[BACKGROUND] E-mail enviado para {email} sobre o 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}

Múltiplas chamadas a .add_task() na mesma requisição são executadas sequencialmente, na ordem em que foram registradas — não em paralelo. Se as tarefas forem independentes e IO-bound, isso ainda é aceitável porque elas já rodam fora do caminho crítico da resposta; mas não é um mecanismo de concorrência real.

Por que não usar time.sleep() ou threading.Thread() direto

Chamar uma função bloqueante com time.sleep() dentro do handler da rota antes do return trava o event loop inteiro do servidor ASGI para todas as outras requisições concorrentes — o oposto do efeito desejado. Criar uma threading.Thread() manualmente também é desaconselhado: threads soltas não são gerenciadas pelo ciclo de vida do Starlette, exceções dentro delas somem silenciosamente (não aparecem nos logs por padrão) e não há garantia de que a thread termine antes do processo Uvicorn ser encerrado. BackgroundTasks resolve os dois problemas: é integrado ao ciclo de vida ASGI e suas exceções, se não tratadas manualmente, ao menos aparecem no log padrão de erros não capturados.

Tratamento de erros dentro da tarefa

Como a tarefa roda depois da resposta já ter sido enviada, uma exceção não tratada dentro dela não pode mais virar um código de status HTTP — o cliente já recebeu 200 OK e não há como “desfazer” essa resposta. Por isso, toda tarefa em background deve tratar suas próprias falhas internamente, tipicamente registrando em um log de auditoria:

import logging

logger = logging.getLogger("tarefas_background")

def tarefa_segura(dados: dict | None) -> None:
    try:
        if not dados:
            raise ValueError("Dados vazios recebidos pela tarefa")
        print(f"Processado com sucesso: {dados}")
    except Exception:
        logger.exception("Falha ao processar tarefa em background")

A limitação estrutural: memória do processo, sem persistência

BackgroundTasks vive inteiramente na memória do processo Uvicorn que atendeu a requisição. Se o processo cair, reiniciar (deploy, crash, autoscaling) ou o worker for reciclado antes da tarefa terminar, ela é perdida silenciosamente — não existe fila persistida em disco ou banco, não existe retry automático, não existe visibilidade de “quantas tarefas estão pendentes agora”. Isso é aceitável para uma notificação best-effort, mas inaceitável para uma cobrança financeira ou geração de nota fiscal.

Quando migrar para Celery/RQ + Redis/RabbitMQ

A régua prática: se a tarefa (a) leva mais que alguns segundos, (b) precisa de retry automático em caso de falha, (c) precisa sobreviver a um reinício do servidor, ou (d) precisa ser distribuída entre múltiplos workers dedicados (não competindo por CPU com o servidor web), ela pertence a uma fila real como Celery ou RQ, apoiada em um broker como Redis ou RabbitMQ — não a BackgroundTasks:

def recomendar_estrategia(tempo_estimado_segundos: float, exige_retry: bool) -> str:
    if tempo_estimado_segundos > 5 or exige_retry:
        return "Fila dedicada (Celery/RQ + Redis/RabbitMQ)"
    return "BackgroundTasks nativas do FastAPI"

print(recomendar_estrategia(1.5, exige_retry=False))   # e-mail simples
print(recomendar_estrategia(120, exige_retry=True))    # processamento de vídeo

🔗 Recursos Pedagógicos do Capítulo 13

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 ➡️