Django bietet Entwicklern eine Vielzahl von Werkzeugen zur Handhabung von ereignisgesteuerter Logik und asynchroner Verarbeitung. Zwei der wichtigsten Mechanismen sind Django Signals und Celery Tasks. Obwohl beide auf den ersten Blick ähnliche Probleme lösen, unterscheiden sie sich grundlegend in ihrer Architektur und ihren Anwendungsfällen.

Der entscheidende Unterschied: Django Signals sind synchron und prozessintern, während Celery Tasks asynchron in separaten Worker-Prozessen ausgeführt werden.

Was sind Django Signals?

Django Signals implementieren das Observer-Pattern und ermöglichen eine lose Kopplung zwischen verschiedenen Teilen einer Anwendung. Sie erlauben es, bestimmte Aktionen auszulösen, wenn definierte Ereignisse eintreten, ohne dass der Sender den Empfänger kennen muss.

python

from django.db.models.signals import post_save
from django.dispatch import receiver
from django.contrib.auth.models import User
from .models import UserProfile

@receiver(post_save, sender=User)
def create_user_profile(sender, instance, created, **kwargs):
    if created:
        UserProfile.objects.create(user=instance)

Dieses Beispiel zeigt einen klassischen Anwendungsfall: Automatisches Erstellen eines UserProfile-Objekts bei der Registrierung eines neuen Benutzers.

Eingebaute Django Signals

Django bietet mehrere vordefinierte Signals:

python

# Model Signals
from django.db.models.signals import (
    pre_save,
    post_save,
    pre_delete,
    post_delete,
    m2m_changed
)

# Request/Response Signals
from django.core.signals import (
    request_started,
    request_finished
)

# Database Signals
from django.db.backends.signals import connection_created

Eigene Signals definieren

Für spezifische Anwendungsfälle lassen sich eigene Signals erstellen:

python

from django.dispatch import Signal

# Signal definieren
order_completed = Signal()

# Signal senden
class Order:
    def complete(self):
        self.status = "completed"
        self.save()
        order_completed.send(sender=self.__class__, order=self)

# Signal empfangen
@receiver(order_completed)
def send_order_confirmation(sender, order, **kwargs):
    send_email_notification(order.customer.email, order)

Was sind Celery Tasks?

Celery ist eine verteilte Task Queue, die asynchrone Aufgaben in separaten Worker-Prozessen ausführt. Tasks werden in eine Message Queue (Redis, RabbitMQ) eingereiht und von Workern abgearbeitet.

python

from celery import shared_task
from django.core.mail import send_mail

@shared_task
def send_welcome_email(user_id):
    from django.contrib.auth.models import User
    user = User.objects.get(id=user_id)
    send_mail(
        subject="Willkommen!",
        message=f"Hallo {user.username}, willkommen auf unserer Plattform.",
        from_email="noreply@example.com",
        recipient_list=[user.email]
    )

Celery Konfiguration in Django

python

# 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()

# settings.py
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/0"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"

Erweiterte Celery-Funktionen

Celery bietet zahlreiche fortgeschrittene Funktionen:

python

from celery import shared_task
from celery.exceptions import MaxRetriesExceededError

@shared_task(bind=True, max_retries=3, default_retry_delay=60)
def process_payment(self, order_id):
    try:
        order = Order.objects.get(id=order_id)
        result = payment_gateway.charge(order.amount)
        order.payment_status = "completed"
        order.save()
        return result
    except PaymentError as exc:
        try:
            self.retry(exc=exc)
        except MaxRetriesExceededError:
            order.payment_status = "failed"
            order.save()
            raise

# Task mit Zeitplanung
@shared_task
def daily_report():
    generate_and_send_report()

# In celery.py
app.conf.beat_schedule = {
    "daily-report": {
        "task": "myapp.tasks.daily_report",
        "schedule": crontab(hour=8, minute=0),
    },
}

Wann Django Signals verwenden?

Django Signals eignen sich optimal für:

1. Synchrone Nebeneffekte innerhalb des Request-Zyklus

python

@receiver(post_save, sender=Order)
def update_inventory(sender, instance, created, **kwargs):
    if created:
        for item in instance.items.all():
            item.product.stock -= item.quantity
            item.product.save()

2. Lose Kopplung zwischen Django-Apps

python

# In der Payments-App
@receiver(post_save, sender=Payment)
def notify_order_app(sender, instance, **kwargs):
    if instance.status == "successful":
        payment_received.send(sender=sender, payment=instance)

3. Datenkonsistenz bei Model-Änderungen

python

@receiver(pre_save, sender=Article)
def generate_slug(sender, instance, **kwargs):
    if not instance.slug:
        instance.slug = slugify(instance.title)

Wann Celery Tasks verwenden?

Celery ist die richtige Wahl für:

1. Langwierige Operationen

python

@shared_task
def generate_pdf_report(report_id):
    report = Report.objects.get(id=report_id)
    pdf = generate_complex_pdf(report.data)  # Kann Minuten dauern
    report.pdf_file.save(f"report_{report_id}.pdf", ContentFile(pdf))
    report.status = "completed"
    report.save()

2. Externe API-Aufrufe

python

@shared_task(bind=True, max_retries=5)
def sync_with_external_crm(self, customer_id):
    try:
        customer = Customer.objects.get(id=customer_id)
        crm_api.update_contact(customer.to_dict())
    except CRMAPIError as exc:
        self.retry(exc=exc, countdown=2**self.request.retries)

3. Massenverarbeitung

python

from celery import group

@shared_task
def process_single_image(image_id):
    image = Image.objects.get(id=image_id)
    image.create_thumbnails()
    image.optimize()
    image.processed = True
    image.save()

def process_all_images(image_ids):
    job = group(process_single_image.s(id) for id in image_ids)
    result = job.apply_async()
    return result

4. Geplante und wiederkehrende Aufgaben

python

@shared_task
def cleanup_expired_sessions():
    from django.contrib.sessions.models import Session
    from django.utils import timezone
    Session.objects.filter(expire_date__lt=timezone.now()).delete()

Kombination von Signals und Celery

Eine häufige Best Practice ist die Verwendung von Signals als Trigger für Celery Tasks:

python

@receiver(post_save, sender=Order)
def handle_new_order(sender, instance, created, **kwargs):
    if created:
        # Synchron: Kritische Daten aktualisieren
        instance.order_number = generate_order_number()
        instance.save(update_fields=["order_number"])
        
        # Asynchron: Zeitaufwändige Aufgaben auslagern
        send_order_confirmation_email.delay(instance.id)
        notify_warehouse.delay(instance.id)
        sync_with_erp.delay(instance.id)

Bereit für deine Django-Interviews?

Übe mit unseren interaktiven Simulatoren, Flashcards und technischen Tests.

Django erkunden

Vergleichstabelle: Signals vs. Celery

Aspekt	Django Signals	Celery Tasks
Ausführung	Synchron	Asynchron
Prozess	Im gleichen Prozess	Separater Worker
Infrastruktur	Keine zusätzliche	Message Broker erforderlich
Fehlerbehandlung	Exception propagiert	Retry-Mechanismus
Skalierung	Begrenzt	Horizontal skalierbar
Debugging	Einfach	Komplexer
Latenz	Erhöht Request-Zeit	Keine Auswirkung

Häufige Fehler und Anti-Patterns

Signals

python

# FALSCH: Langsame Operationen in Signals
@receiver(post_save, sender=User)
def bad_signal_handler(sender, instance, **kwargs):
    send_email(instance.email)  # Blockiert den Request
    call_external_api(instance)  # Kann fehlschlagen

# RICHTIG: Celery für aufwändige Aufgaben nutzen
@receiver(post_save, sender=User)
def good_signal_handler(sender, instance, created, **kwargs):
    if created:
        send_welcome_email.delay(instance.id)

Celery

python

# FALSCH: Model-Instanzen als Parameter übergeben
@shared_task
def bad_task(user):
    user.send_notification()  # Objekt ist möglicherweise veraltet

# RICHTIG: Nur IDs übergeben
@shared_task
def good_task(user_id):
    user = User.objects.get(id=user_id)
    user.send_notification()

Interview-Fragen und Antworten

Frage 1: Was ist der Hauptunterschied zwischen Django Signals und Celery Tasks?

Django Signals sind synchron und laufen im gleichen Prozess wie der aufrufende Code. Celery Tasks werden asynchron in separaten Worker-Prozessen ausgeführt. Signals eignen sich für schnelle, prozessinterne Reaktionen auf Ereignisse, während Celery für zeitaufwändige, skalierbare Hintergrundaufgaben konzipiert ist.

Frage 2: Warum sollte man in Signal-Handlern keine Objekte, sondern IDs übergeben?

Wenn man Django Signals mit Celery kombiniert und Model-Instanzen übergibt, kann es zu Problemen kommen: Das Objekt muss serialisiert werden, was bei komplexen Objekten fehlschlagen kann. Außerdem könnte das Objekt zum Zeitpunkt der Task-Ausführung bereits veraltet sein. Die Übergabe von IDs und das erneute Laden aus der Datenbank garantiert aktuelle Daten.

Frage 3: Wie verhindert man Endlosschleifen bei Signals?

python

@receiver(post_save, sender=Profile)
def update_profile(sender, instance, **kwargs):
    if not getattr(instance, "_skip_signal", False):
        instance._skip_signal = True
        instance.last_updated = timezone.now()
        instance.save(update_fields=["last_updated"])

Alternativ kann man update_fields in der save()-Methode nutzen oder update() statt save() verwenden.

Frage 4: Wie implementiert man Retry-Logik in Celery Tasks?

Celery bietet eingebaute Retry-Funktionalität:

python

@shared_task(bind=True, max_retries=3, autoretry_for=(ConnectionError,))
def resilient_task(self, data):
    try:
        process(data)
    except TemporaryError as exc:
        raise self.retry(exc=exc, countdown=60)

Frage 5: Wann sollte man eigene Signals definieren statt eingebauter?

Eigene Signals sind sinnvoll, wenn domänenspezifische Ereignisse signalisiert werden sollen, die nicht direkt mit Model-Operationen zusammenhängen. Beispiele: Geschäftsprozesse wie "Bestellung abgeschlossen" oder "Zahlung eingegangen", die von mehreren unabhängigen Komponenten behandelt werden müssen.

Performance-Überlegungen

Signal-Performance

python

# Verbindung nur einmal herstellen
from django.apps import AppConfig

class MyAppConfig(AppConfig):
    def ready(self):
        import myapp.signals  # noqa: F401

Celery-Performance

python

# Task-Routing für bessere Performance
app.conf.task_routes = {
    "myapp.tasks.quick_*": {"queue": "fast"},
    "myapp.tasks.slow_*": {"queue": "slow"},
}

# Prefetch-Multiplikator anpassen
app.conf.worker_prefetch_multiplier = 1  # Für langsame Tasks

Fazit

Django Signals und Celery Tasks sind komplementäre Werkzeuge mit unterschiedlichen Anwendungsbereichen. Signals eignen sich hervorragend für synchrone, prozessinterne Event-Handling-Szenarien mit loser Kopplung. Celery ist die optimale Lösung für asynchrone, skalierbare Hintergrundverarbeitung.

Die beste Architektur kombiniert beide Ansätze: Signals als leichtgewichtige Trigger, die bei Bedarf Celery Tasks für aufwändige Operationen starten. Diese Kombination gewährleistet reaktionsschnelle Anwendungen und robuste Hintergrundverarbeitung.

Django Signals vs. Celery Tasks 2026: Wann man was verwendet und Interviewfragen

Was sind Django Signals?

Eingebaute Django Signals

Eigene Signals definieren

Was sind Celery Tasks?

Celery Konfiguration in Django

Erweiterte Celery-Funktionen

Wann Django Signals verwenden?

Wann Celery Tasks verwenden?

Kombination von Signals und Celery

Bereit für deine Django-Interviews?

Vergleichstabelle: Signals vs. Celery

Häufige Fehler und Anti-Patterns

Signals

Celery

Interview-Fragen und Antworten

Performance-Überlegungen

Signal-Performance

Celery-Performance

Fazit

Findest du den Bug in Django?

Verwandte Artikel

Django und Celery: Asynchrone Aufgabenverarbeitung und Interviewfragen 2026

Django Async Views und ASGI 2026: Performance und Interview-Fragen

Django 5.2: Custom Middleware und Signal-Handling für technische Interviews