background-tasks

background-tasks

Covers async task execution: Celery + beat for scheduled/queued work, APScheduler for in-process cron, and bare Redis queues for lightweight pipelines.

1) Celery setup (Python)

# celery_app.py
from celery import Celery

celery = Celery(
    "tasks",
    broker="redis://localhost:6379/1",    # DB 1 for queues (never evicted)
    backend="redis://localhost:6379/2",   # DB 2 for results
    include=["app.tasks"],
)

celery.conf.update(
    task_serializer="json",
    result_expires=3600,
    timezone="UTC",
    enable_utc=True,
    worker_prefetch_multiplier=1,   # process one task at a time (prevents memory spike)
    task_acks_late=True,            # ack after success, not on receive (safe retry on crash)
)

2) Defining tasks

from celery_app import celery

@celery.task(bind=True, max_retries=3, default_retry_delay=60)
def generate_article(self, domain: str, slug: str) -> dict:
    try:
        result = run_pipeline(domain, slug)
        return result
    except TemporaryError as exc:
        raise self.retry(exc=exc)     # exponential back-off via default_retry_delay
    except PermanentError:
        # Don't retry — log and fail cleanly
        logger.error("permanent_failure", domain=domain, slug=slug)
        return {"status": "failed"}

# Enqueue
task = generate_article.delay(domain="keto", slug="keto-diet-guide")
print(task.id)            # track this ID

3) Celery Beat (scheduled tasks)

from celery.schedules import crontab

celery.conf.beat_schedule = {
    "serp-delta-check": {
        "task":     "app.tasks.run_serp_delta",
        "schedule": crontab(hour="*/6"),      # every 6h
    },
    "link-score-refresh": {
        "task":     "app.tasks.refresh_link_scores",
        "schedule": crontab(hour=3, minute=0),  # 3am UTC
    },
}

4) Docker Compose: worker + beat

services:
  worker-default:
    build: .
    command: celery -A celery_app worker -Q default -c 2 --loglevel=info
    environment:
      - REDIS_URL=redis://redis:6379/1
    depends_on: [redis]
    restart: unless-stopped

  worker-scraping:
    build: .
    command: celery -A celery_app worker -Q scraping -c 1 --loglevel=info
    depends_on: [redis]
    restart: unless-stopped

  beat:
    build: .
    command: celery -A celery_app beat --loglevel=info --scheduler celery.beat.PersistentScheduler
    depends_on: [redis]
    restart: unless-stopped