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.

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.

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.

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.

@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:

• Check-before-write: Verify state before action

• Idempotency keys: Use unique tokens with external services

• Upsert patterns: INSERT ... ON CONFLICT UPDATE

• Deduplication window: Track processed IDs for N hours

Pattern 4: Job State Management

Persist job state transitions for visibility and debugging.

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,
    )

Advanced Patterns

Pattern 5: Dead Letter Queue

Handle permanently failed tasks for manual inspection.

@app.task(bind=True, max_retries=3) def process_webhook(self, webhook_id: str, payload: dict) -> None: """Process webhook with DLQ for failures.""" try: result = send_webhook(payload) if not result.success: raise WebhookFailedError(result.error) except Exception as e: if self.request.retries >= self.max_retries: # Move to dead letter queue for manual inspection dead_letter_queue.send({ "task": "process_webhook", "webhook_id": webhook_id, "payload": payload, "error": str(e), "attempts": self.request.retries + 1, "failed_at": datetime.utcnow().isoformat(), }) logger.error( "Webhook moved to DLQ after max retries", webhook_id=webhook_id, error=str(e), ) return

    # Exponential backoff retry
    raise self.retry(exc=e, countdown=2 ** self.request.retries * 60)

Pattern 6: Status Polling Endpoint

Provide an endpoint for clients to check job status.

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/jobs/{job_id}") async def get_job_status(job_id: str) -> JobStatusResponse: """Get current status of a background job.""" job = await jobs_repo.get(job_id)

if job is None:
    raise HTTPException(404, f"Job {job_id} not found")

return JobStatusResponse(
    job_id=job.id,
    status=job.status.value,
    created_at=job.created_at,
    started_at=job.started_at,
    completed_at=job.completed_at,
    result=job.result if job.status == JobStatus.SUCCEEDED else None,
    error=job.error if job.status == JobStatus.FAILED else None,
    # Helpful for clients
    is_terminal=job.status in (JobStatus.SUCCEEDED, JobStatus.FAILED),
)

Pattern 7: Task Chaining and Workflows

Compose complex workflows from simple tasks.

from celery import chain, group, chord

Simple chain: A → B → C

workflow = chain( extract_data.s(source_id), transform_data.s(), load_data.s(destination_id), )

Parallel execution: A, B, C all at once

parallel = group( send_email.s(user_email), send_sms.s(user_phone), update_analytics.s(event_data), )

Chord: Run tasks in parallel, then a callback

Process all items, then send completion notification

workflow = chord( [process_item.s(item_id) for item_id in item_ids], send_completion_notification.s(batch_id), )

workflow.apply_async()

Pattern 8: Alternative Task Queues

Choose the right tool for your needs.

RQ (Redis Queue): Simple, Redis-based

from rq import Queue from redis import Redis

queue = Queue(connection=Redis()) job = queue.enqueue(send_email, "user@example.com", "Subject", "Body")

Dramatiq: Modern Celery alternative

import dramatiq from dramatiq.brokers.redis import RedisBroker

dramatiq.set_broker(RedisBroker())

@dramatiq.actor def send_email(to: str, subject: str, body: str) -> None: email_client.send(to, subject, body)

Cloud-native options:

• AWS SQS + Lambda

• Google Cloud Tasks

• Azure Functions

Best Practices Summary

• Return immediately - Don't block requests for long operations

• Persist job state - Enable status polling and debugging

• Make tasks idempotent - Safe to retry on any failure

• Use idempotency keys - For external service calls

• Set timeouts - Both soft and hard limits

• Implement DLQ - Capture permanently failed tasks

• Log transitions - Track job state changes

• Retry appropriately - Exponential backoff for transient errors

• Don't retry permanent failures - Validation errors, invalid credentials

• Monitor queue depth - Alert on backlog growth