python-background-jobs

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

Broker note: The localhost broker address shown here is for local development only. In production, load the broker address from configuration, use an authenticated and encrypted broker connection on a private network, keep broker credentials in a secret store, and keep the JSON serializer. Where possible, pass record or user IDs to tasks and look up email addresses and other personal data inside the worker, so personal data stays out of the broker, result backend, and logs.

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

Security notes for this pattern:

• Store the owning user or tenant with each job record when it is created. The status endpoint confirms the caller owns the job before returning its status, result, or error, because a job ID alone is not proof of authorization.
• Validate the export request strictly, with field constraints and size limits, before enqueueing, and enqueue only the fields the task needs.
• Apply per-user rate limits and quotas, since each task may run for up to an hour.
• Workers treat queued parameters as untrusted input and validate them again before use.
• In recent Python versions, prefer timezone-aware timestamps over the naive UTC call shown in these examples.

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)

Payment retry caution: Automatic retry and late acknowledgement are safe only around idempotent operations. A timeout or worker crash can hide a charge that already succeeded, so a payment task is retried only when its charge carries a stable idempotency key, as Pattern 3 shows. The task also checks persisted payment state and returns early when the payment is already charged. Treat this example as illustrative and adapt it before relying on it.

Error and broker notes: Map gateway and task exceptions to safe error codes in task results, and keep full error details in server-side logs only. The broker note from Quick Start applies here as well: the localhost address is for local development, and production uses an authenticated, encrypted broker connection loaded from configuration.

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)

Concurrency note: The status check and the later update above are separate steps, so two workers can process the same order at once. Claim the order first in one atomic database step, such as a conditional update from pending to processing, a row lock, or a distributed lock, and stop or retry when the claim fails. The processing branch shown is a placeholder that needs real handling. The idempotency key protects only the payment call.

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
5. Atomic claim: Claim the record in one conditional update or lock before doing any work

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

Query safety: The elided query above is illustrative only. Check the extra fields against a fixed allowlist of job columns, build the update clause only from those allowlisted column names, and bind every value as a parameter. Keep caller-supplied keys and values out of the query text.

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, especially payment charges that may be retried
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
11. Check job ownership - Confirm the caller owns a job before returning its status or result
12. Protect task data - Use authenticated, encrypted brokers in production and keep personal data and secrets out of task arguments