Template mínimo de API FastAPI com Celery + Redis para processamento pesado assíncrono via Docker.

README.md

Template Básico de API HTTP com Processamento Pesado Assíncrono

Este projeto é um template mínimo para APIs HTTP que precisam executar tarefas pesadas (ex: processamento de vídeo, áudio, dados grandes) sem travar o servidor.

---

Filosofia

Quando uma tarefa demora muito para processar, executar ela diretamente dentro da API bloqueia o servidor e prejudica o atendimento dos usuários.

Solução:

• A API apenas recebe o pedido e dispara a tarefa para um worker separado (processo/serviço).
• O worker executa a tarefa pesada de forma isolada, em background.
• A API permite consultar o status da tarefa.
• Quando a tarefa termina, a API permite baixar o resultado.

---

Como funciona neste template

1. O cliente chama POST /submit/{some_kind_of_id} para enviar a tarefa.
2. A API dispara o processamento pesado no Celery (worker separado).
3. O cliente consulta GET /status/{job_id} para saber se o processamento terminou.
4. Quando o status for SUCCESS, o cliente pode chamar GET /download/{job_id} para obter o resultado (ainda a implementar neste template).

---

Tecnologias usadas

• FastAPI (API HTTP assíncrona)
• Celery (gerenciamento e execução de tarefas assíncronas)
• Redis (broker e backend para Celery)
• Docker + Docker Compose (containerização e orquestração simples)

---

Como usar

1. Instale Docker e Docker Compose.
2. Clone o repositório ou copie os arquivos de código fonte diretamente conforme o necessário.
3. Execute:

docker-compose up --build

4. A API estará disponível em http://localhost:8000.
5. Endpoints principais:

• POST /submit/{some_kind_of_id}: inicia o processamento.
• GET /status/{job_id}: verifica o status da tarefa.
• GET /download/{job_id}: baixa o resultado (placeholder no template).

6. Para parar:

docker-compose down

---

Explicações técnicas

• O serviço API roda o FastAPI e apenas dispara tarefas no Celery via Redis.
• O worker Celery fica escutando tarefas e processa-as separadamente, evitando travar a API.
• O Redis é usado como fila e para armazenar estado/resultados das tarefas.
• O endpoint /health serve para monitoramento simples da API.

---

Considerações

• Escalar múltiplos workers para processar tarefas em paralelo é possível criando vários containers worker.
• O download do resultado ainda precisa ser implementado para casos reais e isso deve ser pensado conforme o caso em questão.
• Para produção, configure autenticação, segurança e monitoramento.

---

Este template serve como inspiração para projetos que precisam desacoplar processamento pesado da API HTTP, mantendo a responsividade e escalabilidade.

celery_worker.py

# celery_worker.py
import time

from celery import Celery

celery_app = Celery(
    "worker",
    broker="redis://redis:6379/0",
    backend="redis://redis:6379/0"
)


@celery_app.task
def fake_heavy_processing(job_id: str):
    print(f"Starting the fake job {job_id}...")
    # Ponto adequado para processamento pesado
    time.sleep(30)  # simulando algo fake
    print(f"{job_id} finished.")

docker-compose.yaml

services:
  api:
    build: .
    volumes:
      - ./app:/app
    ports:
      - "8000:8000"
    depends_on:
      - redis
    healthcheck:
      test: [ "CMD", "curl", "-f", "http://localhost:8000/health" ]
      interval: 10s
      timeout: 5s
      retries: 3

  # TODO: confirm if celery args are ok and enough
  # TODO: confirm if resources definitions are working
  worker:
    build: .
    command: [
      "celery", "-A", "celery_worker", "worker",
      "--loglevel=info",
      "--concurrency=4",
      "--prefetch-multiplier=1",
      "--acks-late"
    ]
    depends_on:
      - redis
    mem_limit: 4g
    cpus: 2.0

  redis:
    image: redis:8.0.0-alpine
    ports:
      - "6379:6379"
    healthcheck:
      test: [ "CMD", "redis-cli", "ping" ]
      interval: 3s
      timeout: 3s
      retries: 5

Dockerfile

FROM python:3.10-slim
WORKDIR /app
COPY requirements.txt .
COPY app/ .
RUN pip install --no-cache-dir -r requirements.txt
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

main.py

# main.py
from celery.result import AsyncResult
from fastapi import FastAPI

from celery_worker import fake_heavy_processing

app = FastAPI()


@app.get("/health")
async def root():
    return {"message": "We are healthy"}


@app.post("/submit/{some_kind_of_id}")
def submit(some_kind_of_id: str):
    try:
        task = fake_heavy_processing.apply_async(args=[some_kind_of_id], task_id=some_kind_of_id)
    except Exception as e:
        return {"error": str(e)}
    return {"job_id": some_kind_of_id, "task_id": task.id}


@app.get("/status/{job_id}")
async def status(job_id: str):
    task_result = AsyncResult(job_id, app=fake_heavy_processing.app)
    return {"status": task_result.status}


@app.get("/download/{job_id}")
async def download(job_id: str):
    return {"job_id": "None"}

requirements.txt

# requirements.txt
fastapi
uvicorn
redis
celery[redis]