Async Jobs

references/

Quick Reference

Quick Start

# Celery task with retry
from celery import shared_task

@shared_task(
    bind=True,
    max_retries=3,
    autoretry_for=(ConnectionError, TimeoutError),
    retry_backoff=True,
)
def process_order(self, order_id: str) -> dict:
    result = do_processing(order_id)
    return {"order_id": order_id, "status": "completed"}

# ARQ task with FastAPI
from arq import create_pool
from arq.connections import RedisSettings

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

@router.post("/api/v1/reports")
async def create_report(data: ReportRequest, arq: ArqRedis = Depends(get_arq_pool)):
    job = await arq.enqueue_job("generate_report", data.report_id)
    return {"job_id": job.job_id}

Configuration

Key Patterns

• JSON serialization with

  task_serializer="json"

  for safety
• Late acknowledgment with

  task_acks_late=True

  to prevent task loss on crash
• Time limits with both

  task_time_limit

  (hard) and

  task_soft_time_limit

  (soft)
• Fair distribution with

  worker_prefetch_multiplier=1

• Reject on lost with

  task_reject_on_worker_lost=True

Key Decisions

task_acks_late=True

Task Routing

Key Patterns

• Named queues for critical/high/default/low/bulk separation
• Redis priority with

  queue_order_strategy: "priority"

  and 0-9 levels
• Task router classes for dynamic routing based on task attributes
• Per-queue workers with tuned concurrency and prefetch settings
• Content-based routing for dynamic workflow dispatch

Key Decisions

x-max-priority

Canvas Workflows

Key Patterns

• Chain for sequential ETL pipelines with result passing
• Group for parallel execution of independent tasks
• Chord for fan-out/fan-in with aggregation callback
• Immutable signatures (

  si()

  ) for steps that ignore input
• Nested workflows combining groups inside chains
• Link error callbacks for workflow-level error handling

Key Decisions

s()

si()

Retry Strategies

Key Patterns

• Exponential backoff with

  retry_backoff=True

  and

  retry_backoff_max

• Jitter with

  retry_jitter=True

  to prevent thundering herd
• Idempotency keys in Redis to prevent duplicate processing
• Dead letter queues for failed tasks requiring manual review
• Task locking to prevent concurrent execution of singleton tasks
• Base task classes with shared retry configuration

Key Decisions

Scheduling

Key Patterns

• Crontab for time-based schedules (daily, weekly, monthly)
• Interval for fixed-frequency tasks (every N seconds)
• Database scheduler with

  django-celery-beat

  for dynamic schedules
• Schedule locks to prevent overlapping long-running scheduled tasks
• Adaptive polling with self-rescheduling tasks

Key Decisions

django-celery-beat

Monitoring

Key Patterns

• Flower dashboard for real-time task monitoring
• Celery signals (

  task_prerun

  ,

  task_postrun

  ,

  task_failure

  ) for metrics
• Health check endpoint verifying broker connection and active workers
• Queue depth monitoring for autoscaling decisions
• Beat monitoring for scheduled task dispatch tracking

Key Decisions

Result Backends

Key Patterns

• Redis backend for task status and small results
• Custom task states (VALIDATING, PROCESSING, UPLOADING) for progress

• update_state()

  for real-time progress reporting
• S3/database for large result storage (never Redis)
• AsyncResult for querying task state and progress

Key Decisions

update_state()

ARQ Patterns

Key Patterns

• Native async/await with

  arq

  for FastAPI integration
• Worker lifecycle with

  startup

  /

  shutdown

  hooks for resource management
• Job enqueue from FastAPI routes with

  enqueue_job()

• Job status tracking with

  Job.status()

  and

  Job.result()

• Delayed tasks with

  _delay=timedelta()

  for deferred execution

Key Decisions

Tool Selection

Anti-Patterns (FORBIDDEN)

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

# NEVER block on results inside tasks (causes deadlock)
@celery_app.task
def bad_task():
    result = other_task.delay()
    return result.get()  # Blocks worker!

# 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 skip idempotency for retried tasks
@celery_app.task(max_retries=3)
def create_order(order):
    Order.create(order)  # Creates duplicates on retry!

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

# NEVER ignore task acknowledgment settings
celery_app.conf.task_acks_late = False  # Default loses tasks on crash

# ALWAYS use immutable signatures in chords
chord([task.s(x) for x in items], callback.si())  # si() prevents arg pollution

Temporal Workflows

Key Patterns

• Workflow definitions with

  @workflow.defn

  and deterministic code
• Saga pattern with compensation for multi-step transactions
• Signals and queries for external interaction with running workflows
• Timers with

  workflow.wait_condition()

  for human-in-the-loop
• Parallel activities via

  asyncio.gather

  inside workflows

Key Decisions

workflow.random()

workflow.now()

Temporal Activities

Key Patterns

• Activity definitions with

  @activity.defn

  for all I/O
• Heartbeating for long-running activities (> 60s)
• Error classification with

  ApplicationError(non_retryable=True)

  for business errors
• Worker configuration with dedicated task queues
• Testing with

  WorkflowEnvironment.start_local()

Key Decisions

start_to_close

Related Skills

• python-backend

  - FastAPI, asyncio, SQLAlchemy patterns

• langgraph

  - LangGraph workflow patterns (use for LLM workflows, not Celery)

• distributed-systems

  - Resilience patterns, circuit breakers

• monitoring-observability

  - Metrics and alerting

Capability Details

celery-config

• Production Celery app configuration
• Broker and backend setup
• Worker tuning and time limits

task-routing

• Premium user task prioritization
• Multi-queue worker deployment
• Dynamic task routing

canvas-workflows

• Complex multi-step task pipelines
• Parallel task execution with aggregation
• Sequential task dependencies

retry-strategies

• Exponential backoff with jitter
• Duplicate prevention for retried tasks
• Failed task handling with DLQ

scheduled-tasks

• Run tasks on schedule (crontab)
• Dynamic schedule management
• Overlap prevention for long tasks

monitoring-health

• Production task monitoring dashboard
• Worker health checks
• Queue depth autoscaling

result-backends

• Task progress tracking with custom states
• Result storage strategies
• Job status API endpoints

arq-patterns

• Lightweight async background tasks for FastAPI
• Simple Redis job queue with async/await
• Job status tracking