Background tasks no FastAPI: TaskGroup, lifespan, cancelamento e retries

Quando e por que usar

Background tasks aparecem quando uma API precisa desencadear trabalho assíncrono que não precisa bloquear a resposta HTTP. Exemplos comuns: envio de e-mail, atualização de cache, indexação em um motor de busca, chamadas para serviços externos de longa duração, ou pipelines de processamento que podem ser processados fora do ciclo de requisição.

Escolher executar trabalho em background dentro do próprio processo do FastAPI faz sentido quando a carga é moderada e o trabalho é relativamente isolado, com requisitos de latência e durabilidade que permitem alguma perda pontual (ou quando você tem replicação e tolerância à perda). Em contrapartida, para workloads críticos com alto volume e necessidade de persistência, filas externas (RabbitMQ, Redis Streams) e workers separados são melhores.

Este artigo assume que você já domina FastAPI básico e asyncio. O objetivo é mapear padrões práticos: uso de BackgroundTasks nativo, como usar asyncio.TaskGroup para coordenar tarefas, integração com o lifespan do app e estratégias para cancelamento e retries controlados.

Cenários onde usar tarefas em background embutidas

Use tarefas embutidas quando:

• O trabalho tolera perda ocasional (stateless ou idempotente).
• O throughput previsto é pequeno a moderado (não milhares de tasks por segundo).
• Você quer evitar infra adicional ou tem requisitos de latência baixos.

Esses cenários funcionam bem para notificações, atualizações de cache, tarefas de limpeza e outros side-effects leves.

Quando não usar

Não use tarefas embutidas quando:

1. Precisa de garantia de entrega (persistência/ack).
2. Trabalho CPU-bound massivo sem pool de workers.
3. Há necessidade de escalonamento independente da API.

Para esses casos, prefira filas externas ou sistemas de orquestração (Celery, RQ, Kafka, etc.).

Comparacao com alternativas

Existem alternativas técnicas para executar tarefas em background. Vamos comparar ponto a ponto: BackgroundTasks do FastAPI, asyncio.TaskGroup, processos/threads, e filas externas.

BackgroundTasks do FastAPI

BackgroundTasks é uma API simples para agendar uma callable (sync ou async) que será executada depois que a resposta for enviada. Ela fica simples de usar dentro de rotas e tem integração direta com dependências.

Limitações:

• Sem controle fino de lifecycle (não ligado ao lifespan do app).
• Sem retries automáticos, cancelamento centralizado ou monitoramento por padrão.

asyncio.TaskGroup e as tarefas do loop

TaskGroup (Python 3.11+) fornece um contexto para agrupar e esperar tarefas com propagação de exceções mais segura. É útil no contexto de lifespan para iniciar workers de longa duração e garantir cancelamento ordenado no shutdown.

Vantagem chave: integração com lifecycle e capacidade de cancelar ou aguardar todas as subtarefas em shutdown.

Filas externas e workers

Celery, RQ, Kafka e afins oferecem durabilidade, escalabilidade independente e retries nativos. Use quando a garantia de processamento é mais importante que a simplicidade da arquitetura.

Trade-off: aumento de complexidade operacional e latência inter-componentes.

Implementacao progressiva

Nesta seção vamos evoluir uma implementação desde BackgroundTasks simples até um padrão de TaskGroup com cancelamento e retries. Em cada passo explico o que muda e por que a decisão importa.

1) Exemplo básico com BackgroundTasks

Contexto: rota que grava um registro e dispara um processamento assíncrono leve (por exemplo, enfileirar uma chamada HTTP não crítica).

O código abaixo mostra como utilizar BackgroundTasks. Ele é funcional, usando httpx para chamada externa assíncrona.

from fastapi import FastAPI, BackgroundTasks
import httpx

app = FastAPI()

async def notify_external_service(payload: dict) -> None:
    async with httpx.AsyncClient(timeout=10) as client:
        await client.post('https://httpbin.org/post', json=payload)

@app.post('/items/')
async def create_item(item: dict, background_tasks: BackgroundTasks):
    # gravação no DB/logic aqui (omitido)
    background_tasks.add_task(notify_external_service, item)
    return {'status': 'accepted'}

O que acontece: FastAPI registra a callable e a executa após o envio da resposta. Simples e direto. Falhas dentro de notify_external_service não são automaticamente re-tentadas nem expostas ao cliente.

2) Migrando para TaskGroup no lifespan

Contexto: queremos workers persistentes que processem uma fila local (asyncio.Queue) e que iniciem/fechem com o app, cancelando corretamente no shutdown.

O seguinte exemplo usa asyncio.TaskGroup no contexto de lifespan para criar dois workers que processam itens da fila.

import asyncio
from fastapi import FastAPI
from contextlib import asynccontextmanager

queue = asyncio.Queue()

async def worker(name: str):
    while True:
        payload = await queue.get()
        try:
            # processa payload
            print(name, 'processing', payload)
            await asyncio.sleep(0.5)
        finally:
            queue.task_done()

@asynccontextmanager
async def lifespan(app: FastAPI):
    tg = asyncio.TaskGroup()
    await tg.__aenter__()
    tg.create_task(worker('w1'))
    tg.create_task(worker('w2'))
    try:
        yield
    finally:
        await tg.__aexit__(None, None, None)

app = FastAPI(lifespan=lifespan)

@app.post('/enqueue/')
async def enqueue(item: dict):
    await queue.put(item)
    return {'enqueued': True}

O que acontece: ao iniciar o app, dois workers são criados. No shutdown, TaskGroup cancela e aguarda as tarefas. Isso evita tasks órfãs e deadlocks durante o encerramento.

3) Cancelamento e retries controlados

Contexto: precisamos de retries com backoff e a capacidade de cancelar tentativas quando o app está fechando. Vamos usar um padrão que combina um loop de retries e a verificação de uma flag de cancelamento baseada em asyncio.Event.

O exemplo abaixo demonstra uma função de retry simples e cancelável. Evite reinventar a roda em cenários complexos — bibliotecas como tenacity podem ajudar, mas aqui mostramos o princípio para controle fino.

import asyncio
import random

async def retryable_call(payload: dict, cancel_event: asyncio.Event):
    max_attempts = 5
    base = 0.5
    for attempt in range(1, max_attempts + 1):
        if cancel_event.is_set():
            raise asyncio.CancelledError('shutdown requested')
        try:
            # simula chamada falha/sucesso
            if random.random() < 0.7:
                raise RuntimeError('temporary failure')
            return {'ok': True}
        except Exception as exc:
            if attempt == max_attempts:
                raise
            backoff = base * (2 ** (attempt - 1))
            await asyncio.sleep(backoff)

# uso: o cancel_event vem do lifespan e é setado no shutdown

O que acontece: a função respeita cancel_event para abortar o loop de retries durante o shutdown, evitando bloqueios longos e permitindo um encerramento rápido.

Integracao com ecossistema real

Integrações práticas com banco de dados, filas externas, observabilidade e deploy são cruciais. Nesta seção exploramos como ligar as peças sem perder garantias importantes.

Banco de dados e sessões

Tarefas em background que acessam o banco precisam de sua própria sessão/connection. Não reutilize objetos de sessão ligados ao escopo da requisição. Use um pool de conexões e crie/feche sessões dentro da task.

Exemplo de anti-padrão: passar uma sessão SQLAlchemy da rota para a task e esperar que ainda exista no momento de execução. Isso leva a erros ou conexões fechadas.

Erros comuns e armadilhas

Armadilha 1: confundir BackgroundTasks com filas duráveis. BackgroundTasks roda no processo atual — um crash do processo perde as tasks pendentes.

Armadilha 2: bloquear o event loop com trabalho CPU-bound. Se precisar de CPU-bound, use ProcessPoolExecutor ou workers externos.

Armadilha 3: não lidar com cancelamento. Tasks que ignoram cancelamento podem atrasar ou travar o shutdown do servidor.

Tip: sempre projete suas tasks para serem idempotentes e canceláveis. Use eventos de cancelamento e limites de timeout.

Observabilidade e tracing

Para production, integre tracing (OpenTelemetry), métricas (Prometheus) e logs estruturados. Exponha métricas como: tasks em fila, tasks falhadas, retries, latência média de processamento.

Se usar TaskGroup com workers, reporte a contagem de tasks ativas e tempo médio de processamento por worker. Isso ajuda a detectar hotspots e dimensionar corretamente.

Checklist de qualidade para producao

Antes de colocar qualquer padrão de background tasks em produção, rode o checklist abaixo. Ele cobre confiabilidade, desempenho e segurança.

Observabilidade e métricas

• Métricas básicas: enqueued, processed_success, processed_failed, retry_count, queue_size.
• Tracing de chamadas externas com contexto que liga request -> task -> external call.
• Logs estruturados com correlação (request_id, task_id).

Sem métricas, é impossível operar em produção de forma pró-ativa.

Gerenciamento de erros e retries

Padronize a estratégia de retries: quais exceções são retriáveis, backoff máximo, número de tentativas e política de dead-letter. Registre failures em uma DLQ ou armazenagem para investigação posterior.

Se optar por retries automáticos, proteja endpoints externos contra retry storms usando jitter e limites.

Testes e validação

Tenha testes que simulem shutdown em diferentes momentos (durante processamento, entre retries) para validar cancelamento seguro. Mock de chamadas externas e uso de event loop real em testes de integração ajudam a encontrar problemas de concorrência.

Deploy e scaling

Se executar múltiplas réplicas, tenha certeza que tarefas não idempotentes não serão executadas múltiplas vezes sem coordenação. Para jobs que não toleram duplicação, prefira filas externas com mecanismos de lock/lease (Redis, SQS FIFO, etc.).

Conclusao e passos práticos

Resumo prático: para cargas leves a moderadas, BackgroundTasks + TaskGroup no lifespan oferece uma solução simples e robusta. Use BackgroundTasks para side-effects rápidos; use TaskGroup para workers persistentes que precisam iniciar/fechar com o app. Implemente cancelamento baseado em asyncio.Event para shutdowns ágeis e combine com uma estratégia de retries com backoff e jitter.

Checklist prático de implantação:

1. Defina quais tasks podem rodar in-process e quais exigem persistência.
2. Implemente TaskGroup no lifespan para workers long-lived.
3. Adicione cancel_event para permitir shutdown rápido e seguro.
4. Padronize retries e DLQ para falhas persistentes.
5. Exponha métricas e tracing para monitoramento.

Implementar essas práticas reduz riscos operacionais e facilita a transição para filas externas quando a carga crescer. Adote sempre o princípio de menor privilégio e projetar tarefas idempotentes sempre que possível.

Se quiser um exemplo avançado combinando TaskGroup, cancelamento e integração com Redis como DLQ, posso expandir com um repositório de referência contendo deploy com Docker, health checks e testes de integração.