Jobs em Segundo Plano e Filas de Tarefas em Python

Desacople trabalhos demorados ou não confiáveis dos ciclos de requisição/resposta. Retorne imediatamente para o usuário enquanto os workers em segundo plano cuidam do trabalho pesado de forma assíncrona.

Quando Usar Esta Habilidade

• Processar tarefas que levam mais de alguns segundos
• Enviar e-mails, notificações ou webhooks
• Gerar relatórios ou exportar dados
• Processar uploads ou transformações de mídia
• Integrar com serviços externos não confiáveis
• Construir arquiteturas orientadas a eventos

Conceitos Principais

1. Padrão de Fila de Tarefas

A API aceita a requisição, enfileira um trabalho e retorna imediatamente com um ID do trabalho. Workers processam os trabalhos de forma assíncrona.

2. Idempotência

Tarefas podem ser retentadas em caso de falha. Projete para reexecução segura.

3. Máquina de Estados do Trabalho

Trabalhos transitam por estados: pendente → em execução → bem-sucedido/falhou.

4. Entrega "Pelo Menos Uma Vez"

A maioria das filas garante a entrega "pelo menos uma vez". Seu código deve lidar com duplicatas.

Início Rápido

Esta habilidade usa o Celery para exemplos, uma fila de tarefas amplamente adotada. Alternativas como RQ, Dramatiq e soluções nativas da nuvem (AWS SQS, GCP Tasks) são escolhas igualmente válidas.

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

@app.task
def send_email(to: str, subject: str, body: str) -> None:
    # This runs in a background worker
    email_client.send(to, subject, body)

# In your API handler
send_email.delay("user@example.com", "Welcome!", "Thanks for signing up")

Padrões Fundamentais

Padrão 1: Retornar ID do Trabalho Imediatamente

Para operações que excedem alguns segundos, retorne um ID do trabalho e processe de forma assíncrona.

from uuid import uuid4
from dataclasses import dataclass
from enum import Enum
from datetime import datetime

class JobStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"

@dataclass
class Job:
    id: str
    status: JobStatus
    created_at: datetime
    started_at: datetime | None = None
    completed_at: datetime | None = None
    result: dict | None = None
    error: str | None = None

# API endpoint
async def start_export(request: ExportRequest) -> JobResponse:
    """Start export job and return job ID."""
    job_id = str(uuid4())

    # Persist job record
    await jobs_repo.create(Job(
        id=job_id,
        status=JobStatus.PENDING,
        created_at=datetime.utcnow(),
    ))

    # Enqueue task for background processing
    await task_queue.enqueue(
        "export_data",
        job_id=job_id,
        params=request.model_dump(),
    )

    # Return immediately with job ID
    return JobResponse(
        job_id=job_id,
        status="pending",
        poll_url=f"/jobs/{job_id}",
    )

Padrão 2: Configuração de Tarefas Celery

Configure tarefas Celery com configurações adequadas de retentativa e tempo limite.

from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

# Global configuration
app.conf.update(
    task_time_limit=3600,          # Hard limit: 1 hour
    task_soft_time_limit=3000,      # Soft limit: 50 minutes
    task_acks_late=True,            # Acknowledge after completion
    task_reject_on_worker_lost=True,
    worker_prefetch_multiplier=1,   # Don't prefetch too many tasks
)

@app.task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
)
def process_payment(self, payment_id: str) -> dict:
    """Process payment with automatic retry on transient errors."""
    try:
        result = payment_gateway.charge(payment_id)
        return {"status": "success", "transaction_id": result.id}
    except PaymentDeclinedError as e:
        # Don't retry permanent failures
        return {"status": "declined", "reason": str(e)}
    except TransientError as e:
        # Retry with exponential backoff
        raise self.retry(exc=e, countdown=2 ** self.request.retries * 60)

Padrão 3: Tornar Tarefas Idempotentes

Workers podem retentar em caso de falha ou tempo limite. Projete para reexecução segura.

@app.task(bind=True)
def process_order(self, order_id: str) -> None:
    """Processa o pedido de forma idempotente."""
    order = orders_repo.get(order_id)

    # Já processado? Retorne cedo
    if order.status == OrderStatus.COMPLETED:
        logger.info("Pedido já processado", order_id=order_id)
        return

    # Já em andamento? Verifique se devemos continuar
    if order.status == OrderStatus.PROCESSING:

…