Background Jobs & Async Processing

# Client initiates export
POST /api/exports
Content-Type: application/json
{
  "format": "csv",
  "filters": {"date_range": "2024-01-01 to 2024-12-31"}
}

# Server responds immediately with job ID
HTTP/1.1 202 Accepted
Content-Type: application/json
{
  "job_id": "exp_7h2k9m3p4q",
  "status": "pending",
  "status_url": "/api/exports/exp_7h2k9m3p4q"
}

# Client polls for status
GET /api/exports/exp_7h2k9m3p4q

# While processing
HTTP/1.1 200 OK
{
  "job_id": "exp_7h2k9m3p4q",
  "status": "processing",
  "progress": 45,  # Percentage complete
  "created_at": "2024-06-15T10:00:00Z"
}

# When complete
HTTP/1.1 200 OK
{
  "job_id": "exp_7h2k9m3p4q",
  "status": "completed",
  "progress": 100,
  "download_url": "/api/exports/exp_7h2k9m3p4q/download",
  "created_at": "2024-06-15T10:00:00Z",
  "completed_at": "2024-06-15T10:02:30Z"
}

pip install celery[redis] redis

# celery_app.py
from celery import Celery
import time

# Configure Celery
celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",  # Message queue (from Lesson 8)
    backend="redis://localhost:6379/1"  # Result storage
)

# Celery configuration
celery_app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_track_started=True,  # Track when tasks start (for progress)
    task_time_limit=600,      # Kill tasks after 10 minutes
    task_soft_time_limit=540, # Warn tasks at 9 minutes
)

# Example task: Generate large CSV export
@celery_app.task(bind=True, name="generate_csv_export")
def generate_csv_export(self, filters: dict, user_id: int):
    """
    Generate CSV export with progress tracking.
    'bind=True' gives us access to 'self' for updating progress.
    """
    total_rows = 10000  # In reality, query database for count

    for i in range(total_rows):
        # Simulate row processing
        time.sleep(0.001)  # 0.001s per row = 10s total

        # Update progress every 1000 rows
        if i % 1000 == 0:
            progress = int((i / total_rows) * 100)
            self.update_state(
                state="PROCESSING",
                meta={
                    "progress": progress,
                    "current_row": i,
                    "total_rows": total_rows
                }
            )

    # Generate final CSV
    csv_path = f"/tmp/export_{self.request.id}.csv"
    # ... actual CSV generation logic ...

    return {
        "csv_path": csv_path,
        "row_count": total_rows,
        "file_size_mb": 2.4
    }

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from celery.result import AsyncResult
from celery_app import celery_app, generate_csv_export

app = FastAPI()

class ExportRequest(BaseModel):
    filters: dict
    user_id: int

@app.post("/api/exports", status_code=202)
def create_export(request: ExportRequest):
    """
    Create background export job.
    Returns 202 Accepted with job ID immediately.
    """
    # Queue the job (non-blocking)
    task = generate_csv_export.delay(
        filters=request.filters,
        user_id=request.user_id
    )

    return {
        "job_id": task.id,
        "status": "pending",
        "status_url": f"/api/exports/{task.id}"
    }

@app.get("/api/exports/{job_id}")
def get_export_status(job_id: str):
    """
    Check job status and progress.
    Clients poll this endpoint every 2-5 seconds.
    """
    task_result = AsyncResult(job_id, app=celery_app)

    if task_result.state == "PENDING":
        return {"job_id": job_id, "status": "pending", "progress": 0}

    elif task_result.state == "PROCESSING":
        return {
            "job_id": job_id,
            "status": "processing",
            "progress": task_result.info.get("progress", 0),
            "current_row": task_result.info.get("current_row", 0),
            "total_rows": task_result.info.get("total_rows", 0)
        }

    elif task_result.state == "SUCCESS":
        result = task_result.result
        return {
            "job_id": job_id,
            "status": "completed",
            "progress": 100,
            "download_url": f"/api/exports/{job_id}/download",
            "row_count": result["row_count"],
            "file_size_mb": result["file_size_mb"]
        }

    elif task_result.state == "FAILURE":
        return {
            "job_id": job_id,
            "status": "failed",
            "error": str(task_result.info)
        }

    else:
        return {"job_id": job_id, "status": task_result.state.lower()}

# Terminal 1: Start Redis (if not running)
redis-server

# Terminal 2: Start Celery worker
celery -A celery_app worker --loglevel=info

# For production: multiple workers with concurrency
celery -A celery_app worker --loglevel=info --concurrency=4

# On different servers (scale horizontally)
# Server 1:
celery -A celery_app worker --loglevel=info --hostname=worker1@%h

# Server 2:
celery -A celery_app worker --loglevel=info --hostname=worker2@%h

# Terminal 3: Start FastAPI
uvicorn main:app --reload

# Install RQ
pip install rq redis

# tasks.py (RQ version)
import time

def generate_csv_export(filters: dict, user_id: int):
    """Simple RQ task - just a regular Python function"""
    total_rows = 10000

    for i in range(total_rows):
        time.sleep(0.001)
        # Note: RQ doesn't have built-in progress tracking
        # You'd need to store progress in Redis manually

    csv_path = f"/tmp/export_{user_id}.csv"
    # ... CSV generation ...

    return {"csv_path": csv_path, "row_count": total_rows}

# main.py (FastAPI with RQ)
from fastapi import FastAPI
from redis import Redis
from rq import Queue
from rq.job import Job
from tasks import generate_csv_export

app = FastAPI()
redis_conn = Redis(host="localhost", port=6379, db=0)
queue = Queue(connection=redis_conn)

@app.post("/api/exports", status_code=202)
def create_export(filters: dict, user_id: int):
    # Queue job with RQ (much simpler than Celery)
    job = queue.enqueue(
        generate_csv_export,
        filters=filters,
        user_id=user_id,
        job_timeout="10m"  # Kill after 10 minutes
    )

    return {
        "job_id": job.id,
        "status": "pending",
        "status_url": f"/api/exports/{job.id}"
    }

@app.get("/api/exports/{job_id}")
def get_export_status(job_id: str):
    job = Job.fetch(job_id, connection=redis_conn)

    if job.is_finished:
        return {
            "job_id": job_id,
            "status": "completed",
            "result": job.result
        }
    elif job.is_failed:
        return {
            "job_id": job_id,
            "status": "failed",
            "error": str(job.exc_info)
        }
    else:
        return {
            "job_id": job_id,
            "status": job.get_status()  # "queued" or "started"
        }

# Start RQ worker
# rq worker --with-scheduler

# Celery task with detailed progress tracking
@celery_app.task(bind=True, name="process_batch")
def process_batch(self, items: list):
    """
    Process large batch with granular progress updates.
    """
    total = len(items)

    for i, item in enumerate(items):
        try:
            # Process item
            process_single_item(item)

            # Update progress every item (or every N items for huge batches)
            progress = int(((i + 1) / total) * 100)
            self.update_state(
                state="PROCESSING",
                meta={
                    "progress": progress,
                    "current": i + 1,
                    "total": total,
                    "current_item": item.get("id"),
                    "message": f"Processing {i+1} of {total}"
                }
            )

        except Exception as e:
            # Log error but continue processing
            logger.error(f"Failed to process {item}: {e}")

    return {"processed": total, "success": True}

# Client-side polling pattern (JavaScript)
async function pollJobStatus(jobId) {
    while (true) {
        const response = await fetch(`/api/jobs/${jobId}`);
        const data = await response.json();

        if (data.status === "completed") {
            console.log("Job complete!", data.result);
            break;
        } else if (data.status === "failed") {
            console.error("Job failed:", data.error);
            break;
        } else if (data.status === "processing") {
            console.log(`Progress: ${data.progress}%`);
            updateProgressBar(data.progress);
        }

        // Poll every 3 seconds
        await new Promise(resolve => setTimeout(resolve, 3000));
    }
}

# Celery task with retry logic
from celery import Task
from requests.exceptions import RequestException
import random

class BaseTaskWithRetry(Task):
    """Base task class with exponential backoff + jitter"""
    autoretry_for = (RequestException,)  # Auto-retry on network errors
    retry_kwargs = {"max_retries": 5}
    retry_backoff = True  # Enable exponential backoff
    retry_backoff_max = 600  # Max 10 minutes between retries
    retry_jitter = True  # Add random jitter

@celery_app.task(bind=True, base=BaseTaskWithRetry, name="send_webhook")
def send_webhook(self, url: str, payload: dict):
    """
    Send webhook with automatic retries.
    Integrates with Lesson 9 webhook patterns.
    """
    try:
        response = requests.post(
            url,
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=30
        )
        response.raise_for_status()

        return {
            "status": "success",
            "status_code": response.status_code,
            "url": url
        }

    except RequestException as e:
        # Celery auto-retries due to autoretry_for
        # Retry delays: ~10s, ~20s, ~40s, ~80s, ~160s
        logger.warning(f"Webhook failed (attempt {self.request.retries + 1}/5): {e}")
        raise  # Re-raise to trigger retry

    except Exception as e:
        # Non-retryable error (e.g., invalid payload)
        logger.error(f"Permanent webhook failure: {e}")
        raise self.retry(exc=e, countdown=0, max_retries=0)  # Don't retry

# Manual retry with custom logic
@celery_app.task(bind=True, name="fetch_external_data")
def fetch_external_data(self, api_url: str):
    """Manual retry control for complex scenarios"""
    try:
        response = requests.get(api_url, timeout=10)

        if response.status_code == 429:  # Rate limited
            # Retry after longer delay
            retry_after = int(response.headers.get("Retry-After", 60))
            raise self.retry(countdown=retry_after, max_retries=3)

        response.raise_for_status()
        return response.json()

    except RequestException as e:
        if self.request.retries < 5:
            # Exponential backoff: 2^retries * 10 seconds
            countdown = (2 ** self.request.retries) * 10
            raise self.retry(exc=e, countdown=countdown)
        else:
            # Max retries reached - send to dead letter queue
            logger.error(f"Job failed permanently after 5 retries: {e}")
            raise

# Dead letter queue implementation
from celery.signals import task_failure
import json

@task_failure.connect
def handle_task_failure(sender=None, task_id=None, exception=None,
                        args=None, kwargs=None, **other_kwargs):
    """
    Send failed jobs to dead letter queue.
    Triggered when task exhausts all retries.
    """
    dead_letter_data = {
        "task_id": task_id,
        "task_name": sender.name,
        "args": args,
        "kwargs": kwargs,
        "exception": str(exception),
        "timestamp": datetime.now(timezone.utc).isoformat()
    }

    # Store in Redis list for later inspection
    redis_client.lpush("dead_letter_queue", json.dumps(dead_letter_data))

    # Alert ops team
    send_alert(f"Job {task_id} failed permanently: {exception}")

# Retrieve DLQ items for manual inspection
@app.get("/api/admin/dead-letter-queue")
def get_dead_letter_queue():
    """View failed jobs for debugging"""
    dlq_items = redis_client.lrange("dead_letter_queue", 0, 99)
    return [json.loads(item) for item in dlq_items]

# Circuit breaker for external API calls
from datetime import datetime, timedelta

class CircuitBreaker:
    def __init__(self, failure_threshold=5, timeout=60):
        self.failure_threshold = failure_threshold
        self.timeout = timeout  # Seconds before trying again
        self.failures = 0
        self.last_failure_time = None
        self.state = "CLOSED"  # CLOSED = allow, OPEN = block

    def call(self, func, *args, **kwargs):
        # If circuit is OPEN, check if timeout expired
        if self.state == "OPEN":
            if datetime.now() - self.last_failure_time > timedelta(seconds=self.timeout):
                self.state = "HALF_OPEN"  # Try one request
            else:
                raise Exception("Circuit breaker OPEN - service unavailable")

        try:
            result = func(*args, **kwargs)
            # Success - reset failures
            self.failures = 0
            self.state = "CLOSED"
            return result

        except Exception as e:
            self.failures += 1
            self.last_failure_time = datetime.now()

            if self.failures >= self.failure_threshold:
                self.state = "OPEN"
                logger.error(f"Circuit breaker OPEN after {self.failures} failures")

            raise

# Usage in Celery task
external_api_breaker = CircuitBreaker(failure_threshold=5, timeout=300)

@celery_app.task(name="call_external_api")
def call_external_api(endpoint: str, data: dict):
    def api_call():
        return requests.post(f"https://api.example.com/{endpoint}", json=data)

    return external_api_breaker.call(api_call)

# Install SQS transport
pip install celery[sqs] boto3

# celery_app.py with AWS SQS
from celery import Celery
import os

celery_app = Celery(
    "tasks",
    broker="sqs://",  # Use SQS as broker
    backend="redis://localhost:6379/0"  # Still use Redis for results
)

# Configure AWS credentials (or use IAM roles)
celery_app.conf.update(
    broker_transport_options={
        "region": "us-east-1",
        "queue_name_prefix": "myapp-",
        # AWS credentials from environment or IAM role
        "aws_access_key_id": os.getenv("AWS_ACCESS_KEY_ID"),
        "aws_secret_access_key": os.getenv("AWS_SECRET_ACCESS_KEY"),
    },
    task_default_queue="default",
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
)

# Tasks work the same way
@celery_app.task(name="process_upload")
def process_upload(file_path: str):
    # ... processing logic ...
    return {"status": "success"}

# SQS automatically scales, handles durability, retries

# Celery task with webhook callback (integrates Lesson 9)
import requests
import hmac
import hashlib

@celery_app.task(bind=True, name="export_with_callback")
def export_with_callback(self, filters: dict, callback_url: str, webhook_secret: str):
    """
    Generate export and notify client via webhook when done.
    Uses webhook signing from Lesson 9.
    """
    try:
        # Generate export
        csv_path = generate_csv_file(filters)

        # Prepare callback payload
        payload = {
            "job_id": self.request.id,
            "status": "completed",
            "download_url": f"https://api.example.com/exports/{self.request.id}/download",
            "completed_at": datetime.now(timezone.utc).isoformat()
        }

        # Sign webhook (Lesson 9 pattern)
        signature = hmac.new(
            webhook_secret.encode(),
            json.dumps(payload).encode(),
            hashlib.sha256
        ).hexdigest()

        # Send webhook
        response = requests.post(
            callback_url,
            json=payload,
            headers={
                "X-Webhook-Signature": signature,
                "Content-Type": "application/json"
            },
            timeout=10
        )

        if response.status_code != 200:
            logger.warning(f"Webhook callback failed: {response.status_code}")
            # Optionally retry webhook delivery

        return payload

    except Exception as e:
        # Job failed - send failure webhook
        failure_payload = {
            "job_id": self.request.id,
            "status": "failed",
            "error": str(e),
            "failed_at": datetime.now(timezone.utc).isoformat()
        }

        requests.post(callback_url, json=failure_payload, timeout=10)
        raise

# FastAPI endpoint accepting callback URL
@app.post("/api/exports", status_code=202)
def create_export_with_callback(
    filters: dict,
    callback_url: str,
    webhook_secret: str
):
    """
    Create export with webhook callback.
    Client provides callback_url and webhook_secret.
    """
    task = export_with_callback.delay(
        filters=filters,
        callback_url=callback_url,
        webhook_secret=webhook_secret
    )

    return {
        "job_id": task.id,
        "status": "pending",
        "message": "You'll receive a webhook at completion"
    }

# Client's callback endpoint (receives webhook)
@app.post("/webhooks/export-complete")
async def handle_export_webhook(
    request: Request,
    signature: str = Header(alias="X-Webhook-Signature")
):
    """Client implements this endpoint to receive job completion webhooks"""
    body = await request.body()

    # Verify signature (Lesson 9)
    expected_sig = hmac.new(
        WEBHOOK_SECRET.encode(),
        body,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(signature, expected_sig):
        raise HTTPException(status_code=401, detail="Invalid signature")

    data = await request.json()

    if data["status"] == "completed":
        # Download the file
        download_file(data["download_url"])

    return {"status": "received"}

# Install Prometheus client
pip install prometheus-client

# metrics.py - Celery metrics exporter
from prometheus_client import Counter, Histogram, Gauge, generate_latest
from celery.signals import task_prerun, task_postrun, task_failure
import time

# Metrics
task_started = Counter("celery_task_started_total", "Tasks started", ["task_name"])
task_succeeded = Counter("celery_task_succeeded_total", "Tasks succeeded", ["task_name"])
task_failed = Counter("celery_task_failed_total", "Tasks failed", ["task_name"])
task_duration = Histogram("celery_task_duration_seconds", "Task duration", ["task_name"])
queue_length = Gauge("celery_queue_length", "Queue length", ["queue_name"])

# Track task lifecycle
@task_prerun.connect
def task_prerun_handler(sender=None, task_id=None, task=None, **kwargs):
    task_started.labels(task_name=sender.name).inc()

@task_postrun.connect
def task_postrun_handler(sender=None, task_id=None, retval=None, **kwargs):
    task_succeeded.labels(task_name=sender.name).inc()

@task_failure.connect
def task_failure_handler(sender=None, task_id=None, exception=None, **kwargs):
    task_failed.labels(task_name=sender.name).inc()

# Expose metrics endpoint
from fastapi import Response

@app.get("/metrics")
def metrics():
    """Prometheus scrapes this endpoint every 15 seconds"""
    # Update queue length gauge
    inspector = celery_app.control.inspect()
    active_queues = inspector.active_queues() or {}

    for worker, queues in active_queues.items():
        for queue in queues:
            length = len(celery_app.control.inspect().reserved().get(worker, []))
            queue_length.labels(queue_name=queue["name"]).set(length)

    return Response(content=generate_latest(), media_type="text/plain")

# Prometheus scrape config (prometheus.yml)
# scrape_configs:
#   - job_name: 'celery'
#     static_configs:
#       - targets: ['localhost:8000']
#     metrics_path: '/metrics'
#     scrape_interval: 15s

# Install Flower
pip install flower

# Start Flower dashboard
celery -A celery_app flower --port=5555

# Access dashboard at http://localhost:5555
# Features:
# - Real-time task monitoring (pending, active, completed, failed)
# - Worker status (online, offline, CPU, memory)
# - Task details (args, kwargs, result, traceback on failure)
# - Rate limiting and pool stats
# - Restart/shutdown workers remotely

# celery_app.py - Configure scheduled tasks
from celery.schedules import crontab

celery_app.conf.beat_schedule = {
    # Daily report at 8 AM UTC
    "generate-daily-report": {
        "task": "generate_daily_report",
        "schedule": crontab(hour=8, minute=0),
        "args": ("daily",)
    },

    # Cleanup old exports every hour
    "cleanup-old-exports": {
        "task": "cleanup_exports",
        "schedule": crontab(minute=0),  # Every hour at :00
    },

    # Sync external data every 15 minutes
    "sync-external-data": {
        "task": "sync_external_data",
        "schedule": 900.0,  # 900 seconds = 15 minutes
    },

    # Weekly billing run on Monday at 9 AM
    "weekly-billing": {
        "task": "process_weekly_billing",
        "schedule": crontab(hour=9, minute=0, day_of_week=1),
    },
}

# Define the tasks
@celery_app.task(name="generate_daily_report")
def generate_daily_report(report_type: str):
    """Generate and email daily report"""
    # ... report logic ...
    return {"status": "success", "report_type": report_type}

@celery_app.task(name="cleanup_exports")
def cleanup_exports():
    """Delete exports older than 7 days"""
    cutoff = datetime.now() - timedelta(days=7)
    deleted_count = delete_old_files("/tmp/exports", cutoff)
    return {"deleted": deleted_count}

@celery_app.task(name="sync_external_data")
def sync_external_data():
    """Sync data from external API"""
    # ... sync logic ...
    return {"synced_records": 1234}

# Start Celery Beat scheduler (separate process)
# celery -A celery_app beat --loglevel=info

# Beat scheduler queues tasks at scheduled times
# Workers execute them as normal background jobs