Background Tasks

Background Tasks

Category: Backend Development
Skill Level: Intermediate
Domain: Asynchronous Processing, Task Queues, Distributed Systems

Overview

Background Tasks are asynchronous operations that execute independently of the main request-response cycle, enabling applications to handle long-running processes, scheduled jobs, and event-driven workflows without blocking user-facing operations or degrading performance.

Description

Background task processing is essential for building responsive applications that can handle computationally intensive operations, external API calls, batch processing, and scheduled maintenance without impacting user experience. By offloading work to asynchronous workers, applications can accept requests quickly, process them in the background, and optionally notify users upon completion through webhooks, polling, or websocket connections.

Task queue systems provide the infrastructure for distributing work across multiple workers, managing retries for failed tasks, and ensuring at-least-once or exactly-once delivery semantics. Celery remains a dominant choice for Python applications, offering robust support for scheduled tasks (beat), task routing, and result storage. For cloud-native environments, managed services like AWS SQS, Google Cloud Tasks, and Azure Queue Storage provide fully-managed queue infrastructure, while serverless platforms enable event-driven task execution through AWS Lambda, Google Cloud Functions, or Azure Functions.

Message brokers serve as the backbone of reliable task queue systems, providing persistence, ordering guarantees, and delivery semantics. RabbitMQ offers flexible routing with exchanges and queues, while Redis provides lightweight in-memory queuing with pub/sub capabilities. Apache Kafka excels at high-throughput event streaming with durability guarantees, making it suitable for event-sourced architectures and real-time data pipelines.

Error handling and retry strategies are critical for production-grade background task systems. Exponential backoff with jitter prevents thundering herd problems during outages, while dead letter queues capture failed tasks for later analysis and replay. Monitoring task latency, throughput, failure rates, and queue depths through metrics and alerts enables proactive identification of bottlenecks and failures before they impact users.

Prerequisites

• Understanding of asynchronous programming concepts
• Familiarity with message queue patterns and broker technologies
• Knowledge of distributed systems challenges (idempotency, consistency)
• Experience with monitoring and observability practices

Core Competencies

• Designing task schemas with idempotency and retry handling
• Implementing task queues using Celery, RQ, or similar frameworks
• Configuring scheduled tasks and cron-like execution
• Setting up dead letter queues and error handling workflows
• Monitoring task health, latency, and throughput
• Scaling worker processes for varying workloads

Implementation

from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Any, Callable, Dict, Optional, List
import json
import uuid
import time
from concurrent.futures import ThreadPoolExecutor

class TaskStatus(Enum):
    PENDING = "pending"
    QUEUED = "queued"
    RUNNING = "running"
    COMPLETED = "completed"
    FAILED = "failed"
    RETRYING = "retrying"

@dataclass
class Task:
    id: str
    name: str
    payload: Dict[str, Any]
    status: TaskStatus = TaskStatus.PENDING
    max_retries: int = 3
    retry_count: int = 0
    retry_delay: int = 60
    created_at: datetime = field(default_factory=datetime.now)
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    result: Any = None
    error: Optional[str] = None

class TaskHandler:
    def get_task_name(self) -> str:
        return self.__class__.__name__.lower().replace("handler", "")
    
    def execute(self, payload: Dict[str, Any]) -> Any:
        raise NotImplementedError

class BackgroundTask:
    def __init__(self, max_workers: int = 4):
        self.max_workers = max_workers
        self.task_queue: List[Task] = []
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.handlers: Dict[str, TaskHandler] = {}
        self.results: Dict[str, Task] = {}
    
    def register_handler(self, handler: TaskHandler):
        self.handlers[handler.get_task_name()] = handler
    
    def enqueue(
        self,
        task_name: str,
        payload: Dict[str, Any],
        priority: int = 0,
        max_retries: Optional[int] = None
    ) -> str:
        task = Task(
            id=str(uuid.uuid4()),
            name=task_name,
            payload=payload,
            max_retries=max_retries or 3
        )
        self.task_queue.append(task)
        return task.id
    
    def process_queue(self):
        while self.task_queue:
            priority, task = self.task_queue.pop(0)
            self._execute_task(task)
    
    def _execute_task(self, task: Task):
        task.status = TaskStatus.RUNNING
        task.started_at = datetime.now()
        self.results[task.id] = task
        
        handler = self.handlers.get(task.name)
        if not handler:
            task.status = TaskStatus.FAILED
            task.error = f"No handler for: {task.name}"
            return
        
        try:
            task.result = handler.execute(task.payload)
            task.status = TaskStatus.COMPLETED
            task.completed_at = datetime.now()
        except Exception as e:
            task.error = str(e)
            task.retry_count += 1
            if task.retry_count >= task.max_retries:
                task.status = TaskStatus.FAILED
            else:
                task.status = TaskStatus.RETRYING
    
    def get_task_status(self, task_id: str) -> Optional[Task]:
        return self.results.get(task_id)

class EmailHandler(TaskHandler):
    def execute(self, payload: Dict[str, Any]) -> Any:
        to = payload.get("to")
        subject = payload.get("subject")
        return {"status": "sent", "to": to, "subject": subject}

task_system = BackgroundTask(max_workers=4)
task_system.register_handler(EmailHandler())
task_id = task_system.enqueue("email", {"to": "user@example.com", "subject": "Welcome!"})

Use Cases

• Sending welcome emails and notifications after user registration
• Processing file uploads, image resizing, and video transcoding
• Aggregating analytics data and generating reports on schedules
• Syncing data with external systems through batch operations
• Running health checks and monitoring probes on schedules
• Cleaning up expired sessions and temporary data

Artifacts

• Celery task configurations and beat schedules
• AWS Lambda functions for serverless task processing
• Kubernetes CronJob manifests
• Task monitoring dashboards (Prometheus/Grafana)
• Dead letter queue handlers and replay scripts

Related Skills

• Message Queues
• Celery
• Scheduled Jobs
• Distributed Systems
• Retry Patterns