Celery mit Django: Task Queues richtig einsetzen

Lange HTTP-Requests, E-Mail-Versand, PDF-Generierung — all das gehört nicht in einen synchronen View. Celery ist die Standardlösung in der Django-Welt. Ich nutze es seit über 10 Jahren in Produktionsprojekten.

Warum Celery?

Ohne Task Queue blockieren aufwändige Operationen den Web-Worker und verschlechtern die Antwortzeiten. Celery entkoppelt diese Aufgaben: Der View gibt die Aufgabe an die Queue ab und antwortet sofort. Typische Anwendungsfälle:

• E-Mails und Benachrichtigungen versenden
• PDF-Dokumente generieren
• Externe APIs synchronisieren
• Rechenintensive Auswertungen (NumPy, Pandas)
• Periodische Tasks (Cron-Ersatz)

Installation und Konfiguration

pip install celery redis django-celery-results

# settings.py
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "django-db"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_TIMEZONE = "Europe/Berlin"

Erstelle eine celery.py Datei im Projektverzeichnis:

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

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
app = Celery("myproject")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

# myproject/__init__.py
from .celery import app as celery_app
__all__ = ("celery_app",)

Tasks erstellen und aufrufen

# myapp/tasks.py
from celery import shared_task
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def send_welcome_email(self, user_id):
    from django.contrib.auth import get_user_model
    User = get_user_model()
    try:
        user = User.objects.get(pk=user_id)
        # E-Mail-Logik hier...
        logger.info(f"E-Mail gesendet an {user.email}")
    except Exception as exc:
        raise self.retry(exc=exc)  # Auto-Retry nach 60s

Tasks aus einem View aufrufen:

# Sofort asynchron ausfuehren
send_welcome_email.delay(user.id)

# Mit Optionen (Countdown, Ablaufzeit)
send_welcome_email.apply_async(
    args=[user.id],
    countdown=10,    # 10 Sekunden warten
    expires=3600,   # Task nach 1h verwerfen
)

Workflows: Chains, Groups, Chords

from celery import chain, group, chord

# Chain: Tasks sequenziell (Ergebnis wird weitergegeben)
pipeline = chain(
    fetch_data.s(source_id),
    process_data.s(),
    generate_report.s()
)

# Group: Tasks parallel ausfuehren
parallel = group(process_chunk.s(c) for c in data_chunks)

# Chord: Parallel, dann Callback wenn alle fertig
workflow = chord(parallel)(merge_results.s())
workflow.delay()

Monitoring mit Flower

pip install flower

# Web-UI auf Port 5555
celery -A myproject flower --port=5555

# Mit Basic Auth absichern
celery -A myproject flower --basic_auth=admin:geheimespasswort

Flower zeigt dir in Echtzeit: aktive Worker, laufende Tasks, Erfolgsrate und Queue-Längen. Unverzichtbar in Produktion.

Best Practices aus der Praxis

• Importiere Models immer innerhalb des Tasks — nie auf Modulebene (Circular-Import-Probleme)
• Übergib immer IDs, nie Objekte — Django-Objekte lassen sich nicht serialisieren; übergib user_id statt user
• Setze immer max_retries — ohne Limit können fehlerhafte Tasks endlos wiederholt werden
• Nutze bind=True für Zugriff auf die Task-Instanz und Retry-Logik
• Separate Queues für unterschiedliche Task-Typen (schnell vs. langsam, wichtig vs. normal)

Yevhen Chubchyk

Senior Python / Django Entwickler · Freelancer seit 2016 · 20+ Jahre IT-Erfahrung. Kunden: Siemens Healthineers, Engie, natureOffice u.a.