Background Jobs

Background Job Patterns

Offload long-running tasks with async job queues.

Overview

• Long-running tasks (report generation, data processing)
• Email/notification sending
• Scheduled/periodic tasks
• Webhook processing
• Data export/import pipelines
• Non-LLM async operations (use LangGraph for LLM workflows)

Tool Selection

ARQ (Async Redis Queue)

Setup

# backend/app/workers/arq_worker.py
from arq import create_pool
from arq.connections import RedisSettings

async def startup(ctx: dict):
    """Initialize worker resources."""
    ctx["db"] = await create_db_pool()
    ctx["http"] = httpx.AsyncClient()

async def shutdown(ctx: dict):
    """Cleanup worker resources."""
    await ctx["db"].close()
    await ctx["http"].aclose()

class WorkerSettings:
    redis_settings = RedisSettings(host="redis", port=6379)
    functions = [
        send_email,
        generate_report,
        process_webhook,
    ]
    on_startup = startup
    on_shutdown = shutdown
    max_jobs = 10
    job_timeout = 300  # 5 minutes

Task Definition

from arq import func

async def send_email(
    ctx: dict,
    to: str,
    subject: str,
    body: str,
) -> dict:
    """Send email task."""
    http = ctx["http"]
    response = await http.post(
        "https://api.sendgrid.com/v3/mail/send",
        json={"to": to, "subject": subject, "html": body},
        headers={"Authorization": f"Bearer {SENDGRID_KEY}"},
    )
    return {"status": response.status_code, "to": to}

async def generate_report(
    ctx: dict,
    report_id: str,
    format: str = "pdf",
) -> dict:
    """Generate report asynchronously."""
    db = ctx["db"]
    data = await db.fetch_report_data(report_id)
    pdf_bytes = await render_pdf(data)
    await db.save_report_file(report_id, pdf_bytes)
    return {"report_id": report_id, "size": len(pdf_bytes)}

Enqueue from FastAPI

from arq import create_pool
from arq.connections import RedisSettings

# Dependency
async def get_arq_pool():
    return await create_pool(RedisSettings(host="redis"))

@router.post("/api/v1/reports")
async def create_report(
    data: ReportRequest,
    arq: ArqRedis = Depends(get_arq_pool),
):
    report = await service.create_report(data)

    # Enqueue background job
    job = await arq.enqueue_job(
        "generate_report",
        report.id,
        format=data.format,
    )

    return {"report_id": report.id, "job_id": job.job_id}

@router.get("/api/v1/jobs/{job_id}")
async def get_job_status(
    job_id: str,
    arq: ArqRedis = Depends(get_arq_pool),
):
    job = Job(job_id, arq)
    status = await job.status()
    result = await job.result() if status == JobStatus.complete else None
    return {"job_id": job_id, "status": status, "result": result}

Celery (Enterprise)

Setup

# backend/app/workers/celery_app.py
from celery import Celery

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

celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    task_track_started=True,
    task_time_limit=600,  # 10 minutes hard limit
    task_soft_time_limit=540,  # 9 minutes soft limit
    worker_prefetch_multiplier=1,  # Fair distribution
    task_acks_late=True,  # Acknowledge after completion
    task_reject_on_worker_lost=True,
)

Task Definition

from celery import shared_task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(
    bind=True,
    max_retries=3,
    default_retry_delay=60,
    autoretry_for=(ConnectionError, TimeoutError),
)
def send_email(self, to: str, subject: str, body: str) -> dict:
    """Send email with automatic retry."""
    try:
        response = requests.post(
            "https://api.sendgrid.com/v3/mail/send",
            json={"to": to, "subject": subject, "html": body},
            headers={"Authorization": f"Bearer {SENDGRID_KEY}"},
            timeout=30,
        )
        response.raise_for_status()
        return {"status": "sent", "to": to}
    except Exception as exc:
        logger.error(f"Email failed: {exc}")
        raise self.retry(exc=exc)

@shared_task(bind=True)
def generate_report(self, report_id: str) -> dict:
    """Long-running report generation."""
    self.update_state(state="PROGRESS", meta={"step": "fetching"})
    data = fetch_report_data(report_id)

    self.update_state(state="PROGRESS", meta={"step": "rendering"})
    pdf = render_pdf(data)

    self.update_state(state="PROGRESS", meta={"step": "saving"})
    save_report(report_id, pdf)

    return {"report_id": report_id, "size": len(pdf)}

Chains and Groups

from celery import chain, group, chord

# Sequential execution
workflow = chain(
    extract_data.s(source_id),
    transform_data.s(),
    load_data.s(destination_id),
)
result = workflow.apply_async()

# Parallel execution
parallel = group(
    process_chunk.s(chunk) for chunk in chunks
)
result = parallel.apply_async()

# Parallel with callback
chord_workflow = chord(
    [process_chunk.s(chunk) for chunk in chunks],
    aggregate_results.s(),
)
result = chord_workflow.apply_async()

Periodic Tasks (Celery Beat)

from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    "cleanup-expired-sessions": {
        "task": "app.workers.tasks.cleanup_sessions",
        "schedule": crontab(minute=0, hour="*/6"),  # Every 6 hours
    },
    "generate-daily-report": {
        "task": "app.workers.tasks.daily_report",
        "schedule": crontab(minute=0, hour=2),  # 2 AM daily
    },
    "sync-external-data": {
        "task": "app.workers.tasks.sync_data",
        "schedule": 300.0,  # Every 5 minutes
    },
}

FastAPI Integration

from fastapi import BackgroundTasks

@router.post("/api/v1/users")
async def create_user(
    data: UserCreate,
    background_tasks: BackgroundTasks,
):
    user = await service.create_user(data)

    # Simple background task (in-process)
    background_tasks.add_task(send_welcome_email, user.email)

    return user

# For distributed tasks, use ARQ/Celery
@router.post("/api/v1/exports")
async def create_export(
    data: ExportRequest,
    arq: ArqRedis = Depends(get_arq_pool),
):
    job = await arq.enqueue_job("export_data", data.dict())
    return {"job_id": job.job_id}

Job Status Tracking

from enum import Enum

class JobStatus(Enum):
    PENDING = "pending"
    STARTED = "started"
    PROGRESS = "progress"
    SUCCESS = "success"
    FAILURE = "failure"
    REVOKED = "revoked"

@router.get("/api/v1/jobs/{job_id}")
async def get_job(job_id: str):
    # Celery
    result = AsyncResult(job_id, app=celery_app)
    return {
        "job_id": job_id,
        "status": result.status,
        "result": result.result if result.ready() else None,
        "progress": result.info if result.status == "PROGRESS" else None,
    }

Anti-Patterns (FORBIDDEN)

# NEVER run long tasks synchronously
@router.post("/api/v1/reports")
async def create_report(data: ReportRequest):
    pdf = await generate_pdf(data)  # Blocks for minutes!
    return pdf

# NEVER lose jobs on failure
@shared_task
def risky_task():
    do_work()  # No retry, no error handling

# NEVER store large results in Redis
@shared_task
def process_file(file_id: str) -> bytes:
    return large_file_bytes  # Store in S3/DB instead!

# NEVER use BackgroundTasks for distributed work
background_tasks.add_task(long_running_job)  # Lost if server restarts

Key Decisions

Related Skills

• langgraph-checkpoints - LLM workflow persistence
• resilience-patterns - Retry and fallback
• observability-monitoring - Job metrics

Capability Details

arq-tasks

Keywords: arq, async queue, redis queue, background task Solves:

• How to run async background tasks in FastAPI?
• Simple Redis job queue

celery-tasks

Keywords: celery, task queue, distributed tasks, worker Solves:

• Enterprise task queue
• Complex job workflows

celery-workflows

Keywords: chain, group, chord, celery workflow Solves:

• Sequential task execution
• Parallel task processing

periodic-tasks

Keywords: periodic, scheduled, cron, celery beat Solves:

• Run tasks on schedule
• Cron-like job scheduling

Background Jobs Implementation Checklist

Planning

Task Identification

• [ ] Identify operations that should be background tasks:
• [ ] Email/notification sending
• [ ] File processing/uploads
• [ ] External API calls
• [ ] Report generation
• [ ] Data aggregation
• [ ] Cleanup operations

• [ ] Categorize by type:
• [ ] Fire-and-forget (no result needed)
• [ ] Result-needed (async query for result)
• [ ] Scheduled/periodic
• [ ] Triggered by events

Library Selection

• [ ] Choose task queue library:
• [ ] ARQ: FastAPI, simple async tasks, Redis only
• [ ] Celery: Complex workflows, multiple brokers
• [ ] Dramatiq: Alternative to Celery, simpler

Implementation

Worker Setup

• [ ] Configure worker settings:

  class WorkerSettings:
      redis_settings = RedisSettings(...)
      functions = [task1, task2]
      max_jobs = 10
      job_timeout = 300

• [ ] Implement lifecycle hooks:
• [ ] on_startup: Initialize connections
• [ ] on_shutdown: Cleanup resources

• [ ] Configure concurrency:
• [ ] Max concurrent jobs
• [ ] Job timeout
• [ ] Queue priorities

Task Definition

• [ ] Define tasks with proper signatures:

  async def my_task(ctx: dict, arg1: str, arg2: int) -> dict:
      ...

• [ ] Add logging to all tasks:

  logger.info("task_started", task="my_task", args={"arg1": arg1})

• [ ] Handle errors appropriately:
• [ ] Catch transient errors and retry
• [ ] Log failures with context
• [ ] Update status on failure

FastAPI Integration

• [ ] Initialize task queue in lifespan:

  app.state.arq = await create_pool(settings)

• [ ] Create dependency for queue access:

  async def get_queue(request: Request) -> ArqRedis:
      return request.app.state.arq

• [ ] Enqueue from routes:

  job = await queue.enqueue_job("task_name", arg1=value)

Reliability

Retry Handling

• [ ] Configure retry settings:
• [ ] max_tries: Maximum retry attempts
• [ ] retry_delay: Delay between retries
• [ ] Exponential backoff if needed

• [ ] Use Retry exception for explicit retries:

  raise Retry(defer=60)  # Retry in 60 seconds

Idempotency

• [ ] Make tasks idempotent (safe to run multiple times):

  if await redis.get(f"processed:{id}"):
      return {"status": "already_processed"}

• [ ] Use idempotency keys for external calls
• [ ] Check state before modifying

Error Handling

• [ ] Handle expected errors gracefully
• [ ] Log unexpected errors with full context
• [ ] Update job/entity status on failure
• [ ] Consider dead letter queue for failures

Monitoring

Metrics

• [ ] Track key metrics:
• [ ] Queue depth (pending jobs)
• [ ] Processing time (p50, p95, p99)
• [ ] Success/failure rate
• [ ] Retry rate

Logging

• [ ] Log at key points:
• [ ] Task started
• [ ] Task completed (with duration)
• [ ] Task failed (with error)
• [ ] Task retrying

Alerting

• [ ] Set up alerts for:
• [ ] High queue depth (jobs backing up)
• [ ] High failure rate
• [ ] Long processing time
• [ ] Worker crashes

Operations

Deployment

• [ ] Separate worker deployment from API
• [ ] Configure worker replicas (horizontal scaling)
• [ ] Health check endpoint for workers
• [ ] Graceful shutdown handling

Docker/Kubernetes

# Worker deployment
spec:
  replicas: 3
  template:
    spec:
      containers:
        - name: worker
          command: ["arq", "app.tasks.worker.WorkerSettings"]
          resources:
            requests:
              memory: "256Mi"
              cpu: "250m"

Scaling

• [ ] Configure auto-scaling based on:
• [ ] Queue depth
• [ ] CPU usage
• [ ] Custom metrics

Testing

Unit Tests

• [ ] Test task logic with mocked dependencies:

  async def test_send_email(mock_ctx):
      result = await send_email(mock_ctx, "to@example.com", "Subject", "Body")
      assert result["status"] == "sent"

Integration Tests

• [ ] Test with real Redis
• [ ] Test retry behavior
• [ ] Test timeout behavior
• [ ] Test scheduled tasks

Load Tests

• [ ] Test worker under load
• [ ] Test queue depth limits
• [ ] Test recovery from failures

Job Status API

• [ ] Implement job status endpoint:

  @router.get("/jobs/{job_id}")
  async def get_job_status(job_id: str):
      ...

• [ ] Return useful status info:
• [ ] Current status (pending, running, completed, failed)
• [ ] Enqueue time
• [ ] Start time
• [ ] Finish time
• [ ] Result (if completed)
• [ ] Error (if failed)

Scheduled Tasks

• [ ] Define cron schedules:

  cron_jobs = [
      cron(cleanup_task, hour=3, minute=0),  # 3 AM daily
      cron(report_task, weekday=0, hour=9),  # Monday 9 AM
  ]

• [ ] Handle missed schedules (run immediately vs skip)
• [ ] Prevent duplicate runs (locking)
• [ ] Log scheduled task execution

Quick Reference

Common Pitfalls

• [ ] Not handling retries: Always configure retry behavior
• [ ] Long-running tasks: Break into smaller tasks or use chunking
• [ ] Missing idempotency: Tasks may run multiple times
• [ ] No monitoring: Add logging and metrics from day one
• [ ] Ignoring timeouts: Configure appropriate timeouts
• [ ] No graceful shutdown: Handle SIGTERM properly

ARQ with FastAPI

Complete guide to integrating ARQ (async Redis queue) with FastAPI.

Setup

Installation

pip install arq redis

Project Structure

backend/
├── app/
│   ├── main.py
│   ├── core/
│   │   └── config.py
│   ├── tasks/
│   │   ├── __init__.py
│   │   ├── worker.py        # Worker settings
│   │   ├── email_tasks.py   # Email tasks
│   │   └── analysis_tasks.py # Analysis tasks
│   └── api/
│       └── routes/

Worker Configuration

# app/tasks/worker.py
from arq import cron
from arq.connections import RedisSettings
from app.core.config import settings

# Task imports
from app.tasks.email_tasks import send_email, send_bulk_emails
from app.tasks.analysis_tasks import process_analysis, cleanup_old_analyses


async def startup(ctx: dict):
    """Worker startup - initialize connections."""
    import redis.asyncio as redis
    from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

    ctx["redis"] = redis.from_url(settings.redis_url)
    ctx["db_engine"] = create_async_engine(settings.database_url)


async def shutdown(ctx: dict):
    """Worker shutdown - cleanup connections."""
    await ctx["redis"].close()
    await ctx["db_engine"].dispose()


class WorkerSettings:
    """ARQ worker settings."""

    redis_settings = RedisSettings.from_dsn(settings.redis_url)

    # Functions available to worker
    functions = [
        send_email,
        send_bulk_emails,
        process_analysis,
        cleanup_old_analyses,
    ]

    # Cron jobs
    cron_jobs = [
        cron(cleanup_old_analyses, hour=3, minute=0),  # Daily at 3 AM
    ]

    # Lifecycle
    on_startup = startup
    on_shutdown = shutdown

    # Concurrency
    max_jobs = 10
    job_timeout = 300  # 5 minutes

    # Retry settings
    max_tries = 3
    retry_delay = 60

Task Definitions

Email Tasks

# app/tasks/email_tasks.py
from arq import Retry
from typing import Any
import structlog

logger = structlog.get_logger()


async def send_email(
    ctx: dict,
    to: str,
    subject: str,
    body: str,
    template: str | None = None,
) -> dict[str, Any]:
    """
    Send a single email.

    Args:
        ctx: Worker context with connections
        to: Recipient email
        subject: Email subject
        body: Email body or template variables
        template: Optional template name
    """
    logger.info("sending_email", to=to, subject=subject)

    try:
        # Use email service
        from app.services.email import EmailService

        email_service = EmailService()

        if template:
            await email_service.send_template(to, template, body)
        else:
            await email_service.send(to, subject, body)

        return {"status": "sent", "to": to}

    except ConnectionError as e:
        # Retry on transient errors
        logger.warning("email_send_failed_retrying", error=str(e))
        raise Retry(defer=60)  # Retry in 60 seconds


async def send_bulk_emails(
    ctx: dict,
    recipients: list[str],
    subject: str,
    body: str,
) -> dict[str, Any]:
    """Send emails to multiple recipients."""
    results = {"sent": 0, "failed": 0, "errors": []}

    for recipient in recipients:
        try:
            await send_email(ctx, recipient, subject, body)
            results["sent"] += 1
        except Exception as e:
            results["failed"] += 1
            results["errors"].append({"email": recipient, "error": str(e)})

    return results

Analysis Tasks

# app/tasks/analysis_tasks.py
from arq import Retry
from datetime import datetime, timedelta, timezone
import structlog

logger = structlog.get_logger()


async def process_analysis(
    ctx: dict,
    analysis_id: str,
) -> dict:
    """
    Process an analysis in the background.

    This is a long-running task that:
    1. Fetches content from URL
    2. Generates embeddings
    3. Calls LLM for analysis
    4. Saves results
    """
    from sqlalchemy.ext.asyncio import AsyncSession
    from app.services.analysis_service import AnalysisService
    from app.infrastructure.repositories import PostgresAnalysisRepository

    logger.info("processing_analysis", analysis_id=analysis_id)

    async with AsyncSession(ctx["db_engine"]) as session:
        repo = PostgresAnalysisRepository(session)
        service = AnalysisService(repo, ctx["redis"])

        try:
            # Update status to processing
            await service.update_status(analysis_id, "processing")

            # Process analysis
            result = await service.process(analysis_id)

            # Update status to completed
            await service.update_status(analysis_id, "completed")

            await session.commit()

            return {
                "status": "completed",
                "analysis_id": analysis_id,
                "artifacts_count": len(result.artifacts),
            }

        except Exception as e:
            logger.exception("analysis_processing_failed", analysis_id=analysis_id)
            await service.update_status(analysis_id, "failed", error=str(e))
            await session.commit()
            raise


async def cleanup_old_analyses(ctx: dict) -> dict:
    """
    Periodic task to cleanup old analyses.

    Runs daily via cron.
    """
    from sqlalchemy.ext.asyncio import AsyncSession
    from sqlalchemy import delete
    from app.infrastructure.models import AnalysisModel

    logger.info("cleanup_old_analyses_started")

    cutoff = datetime.now(timezone.utc) - timedelta(days=30)

    async with AsyncSession(ctx["db_engine"]) as session:
        result = await session.execute(
            delete(AnalysisModel).where(AnalysisModel.created_at < cutoff)
        )
        await session.commit()

        deleted_count = result.rowcount
        logger.info("cleanup_old_analyses_completed", deleted=deleted_count)

        return {"deleted": deleted_count}

FastAPI Integration

Enqueue from Routes

# app/api/v1/routes/analyses.py
from fastapi import APIRouter, Depends, BackgroundTasks
from arq import ArqRedis
from app.tasks.analysis_tasks import process_analysis

router = APIRouter()


async def get_task_queue(request: Request) -> ArqRedis:
    """Get ARQ task queue from app state."""
    return request.app.state.arq


@router.post("/analyses", status_code=201)
async def create_analysis(
    request: AnalyzeRequest,
    queue: ArqRedis = Depends(get_task_queue),
    service: AnalysisService = Depends(get_analysis_service),
) -> AnalyzeCreateResponse:
    """Create analysis and enqueue processing."""
    # Create analysis record
    analysis = await service.create(request.url)

    # Enqueue background processing
    job = await queue.enqueue_job(
        "process_analysis",
        analysis_id=str(analysis.id),
    )

    return AnalyzeCreateResponse(
        analysis_id=str(analysis.id),
        job_id=job.job_id,
        status="pending",
    )


@router.post("/analyses/{analysis_id}/reprocess")
async def reprocess_analysis(
    analysis_id: str,
    queue: ArqRedis = Depends(get_task_queue),
) -> dict:
    """Reprocess a failed analysis."""
    job = await queue.enqueue_job(
        "process_analysis",
        analysis_id=analysis_id,
    )
    return {"job_id": job.job_id, "status": "queued"}

App Setup

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from arq import create_pool
from arq.connections import RedisSettings

from app.core.config import settings


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan with ARQ pool."""
    # Create ARQ connection pool
    app.state.arq = await create_pool(
        RedisSettings.from_dsn(settings.redis_url)
    )

    yield

    # Cleanup
    await app.state.arq.close()


app = FastAPI(lifespan=lifespan)

Running Workers

Development

# Single worker
arq app.tasks.worker.WorkerSettings

# Multiple workers (different terminal for each)
arq app.tasks.worker.WorkerSettings --watch  # Auto-reload

Production (Docker)

# Dockerfile.worker
FROM python:3.13-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .

CMD ["arq", "app.tasks.worker.WorkerSettings"]

# docker-compose.yml
services:
  api:
    build: .
    command: uvicorn app.main:app --host 0.0.0.0 --port 8000

  worker:
    build:
      dockerfile: Dockerfile.worker
    command: arq app.tasks.worker.WorkerSettings
    deploy:
      replicas: 3
    depends_on:
      - redis

  redis:
    image: redis:7-alpine

Job Status Tracking

# app/api/v1/routes/jobs.py
from fastapi import APIRouter, Depends
from arq import ArqRedis
from arq.jobs import Job

router = APIRouter()


@router.get("/jobs/{job_id}")
async def get_job_status(
    job_id: str,
    queue: ArqRedis = Depends(get_task_queue),
) -> dict:
    """Get status of a background job."""
    job = Job(job_id, queue)
    status = await job.status()
    info = await job.info()

    return {
        "job_id": job_id,
        "status": status.value,
        "function": info.function if info else None,
        "enqueue_time": info.enqueue_time.isoformat() if info else None,
        "start_time": info.start_time.isoformat() if info and info.start_time else None,
        "finish_time": info.finish_time.isoformat() if info and info.finish_time else None,
        "result": info.result if info and status.value == "complete" else None,
    }


@router.delete("/jobs/{job_id}")
async def cancel_job(
    job_id: str,
    queue: ArqRedis = Depends(get_task_queue),
) -> dict:
    """Cancel a pending job."""
    job = Job(job_id, queue)
    await job.abort()
    return {"status": "cancelled"}

Testing

# tests/test_tasks.py
import pytest
from unittest.mock import AsyncMock, patch
from app.tasks.email_tasks import send_email


@pytest.fixture
def worker_ctx():
    """Mock worker context."""
    return {
        "redis": AsyncMock(),
        "db_engine": AsyncMock(),
    }


@pytest.mark.asyncio
async def test_send_email(worker_ctx):
    with patch("app.services.email.EmailService") as MockEmailService:
        mock_service = MockEmailService.return_value
        mock_service.send = AsyncMock()

        result = await send_email(
            worker_ctx,
            to="test@example.com",
            subject="Test",
            body="Hello",
        )

        assert result["status"] == "sent"
        mock_service.send.assert_called_once_with(
            "test@example.com", "Test", "Hello"
        )

Task Queue Patterns

Comprehensive guide to background job patterns for Python backends.

Queue Architecture

┌─────────────┐                   ┌─────────────┐
│  FastAPI    │                   │   Redis     │
│  (Producer) │──── Enqueue ─────►│   Queue     │
└─────────────┘                   └──────┬──────┘
                                         │
                                         │ Dequeue
                                         ▼
                                  ┌─────────────┐
                                  │   Worker    │
                                  │  (Consumer) │
                                  └─────────────┘

Task Types

1. Fire and Forget

Task runs asynchronously, caller doesn't wait for result.

# Good for: Emails, notifications, logging, analytics
@task
async def send_welcome_email(user_id: str):
    user = await get_user(user_id)
    await send_email(user.email, "Welcome!")

# Usage
await send_welcome_email.enqueue(user_id="123")

2. Delayed Tasks

Task runs after a specified delay.

# Good for: Reminders, scheduled notifications, cooldowns
@task
async def send_reminder(user_id: str):
    await send_push_notification(user_id, "Don't forget!")

# Usage - run in 1 hour
await send_reminder.enqueue(
    user_id="123",
    _delay=timedelta(hours=1),
)

3. Scheduled/Periodic Tasks

Tasks that run on a schedule (cron-like).

# Good for: Reports, cleanup, syncing, aggregations
@task(schedule="0 0 * * *")  # Daily at midnight
async def generate_daily_report():
    await compute_metrics()
    await send_report_email()

4. Task Chains (Workflows)

Sequential tasks where output feeds into next task.

# Task 1 → Task 2 → Task 3
from celery import chain

workflow = chain(
    download_file.s(url),
    process_file.s(),
    upload_results.s(),
)
result = workflow.apply_async()

5. Task Groups (Fan-out)

Parallel execution of multiple tasks.

# All tasks run in parallel
from celery import group

batch = group([
    process_item.s(item_id)
    for item_id in item_ids
])
results = batch.apply_async()

6. Chord (Fan-out + Callback)

Parallel tasks followed by a callback when all complete.

# Parallel tasks → Single callback
from celery import chord

workflow = chord(
    [analyze_chunk.s(chunk) for chunk in chunks],
    aggregate_results.s(),
)
result = workflow.apply_async()

Reliability Patterns

Retry with Backoff

@task(
    max_retries=3,
    retry_backoff=True,
    retry_backoff_max=600,
)
async def unreliable_task():
    try:
        await call_external_api()
    except TransientError as e:
        raise self.retry(exc=e)

Dead Letter Queue

# Move failed tasks to DLQ for manual review
@task(
    max_retries=3,
    on_failure=move_to_dlq,
)
async def important_task():
    ...

async def move_to_dlq(task_id: str, error: Exception):
    await redis.lpush("dlq:important_task", json.dumps({
        "task_id": task_id,
        "error": str(error),
        "timestamp": datetime.now(timezone.utc).isoformat(),
    }))

Idempotency

@task
async def process_payment(payment_id: str):
    # Check if already processed
    if await redis.get(f"processed:payment:{payment_id}"):
        return {"status": "already_processed"}

    # Process payment
    result = await stripe.process(payment_id)

    # Mark as processed
    await redis.setex(
        f"processed:payment:{payment_id}",
        86400,  # 24 hours
        "1",
    )

    return result

Task Locking

@task
async def singleton_task():
    lock_key = "lock:singleton_task"

    # Try to acquire lock
    if not await redis.set(lock_key, "1", nx=True, ex=300):
        return {"status": "already_running"}

    try:
        await do_work()
    finally:
        await redis.delete(lock_key)

Concurrency Control

Rate Limiting Tasks

from arq import cron

@task(max_concurrent=10)  # Max 10 concurrent instances
async def rate_limited_task():
    await call_api()  # API has rate limit

Priority Queues

# High priority queue
@task(queue="high")
async def urgent_notification():
    ...

# Low priority queue
@task(queue="low")
async def batch_report():
    ...

# Worker configuration
QUEUES = ["high", "default", "low"]  # Priority order

Monitoring

Task States

PENDING → STARTED → SUCCESS
                  → FAILURE
                  → RETRY → STARTED → ...

Metrics to Track

Health Checks

async def check_queue_health():
    queue_depth = await redis.llen("arq:queue")
    oldest_task_age = await get_oldest_task_age()

    return {
        "queue_depth": queue_depth,
        "queue_healthy": queue_depth < 10000,
        "oldest_task_age": oldest_task_age,
        "processing_healthy": oldest_task_age < 300,
    }

Comparison: ARQ vs Celery

When to Use Each

Use ARQ when:

• Building with FastAPI/async
• Simple background tasks
• Redis is already in stack
• Want minimal dependencies

Use Celery when:

• Complex workflows (chains, chords)
• Need RabbitMQ
• Enterprise features needed
• Multiple language workers

Related Files

• See examples/arq-fastapi.md for ARQ integration
• See examples/celery-workflows.md for Celery patterns
• See checklists/background-jobs-checklist.md for implementation checklist