--- title: "FastAPI Background Tasks, Celery e Redis: quando usar cada um" url: "https://python.dev.br/blog/fastapi-background-tasks-celery-redis-2026/" markdown_url: "https://python.dev.br/blog/fastapi-background-tasks-celery-redis-2026.MD" description: "Compare BackgroundTasks do FastAPI, Celery e Redis Queue para jobs assíncronos em Python: casos de uso, limites, arquitetura, deploy e exemplos práticos." date: "2026-05-23" author: "Equipe Python Brasil" --- # FastAPI Background Tasks, Celery e Redis: quando usar cada um Compare BackgroundTasks do FastAPI, Celery e Redis Queue para jobs assíncronos em Python: casos de uso, limites, arquitetura, deploy e exemplos práticos. Toda API Python chega em um ponto em que uma operação não deveria acontecer dentro da requisição principal. Enviar email, gerar PDF, processar imagem, sincronizar CRM, chamar uma API instável ou importar uma planilha grande são tarefas úteis, mas ruins para a experiência do usuário quando deixam o endpoint lento ou sujeito a timeout. No ecossistema Python, três caminhos aparecem com frequência: **BackgroundTasks do FastAPI**, **Celery** e filas simples com **Redis**. Eles resolvem problemas parecidos, mas não têm o mesmo custo operacional nem a mesma confiabilidade. Escolher errado pode gerar uma arquitetura pesada demais para um produto pequeno ou frágil demais para um sistema que já está em produção. Este guia complementa o tutorial de [APIs REST com FastAPI](/blog/apis-rest-com-fastapi/), o guia de [Python e Redis](/blog/python-e-redis/) e o verbete sobre [Celery](/glossario/celery/). A ideia é ajudar você a decidir o menor mecanismo seguro para cada caso. ## Resumo rápido da decisão Use esta regra prática: | Situação | Melhor escolha inicial | |---|---| | Enviar email simples depois de responder ao usuário | `BackgroundTasks` do FastAPI | | Registrar evento leve em outra API interna | `BackgroundTasks`, com timeout curto | | Processar imagem, CSV, PDF ou áudio | Celery ou RQ com Redis | | Retry automático, agendamento, workers separados e observabilidade | Celery | | Fila simples para projeto pequeno com Redis já disponível | RQ ou Dramatiq | | Job crítico que não pode se perder se o processo web reiniciar | Fila externa, não `BackgroundTasks` | O ponto central é durabilidade. `BackgroundTasks` roda no mesmo processo da aplicação. Se o worker cair, reiniciar ou for encerrado durante deploy, a tarefa pode não terminar. Celery e filas externas adicionam broker, workers e mais peças, mas dão mais controle sobre retry, concorrência e recuperação. ## O que o BackgroundTasks do FastAPI faz O FastAPI inclui `BackgroundTasks` para executar uma função depois que a resposta HTTP já foi enviada. É ótimo para trabalho pequeno e não crítico. ```python from fastapi import BackgroundTasks, FastAPI from pydantic import BaseModel, EmailStr app = FastAPI(title="Exemplo BackgroundTasks") class CadastroRequest(BaseModel): email: EmailStr nome: str def enviar_email_boas_vindas(email: str, nome: str) -> None: # Em produção, use timeout e tratamento de erro no cliente SMTP/API. print(f"Enviando boas-vindas para {nome} <{email}>") @app.post("/cadastros", status_code=202) def criar_cadastro(payload: CadastroRequest, background_tasks: BackgroundTasks): # 1. Salve o cadastro no banco aqui. # 2. Responda rápido ao usuário. background_tasks.add_task(enviar_email_boas_vindas, payload.email, payload.nome) return {"status": "recebido"} ``` Esse padrão melhora a latência percebida: o usuário recebe resposta antes do envio do email. Mas a tarefa ainda depende do processo web. Ela não é uma fila durável, não distribui carga entre workers independentes e não substitui um sistema de jobs. ## Limites do BackgroundTasks Evite `BackgroundTasks` quando a tarefa: - demora muitos segundos ou minutos; - precisa de retry garantido; - consome CPU ou memória de forma relevante; - precisa sobreviver a deploy, autoscaling ou reinício do processo; - deve ser monitorada por equipe de operação; - envolve cobrança, pagamento, documento fiscal ou outra ação crítica. Um erro comum é usar `BackgroundTasks` para importação de planilhas grandes. O endpoint responde rápido, mas o processo web continua ocupado, sem fila, sem controle fino de concorrência e sem histórico confiável. Se duas pessoas enviarem arquivos grandes ao mesmo tempo, a API pode ficar instável. ## Quando Celery entra Celery é a opção clássica para filas distribuídas em Python. Ele separa três papéis: 1. a aplicação web publica uma tarefa; 2. o broker, como Redis ou RabbitMQ, guarda a mensagem; 3. um ou mais workers executam a tarefa fora do processo web. ```python # celery_app.py from celery import Celery celery_app = Celery( "app_jobs", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1", ) celery_app.conf.update( task_serializer="json", accept_content=["json"], result_serializer="json", timezone="America/Sao_Paulo", task_time_limit=300, task_soft_time_limit=240, ) ``` ```python # tasks.py from celery_app import celery_app @celery_app.task(bind=True, autoretry_for=(TimeoutError,), retry_backoff=True, max_retries=3) def gerar_relatorio_cliente(self, cliente_id: int) -> dict: # Busque dados, gere arquivo, salve em storage e retorne metadados. return {"cliente_id": cliente_id, "status": "gerado"} ``` ```python # api.py from fastapi import FastAPI from tasks import gerar_relatorio_cliente app = FastAPI() @app.post("/clientes/{cliente_id}/relatorios", status_code=202) def solicitar_relatorio(cliente_id: int): task = gerar_relatorio_cliente.delay(cliente_id) return {"task_id": task.id, "status": "enfileirado"} ``` Com Celery, você consegue escalar workers separadamente da API, limitar concorrência, configurar retries, registrar resultados, usar filas diferentes por prioridade e acompanhar falhas. Esse poder cobra preço: mais configuração, mais infraestrutura e mais pontos de falha. ## Redis como broker ou fila simples Redis aparece em dois papéis diferentes. Ele pode ser broker do Celery ou a base de uma fila mais simples, como RQ. Para um produto pequeno, RQ costuma ser mais fácil de entender: você coloca jobs em uma fila e um worker Python consome. ```python # enqueue.py from redis import Redis from rq import Queue from jobs import enviar_recibo redis_conn = Redis.from_url("redis://localhost:6379/0") queue = Queue("emails", connection=redis_conn) job = queue.enqueue(enviar_recibo, pedido_id=123, retry=3) print(job.id) ``` ```python # jobs.py def enviar_recibo(pedido_id: int) -> None: print(f"Enviando recibo do pedido {pedido_id}") ``` RQ é suficiente para muitos backoffices, automações internas e produtos em validação. Celery fica mais atraente quando você precisa de roteamento avançado, scheduling mais robusto, chords, chains, resultados complexos ou uma equipe já familiarizada com ele. ## Arquitetura recomendada para produção Para uma API FastAPI com jobs em produção, uma arquitetura simples e saudável é: ```text Cliente -> FastAPI -> Banco principal -> Broker Redis/RabbitMQ -> Worker Celery/RQ -> Storage/APIs externas ``` A API deve validar a requisição, salvar o estado inicial no banco e publicar o job. O worker deve atualizar o estado conforme avança. Assim, o frontend pode consultar `/jobs/{id}` sem depender diretamente do backend de resultados do Celery. Um modelo de tabela pode ser simples: ```sql CREATE TABLE jobs ( id UUID PRIMARY KEY, tipo TEXT NOT NULL, status TEXT NOT NULL, usuario_id BIGINT NOT NULL, erro TEXT, criado_em TIMESTAMPTZ NOT NULL DEFAULT now(), atualizado_em TIMESTAMPTZ NOT NULL DEFAULT now() ); ``` Mesmo que o Celery guarde resultado, manter uma tabela de domínio ajuda o produto: você exibe histórico para o usuário, controla permissões e não acopla a interface ao detalhe da fila. ## Boas práticas para jobs Python ### 1. Tarefa idempotente Uma tarefa pode rodar mais de uma vez por retry, timeout ou reprocessamento manual. Planeje para isso. Se o job envia um email, salve um registro com chave única antes do envio ou use um provedor com idempotency key. Se gera arquivo, grave em caminho determinístico ou substitua com segurança. ### 2. Payload pequeno Não coloque arquivo gigante, HTML inteiro ou dataframe serializado dentro da mensagem da fila. Salve o dado em banco ou storage e envie apenas IDs. ```python # Evite gerar_relatorio.delay(csv_completo_em_string) # Prefira gerar_relatorio.delay(upload_id=42) ``` ### 3. Timeouts explícitos Jobs que chamam APIs externas precisam de timeout. Sem isso, um worker pode ficar preso até consumir toda a capacidade da fila. ```python import httpx def consultar_api(url: str) -> dict: with httpx.Client(timeout=10.0) as client: response = client.get(url) response.raise_for_status() return response.json() ``` ### 4. Logs com contexto Inclua `job_id`, `usuario_id`, `tipo` e IDs de entidade nos logs. Em produção, "falhou ao processar" não ajuda se você não sabe qual job falhou. ### 5. Dead letter ou revisão manual Depois de retries, um job pode continuar falhando por dado inválido ou serviço externo fora do ar. Tenha um status como `failed`, uma tela simples de revisão ou pelo menos um comando de reprocessamento. ## Exemplo completo: upload que gera relatório Imagine uma API que recebe uma planilha e gera um relatório em segundo plano. O endpoint não deve processar tudo na hora. ```python from uuid import uuid4 from fastapi import FastAPI, UploadFile from tasks import processar_planilha app = FastAPI() @app.post("/relatorios", status_code=202) async def criar_relatorio(arquivo: UploadFile): job_id = uuid4() caminho = f"/tmp/uploads/{job_id}.csv" conteudo = await arquivo.read() with open(caminho, "wb") as f: f.write(conteudo) # Em produção, salve job_id/status no banco antes de enfileirar. processar_planilha.delay(str(job_id), caminho) return { "job_id": str(job_id), "status": "enfileirado", "acompanhar_em": f"/relatorios/{job_id}", } ``` ```python from celery_app import celery_app @celery_app.task(bind=True, max_retries=3) def processar_planilha(self, job_id: str, caminho: str) -> None: try: # 1. Ler arquivo # 2. Validar colunas # 3. Gerar relatório # 4. Atualizar status no banco print(f"Processando job {job_id} em {caminho}") except Exception as exc: # Atualize status parcial se fizer sentido. raise self.retry(exc=exc, countdown=60) ``` Esse fluxo é mais verboso que `BackgroundTasks`, mas entrega uma fronteira operacional melhor: API responde rápido, worker processa fora, retry é explícito e o usuário pode acompanhar estado. ## Checklist de escolha Antes de implementar, responda: - Se o processo web reiniciar, posso perder essa tarefa? - A tarefa precisa de retry automático? - Quanto tempo ela pode demorar no pior caso? - Quantas tarefas simultâneas podem aparecer? - O usuário precisa acompanhar progresso? - Existe risco financeiro, legal ou de suporte se a tarefa duplicar? Se perder a tarefa é aceitável e ela é leve, `BackgroundTasks` resolve. Se perder não é aceitável, use uma fila externa. Se a operação é crítica, pense também em idempotência, auditoria e reprocessamento. ## Como estudar isso para vaga Python Para quem está montando portfólio ou se preparando para [teste técnico Python](/carreira/teste-tecnico-python/), um bom projeto é criar uma API FastAPI que recebe um CSV, enfileira processamento com Redis e expõe status do job. Isso demonstra backend real: validação, filas, Docker, testes e documentação. Você não precisa começar com Kubernetes ou arquitetura complexa. Um `docker-compose.yml` com API, Redis e worker já mostra maturidade. Depois, conecte com [pytest](/guias/testes-com-pytest/), [Docker](/guias/configurando-docker-python/) e [deploy](/blog/deploy-aplicacao-python/) para fechar o ciclo. ## Conclusão `BackgroundTasks`, Celery e Redis não competem exatamente no mesmo nível. `BackgroundTasks` é conveniência dentro do FastAPI. Redis é infraestrutura de cache, broker ou fila. Celery é um sistema completo de tarefas distribuídas. Comece simples, mas não ignore a natureza da tarefa. Para notificações leves, use o recurso nativo do FastAPI. Para processamento importante, coloque uma fila entre a API e o trabalho pesado. Para sistemas que dependem de jobs para operar, trate filas como parte central do produto, não como detalhe escondido no endpoint.