# Install Celery with Redis support
pip install celery[redis]

# Or with RabbitMQ support
pip install celery[librabbitmq]

# Install Redis (for broker and result backend)
pip install redis

# Run Redis with Docker
docker run -d --name redis -p 6379:6379 redis:7-alpine

# celery_app.py — standalone Celery setup
from celery import Celery

app = Celery(
    "myapp",
    broker="redis://localhost:6379/0",      # Where tasks are queued
    backend="redis://localhost:6379/1",     # Where results are stored
    include=["tasks"]                        # Modules containing tasks
)

# Optional configuration
app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="Asia/Kolkata",
    enable_utc=True,
    task_track_started=True,                # Track when task starts running
    task_acks_late=True,                    # Acknowledge after completion
    worker_prefetch_multiplier=1,           # Fair task distribution
    result_expires=3600,                    # Results expire after 1 hour
)

Broker (Redis/RabbitMQ):
→ Stores tasks waiting to be processed
→ Delivers tasks to workers
→ Required component

Result Backend (Redis/Database/etc.):
→ Stores task results after completion
→ Lets you check if a task succeeded
→ Optional — only needed if you need to query results

# tasks.py
from celery_app import app
from celery.utils.log import get_task_logger
import time

logger = get_task_logger(__name__)

# Basic task
@app.task
def add(x, y):
    return x + y

# Task with logging
@app.task(bind=True)
def send_email(self, recipient: str, subject: str, body: str):
    """Send email to a recipient."""
    logger.info(f"Sending email to {recipient}")
    try:
        email_service.send(to=recipient, subject=subject, body=body)
        logger.info(f"Email sent successfully to {recipient}")
        return {"status": "sent", "recipient": recipient}
    except Exception as exc:
        logger.error(f"Email failed: {exc}")
        raise

# Task with configuration
@app.task(
    bind=True,
    name="tasks.process_image",              # Explicit task name
    max_retries=3,                           # Max retry attempts
    default_retry_delay=60,                  # Wait 60s between retries
    time_limit=300,                          # Kill if running > 5 minutes
    soft_time_limit=270,                     # Warn at 4.5 minutes
    queue="media_processing",               # Send to specific queue
    rate_limit="10/m"                        # Max 10 tasks per minute
)
def process_image(self, image_id: int, operations: list):
    """Process an image with specified operations."""
    from images.models import Image

    try:
        image = Image.objects.get(id=image_id)
        for operation in operations:
            image.apply_operation(operation)
        image.save()
        return {"status": "processed", "image_id": image_id}

    except Image.DoesNotExist:
        logger.error(f"Image {image_id} not found")
        return {"status": "error", "message": "Image not found"}

    except Exception as exc:
        logger.warning(f"Image processing failed, retrying: {exc}")
        raise self.retry(exc=exc, countdown=30)  # Retry after 30 seconds

# ─── Calling tasks ───────────────────────────────────

# Delay — most common, runs task asynchronously
result = send_email.delay(
    recipient="user@example.com",
    subject="Welcome!",
    body="Thank you for signing up."
)
print(result.id)      # Task ID: "abc-123-def"

# Apply async — more options
result = process_image.apply_async(
    args=[image_id],
    kwargs={"operations": ["resize", "watermark"]},
    countdown=5,                          # Wait 5 seconds before running
    eta=datetime(2026, 1, 15, 10, 0),    # Run at specific time
    expires=3600,                         # Task expires if not picked up in 1 hour
    queue="media_processing",            # Override default queue
    priority=9                           # High priority (0-9)
)

# Calling synchronously (for testing only!)
result = add.apply(args=[3, 4])
print(result.result)  # 7

# Start a basic worker
celery -A celery_app worker --loglevel=info

# Start worker with multiple processes (one per CPU core)
celery -A celery_app worker --loglevel=info --concurrency=4

# Start worker for specific queue only
celery -A celery_app worker --loglevel=info --queues=media_processing

# Start worker with process pool type
celery -A celery_app worker --pool=prefork       # Default (multiprocessing)
celery -A celery_app worker --pool=gevent        # Green threads (I/O bound)
celery -A celery_app worker --pool=solo          # Single process (debugging)

# Start with autoscaling
celery -A celery_app worker --autoscale=10,3     # Min 3, max 10 processes

# Terminal 1 — General tasks worker
celery -A celery_app worker --queues=default --concurrency=4 --hostname=general@%h

# Terminal 2 — Media processing worker (CPU-intensive)
celery -A celery_app worker --queues=media_processing --concurrency=2 --hostname=media@%h

# Terminal 3 — Email worker
celery -A celery_app worker --queues=emails --concurrency=8 --hostname=email@%h

# Check active workers
celery -A celery_app status

# See active tasks across all workers
celery -A celery_app inspect active

# See reserved (queued, not yet running) tasks
celery -A celery_app inspect reserved

# Revoke (cancel) a task
celery -A celery_app control revoke <task-id> --terminate

# Purge all tasks from queue
celery -A celery_app purge

# myproject/celery.py
import os
from celery import Celery

# Set Django settings module
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")

app = Celery("myproject")

# Load Celery config from Django settings (CELERY_ prefix)
app.config_from_object("django.conf:settings", namespace="CELERY")

# Auto-discover tasks in all installed apps
app.autodiscover_tasks()

# myproject/__init__.py — ensure Celery loads with Django
from .celery import app as celery_app
__all__ = ("celery_app",)

# myproject/settings.py — Celery configuration
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TIMEZONE = "Asia/Kolkata"
CELERY_ENABLE_UTC = True
CELERY_TASK_TRACK_STARTED = True
CELERY_TASK_ACKS_LATE = True
CELERY_WORKER_PREFETCH_MULTIPLIER = 1

# blog/tasks.py — tasks in a Django app
from celery import shared_task
from django.core.mail import send_mail
from django.contrib.auth.models import User

@shared_task(name="blog.tasks.send_welcome_email")
def send_welcome_email(user_id: int):
    """Send welcome email after user registration."""
    try:
        user = User.objects.get(pk=user_id)
        send_mail(
            subject=f"Welcome to FutureTechZone, {user.first_name}!",
            message="Thank you for joining our platform...",
            from_email="noreply@futuretechzone.in",
            recipient_list=[user.email],
            fail_silently=False
        )
        return f"Welcome email sent to {user.email}"
    except User.DoesNotExist:
        return f"User {user_id} not found"

@shared_task(name="blog.tasks.publish_scheduled_posts")
def publish_scheduled_posts():
    """Publish posts scheduled for publication."""
    from blog.models import Post
    from django.utils import timezone

    posts = Post.objects.filter(
        published=False,
        scheduled_at__lte=timezone.now()
    )
    count = posts.update(published=True, published_at=timezone.now())
    return f"Published {count} scheduled posts"

# blog/views.py — calling tasks from Django views
from django.contrib.auth.models import User
from .tasks import send_welcome_email

def register(request):
    if request.method == "POST":
        user = User.objects.create_user(
            username=request.POST["username"],
            email=request.POST["email"],
            password=request.POST["password"]
        )
        # Trigger task asynchronously — view returns immediately
        send_welcome_email.delay(user.id)
        return JsonResponse({"status": "registered", "message": "Check your email!"})