# Python Background Jobs

> Patrones de background jobs en Python: colas de tareas, workers y arquitectura event-driven, para procesamiento asíncrono y desacoplar trabajo del ciclo request/response.

Fuente: https://skillsagentes.com/skills/wshobson/agents/python-background-jobs
Markdown: https://skillsagentes.com/skills/wshobson/agents/python-background-jobs.md

Repositorio: https://github.com/wshobson/agents
Autor: wshobson
Licencia: MIT
Actualizado: hace 4 meses
Coste de contexto: 56 tok instalada, 1.8k tok al activarse, 2.7k tok con todos los archivos del bundle
Bundle: 2 archivos, 10 KB
Permisos que pide: ninguno declarado

## Instalación

Un skill son archivos markdown: los mismos archivos valen para cualquier agente y lo único que cambia es el directorio de destino, es decir la bandera `--agent`. Añade `-g` para instalarlo en todos los proyectos de la máquina.

```bash
# Claude Code
npx -y skills add wshobson/agents --skill python-background-jobs --agent claude-code
# Cursor
npx -y skills add wshobson/agents --skill python-background-jobs --agent cursor
# Codex
npx -y skills add wshobson/agents --skill python-background-jobs --agent codex
# Gemini CLI
npx -y skills add wshobson/agents --skill python-background-jobs --agent gemini
# Windsurf
npx -y skills add wshobson/agents --skill python-background-jobs --agent windsurf
# Cline
npx -y skills add wshobson/agents --skill python-background-jobs --agent cline
```

## Qué hace

- Aplica patrones de background jobs en Python: colas de tareas, workers, idempotencia y máquinas de estado
- Muestra cómo devolver un job ID inmediatamente en vez de bloquear la petición
- Configura Celery con timeouts, reintentos con backoff exponencial y task_acks_late
- Define estrategias de idempotencia (check-before-write, idempotency keys, upsert) para reintentos seguros
- Provee un JobRepository para persistir transiciones de estado pending → running → succeeded/failed

## Cuándo usarla

- Al procesar tareas que tardan más de unos segundos
- Al enviar emails, notificaciones o webhooks
- Al generar reportes o exportar datos
- Al integrar con servicios externos poco confiables o construir arquitecturas event-driven

## Qué la activa

- "Necesito procesar exportaciones de datos en segundo plano con Celery"
- "¿Cómo hago idempotente una tarea de procesamiento de pagos?"
- "Quiero devolver un job ID y consultar el estado de una tarea larga"
- "Ayúdame a configurar reintentos con backoff exponencial en Celery"

## Antes de instalar

- Los ejemplos usan Celery con broker Redis, aunque menciona RQ, Dramatiq, AWS SQS o GCP Tasks como alternativas válidas.

## Archivos

- SKILL.md — 7 KB
- references/details.md — 3 KB

## SKILL.md

Reproducido tal cual desde wshobson/agents bajo MIT. Esta sección es el documento original y está en inglés.

# Python Background Jobs & Task Queues

Decouple long-running or unreliable work from request/response cycles. Return immediately to the user while background workers handle the heavy lifting asynchronously.

## When to Use This Skill

- Processing tasks that take longer than a few seconds
- Sending emails, notifications, or webhooks
- Generating reports or exporting data
- Processing uploads or media transformations
- Integrating with unreliable external services
- Building event-driven architectures

## Core Concepts

### 1. Task Queue Pattern

API accepts request, enqueues a job, returns immediately with a job ID. Workers process jobs asynchronously.

### 2. Idempotency

Tasks may be retried on failure. Design for safe re-execution.

### 3. Job State Machine

Jobs transition through states: pending → running → succeeded/failed.

### 4. At-Least-Once Delivery

Most queues guarantee at-least-once delivery. Your code must handle duplicates.

## Quick Start

This skill uses Celery for examples, a widely adopted task queue. Alternatives like RQ, Dramatiq, and cloud-native solutions (AWS SQS, GCP Tasks) are equally valid choices.

```python
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")
```

## Fundamental Patterns

### Pattern 1: Return Job ID Immediately

For operations exceeding a few seconds, return a job ID and process asynchronously.

```python
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}",
    )
```

### Pattern 2: Celery Task Configuration

Configure Celery tasks with proper retry and timeout settings.

```python
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)
```

### Pattern 3: Make Tasks Idempotent

Workers may retry on crash or timeout. Design for safe re-execution.

```python
@app.task(bind=True)
def process_order(self, order_id: str) -> None:
    """Process order idempotently."""
    order = orders_repo.get(order_id)

    # Already processed? Return early
    if order.status == OrderStatus.COMPLETED:
        logger.info("Order already processed", order_id=order_id)
        return

    # Already in progress? Check if we should continue
    if order.status == OrderStatus.PROCESSING:
        # Use idempotency key to avoid double-charging
        pass

    # Process with idempotency key
    result = payment_provider.charge(
        amount=order.total,
        idempotency_key=f"order-{order_id}",  # Critical!
    )

    orders_repo.update(order_id, status=OrderStatus.COMPLETED)
```

**Idempotency Strategies:**

1. **Check-before-write**: Verify state before action
2. **Idempotency keys**: Use unique tokens with external services
3. **Upsert patterns**: `INSERT ... ON CONFLICT UPDATE`
4. **Deduplication window**: Track processed IDs for N hours

### Pattern 4: Job State Management

Persist job state transitions for visibility and debugging.

```python
class JobRepository:
    """Repository for managing job state."""

    async def create(self, job: Job) -> Job:
        """Create new job record."""
        await self._db.execute(
            """INSERT INTO jobs (id, status, created_at)
               VALUES ($1, $2, $3)""",
            job.id, job.status.value, job.created_at,
        )
        return job

    async def update_status(
        self,
        job_id: str,
        status: JobStatus,
        **fields,
    ) -> None:
        """Update job status with timestamp."""
        updates = {"status": status.value, **fields}

        if status == JobStatus.RUNNING:
            updates["started_at"] = datetime.utcnow()
        elif status in (JobStatus.SUCCEEDED, JobStatus.FAILED):
            updates["completed_at"] = datetime.utcnow()

        await self._db.execute(
            "UPDATE jobs SET status = $1, ... WHERE id = $2",
            updates, job_id,
        )

        logger.info(
            "Job status updated",
            job_id=job_id,
            status=status.value,
        )
```

## Detailed worked examples and patterns

Detailed sections (starting with `## Advanced Patterns`) live in `references/details.md`. Read that file when the navigation summary above is insufficient.

## Best Practices Summary

1. **Return immediately** - Don't block requests for long operations
2. **Persist job state** - Enable status polling and debugging
3. **Make tasks idempotent** - Safe to retry on any failure
4. **Use idempotency keys** - For external service calls
5. **Set timeouts** - Both soft and hard limits
6. **Implement DLQ** - Capture permanently failed tasks
7. **Log transitions** - Track job state changes
8. **Retry appropriately** - Exponential backoff for transient errors
9. **Don't retry permanent failures** - Validation errors, invalid credentials
10. **Monitor queue depth** - Alert on backlog growth

## Dónde encaja

- Categoría: [Desarrollo de APIs](https://skillsagentes.com/categorias/desarrollo-apis.md) — Diseña, prueba y documenta APIs HTTP y GraphQL.
- Creador: [wshobson](https://skillsagentes.com/creators/wshobson.md) — 183 skills en el directorio
- [Todas las skills](https://skillsagentes.com/skills.md)
- [Ranking de instalaciones](https://skillsagentes.com/ranking.md)

---

Skills Agentes · [Índice de páginas en markdown](https://skillsagentes.com/sitemap.md) · [Inicio](https://skillsagentes.com/index.md)