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 (Mobile / Web)"] --> B["Endpoint FastAPI (Handler de Rota)"]
    B --> C["Validação do Pedido e Persistência no Banco"]
    C --> D["background_tasks.add_task(enviar_webhook, pedido_id)"]
    D --> E["FastAPI gera Response HTTP (201 Created)"]
    E --> F["Socket de Resposta Enviado ao Cliente"]
    
    subgraph POS_RESPOSTA["Execução Pós-Resposta (Mesmo Processo)"]
        F --> G["Loop de Background Tasks (Starlette)"]
        G --> H["Executa Função Agendada (Sync em Threadpool ou Async)"]
        H --> I{"Houve Exceção?"}
        I -->|"Sim"| J["Log de Erro Estruturado / Alerta Sentry"]
        I -->|"Não"| K["Conclusão Silenciosa"]
    end

    subgraph MATRIZ_DECISAO["Matriz de Decisão Arquitetural"]
        L{"Exige Persistência ou Retries?"}
        L -->|"Não (Best-effort < 3s)"| M["BackgroundTasks Nativas"]
        L -->|"Sim (Crítico / Pesado)"| N["Fila Distribuída (Celery / Redis / SQS)"]
    end

    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:#e0f2f1,stroke:#009688,stroke-width:2px
    style F fill:#c8e6c9,stroke:#2e7d32,stroke-width:2px
    style G fill:#fce4ec,stroke:#e91e63,stroke-width:2px
    style H fill:#f3e5f5,stroke:#9c27b0,stroke-width:2px
    style J fill:#ffebee,stroke:#e53935,stroke-width:2px
    style K fill:#e8f5e9,stroke:#4caf50,stroke-width:2px
    style M fill:#e1f5fe,stroke:#03a9f4,stroke-width:2px
    style N fill:#fff3e0,stroke:#ff9800,stroke-width:2px

🏛️ 1. O que são BackgroundTasks da Starlette no FastAPI

No ciclo de vida de uma requisição HTTP, operações que dependem de terceiros — como envio de e-mails via provedores SMTP (SendGrid, Mailgun), despacho de webhooks ou registro de métricas analíticas — costumam adicionar entre 200ms e 2000ms de latência desnecessária. Fazer o usuário esperar esse tempo para receber uma resposta de sucesso prejudica a experiência e consome conexões concorrentes no servidor.

O FastAPI integra nativamente a classe BackgroundTasks do Starlette. Trata-se de uma fila em memória vinculada à resposta HTTP que adia a execução de funções para o momento imediatamente posterior ao envio dos bytes da resposta pela rede ao cliente.

Tabela Comparativa de Estratégias Assíncronas

Abordagem Onde Executa Sobrevive a Crash / Restart? Suporte a Retries / Dead-Letter Caso de Uso Indicado
BackgroundTasks Nativas Mesmo processo da API ❌ Não (Volátil em RAM) ❌ Manual via código Disparo de e-mails rápidos (< 2s), auditoria, logs.
threading.Thread Avulsa Thread desvinculada ❌ Não (Vaza memória) ❌ Não rastreável ⛔ Anti-pattern: Não use em produção.
Celery + Redis / RabbitMQ Processos Workers Dedicados ✅ Sim (Persistido no Broker) ✅ Retries exponenciais, DLQ Tarefas pesadas (> 5s), geração de relatórios, vídeo.
Cloud Tasks / AWS SQS Infraestrutura Serverless ✅ Sim (Nuvem resiliente) ✅ Dead-Letter Queue nativa Arquiteturas orientadas a eventos e microsserviços.

⚙️ 2. Injeção de Parâmetros e Cuidados com Sessões de Banco

Uma das armadilhas mais graves cometidas por desenvolvedores FastAPI é tentar repassar a sessão do banco de dados da requisição (AsyncSession injetada via Depends) diretamente como parâmetro para uma tarefa em background:

⚠️ Aviso de Falha Crítica: A sessão do banco injetada no endpoint é fechada pelo context manager (finally: await session.close()) assim que a resposta HTTP é enviada. Quando a função em background tenta utilizar essa mesma sessão, ela falha com StatementError: Cannot operate on a closed session.

O Padrão Correto: Repassar Identificadores e Abrir Nova Sessão

import logging
from typing import Annotated
from fastapi import FastAPI, BackgroundTasks, Depends, status
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import select

logger = logging.getLogger("worker_background")
app = FastAPI()

# 1. Tarefa de Background independente e resiliente
async def auditar_pedido_background(pedido_id: int) -> None:
    """Abre sua própria sessão de banco isolada do ciclo HTTP da requisição."""
    try:
        from core.database import async_session_maker
        async with async_session_maker() as session:
            stmt = select(Pedido).where(Pedido.id == pedido_id)
            resultado = await session.exec(stmt)
            pedido = resultado.first()
            if pedido:
                logger.info(f"Auditoria finalizada com sucesso para Pedido #{pedido.id}")
    except Exception as exc:
        # Erros em background NÃO afetam a resposta HTTP já entregue ao cliente
        logger.exception(f"Falha irrecuperável na tarefa em background do Pedido #{pedido_id}: {exc}")

# 2. Endpoint que agenda a tarefa
@app.post("/pedidos/{pedido_id}/confirmar", status_code=status.HTTP_202_ACCEPTED)
async def confirmar_pedido(
    pedido_id: int,
    background_tasks: BackgroundTasks,
    session: Annotated[AsyncSession, Depends(get_db_session)]
):
    # Executa a regra de negócio e confirma no banco
    stmt = select(Pedido).where(Pedido.id == pedido_id)
    resultado = await session.exec(stmt)
    pedido = resultado.first()
    pedido.status = "CONFIRMADO"
    session.add(pedido)
    await session.commit()

    # Agenda a tarefa passando apenas tipos primitivos (ID do pedido)
    background_tasks.add_task(auditar_pedido_background, pedido_id=pedido.id)

    return {"status": "ACEITO", "pedido_id": pedido.id}

🔍 3. Limitações Estruturais e Limiares de Migração

Para garantir estabilidade em arquiteturas de microsserviços, considere as seguintes diretrizes para decidir quando BackgroundTasks não são mais suficientes:

  1. Ausência de Garantia de Entrega (At-Most-Once): Se o container Docker for reciclado (devido a deploy, autoscaler do Kubernetes ou reinício por erro de memória OOM), todas as tarefas agendadas em RAM pendentes de execução são perdidas para sempre.
  2. Competição por CPU com o Servidor HTTP: Processamentos que exigem cálculo matemático pesado ou manipulação de imagens reduzem a capacidade do Uvicorn de processar requisições de outros usuários, mesmo rodando após o envio da resposta.
  3. Falta de Visibilidade Operacional: Não existe painel nativo para verificar quantas tarefas estão na fila, taxa de erro ou tempo médio de execução.

Regra Prática: Se o fracasso da tarefa custar dinheiro para o negócio (como emissão de NF-e ou débito em cartão) ou durar mais de 3 segundos, migre obrigatoriamente para uma fila de mensagens dedicada (Celery, RQ ou RabbitMQ).