Jetzt ist auch Celery an Bord

Was ist Celery?

Celery ist eine verteilte Aufgaben-Queue. Mit Celery lassen sich asynchron auf verschiedenen Maschinen Aufgaben ausführen und Ergebnisse verarbeiten.

Celery ist komplett in Python implementiert und lässt sich fantastisch mit Django integrieren.

Warum Celery?

Celery bietet die Möglichkeit, Tasks asynchron zu starten. So können zum Beispiel lang laufende Aktionen aus dem Kontext des Django Web-Requests herausgelöst werden und bremsen dessen Verarbeitung dann nicht. Solche Tasks sind beispielsweise der Versand von Benachrichtigungsmails oder die Aktualisierung von Daten über externe Dienste.

Aber auch Aufgaben, die regelmäßig ohne das Zutun eines Users stattfinden, können über Celery abgebildet werden. Das sind beispielsweise Datenbankbereinigungen wie das Löschen abgelaufener Anfragen zur Wiederherstellung von Zugangsdaten. Bei juergen.rocks ersetzen solche "Periodic Tasks" nun die Cron Jobs – und haben damit den großartigen Vorteil, dass sie ohne SSH-Login auf dem Server verwaltbar sind.

Installation mit Redis als Broker

Um asynchron möglicherweise lang laufende Tasks durchzuführen, muss die Durchführung von der eigentlichen Anwendung entkoppelt werden. Diese Aufgabe übernimmt der Broker.

Celery unterstützt eine Vielzahl an Brokern, ich habe mich für Redis entschieden, da ich bereits für andere Zwecke eine Redis-Instanz am Laufen habe und noch ein paar Datenbanken ungenutzt sind. Mit unterschiedlichem Funktionsumfang können auch noch andere Broker wie RabbitMQ, Amazon SQS und Zookeper unterstützt¹.

Was man nun für den weiteren Teil wissen muss, ist, mit welcher URL man an seine Redis-Instanz ran kommt.

• Entweder per TCP/IP: redis://:REDIS_PASSWORT@redis.host:6379/0 – Die Redis-Instanz läuft auf redis.host Port 6379 und die Datenbank Nr. 0 wird verwendet, angemeldet wird sich mit dem Passwort REDIS_PASSWORT.
• Oder per Socket: redis+socket:///var/sock/redis.sock?virtual_host=8 – Die Redis-Instanz ist über den Socket /var/sock/redis.sock erreichbar und die Datenbank Nr. 8 wird verwendet.
• Weitere Möglichkeiten sind in der Dokumentation erläutert².

Innerhalb des bestehenden Python Virtual Environments der Django-Instanz wird nun Celery installiert:

pip install celery[redis]

Damit ist die Grundlage gelegt und die Integration mit Django kann beginnen.

Integration mit Django

Die Integration beginnt mit der Erweiterung der settings.py. Hier werden zentrale Celery-Einstellungen hinterlegt. Alle Einstellungen beginnen mit CELERY_³.

Die wichtigsten Einstellungen sind:

CELERY_BROKER_URL = 'redis+socket:///var/sock/redis.sock?virtual_host=8'
CELERY_RESULT_BACKEND = 'django-db'
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

Die CELERY_BROKER_URL gibt an, welcher Broker unter welcher URL genutzt werden soll.

CELERY_RESULT_BACKEND gibt an, dass die Django-Datenbank für die Ablage der Ergebnisse genutzt werden soll. Dazu wird noch das Paket django-celery-results benötigt, mehr dazu im Folgenden.

CELERY_BEAT_SCHEDULER gibt an, dass der Beat-Scheduler aus dem Paket django-celery-beat verwendet werden soll. Auch dazu mehr im Folgenden. Beats sind wiederkehrende Aufgaben, die nach einem Zeitplan erfolgen. Django Celery Beats ermöglicht die Konfiguration dieser Aufgaben aus dem Django Admin-Bereich heraus.

django-celery-results

Auch hier erfolgt die Installation innerhalb des Python Virtual Environments der Django-Instanz:

pip install django-celery-results

In den INSTALLED_APPS in der settings.py muss 'django_celery_results' ergänzt werden. Dies ist erforderlich, damit die Modelle für Django sichtbar werden.

Anschließend kann man über die manage.py die benötigten Datenbankanpassungen vornehmen lassen:

./manage.py migrate

Damit ist alles getan, damit Celery Ergebnisse in der von Django verwalteten Datenbank ablegen kann.

django-celery-beat

Die Installation von Django Celery Beat erfolgt sehr ähnlich. Zunächst findet auch hier die Paket-Installation innerhalb des Python Virtual Environments der Django-Instanz statt:

pip install django-celery-beat

Auch hier wird die Liste der INSTALLED_APPS in der settings.py erweitert werden, jetzt um 'django_celery_beat'.

Nun werden wiederum die entsprechenden Datenbankanpassungen vorgenommen:

./manage.py migrate

Damit ist die eigentliche Integration abgeschlossen. Wenn man eine eigene Implementierung django.contrib.admin.AdminSite verwendet, könnte es sein, dass man die entsprechenden Admin-Oberflächen erst noch registrieren muss. Vorausgesetzt, die eigene AdminSite heißt admin_site, dann sähe das so aus:

from django_celery_beat.admin import PeriodicTaskAdmin, ClockedScheduleAdmin
from django_celery_beat.models import PeriodicTask, IntervalSchedule, CrontabSchedule, SolarSchedule, ClockedSchedule
from django_celery_results.admin import TaskResultAdmin
from django_celery_results.models import TaskResult


admin_site.register(PeriodicTask, PeriodicTaskAdmin)
admin_site.register(IntervalSchedule)
admin_site.register(CrontabSchedule)
admin_site.register(SolarSchedule)
admin_site.register(ClockedSchedule, ClockedScheduleAdmin)
admin_site.register(TaskResult, TaskResultAdmin)

Django mit Celery ausstatten

Im Package des Django-Projekts (üblicherweise dort, wo die settings.py liegt), legt man nun eine neue Datei celery.py an⁴:

import os

from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_project.settings')

app = Celery('django_project')

app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

Damit die celery.py während der Django-Initialisierung geladen wird, müssen diese Zeilen in der django_project/__init__.py ergänzt werden:

from .celery import app as celery_app

__all__ = ('celery_app',)

Damit ist Celery in der Lage, die Tasks, die später implementiert werden, auch zu finden und auszuführen.

Nun kann die Arbeit mit Celery beginnen.

Celery-Dienst starten

Im Python Virtual Environment der Django-Instanz wird der Celery Worker so gestartet:

celery -A django_project worker --beat --scheduler django --loglevel=info

Hier steht django_project für das Package des Django-Projekts, also üblicherweise das Package, in dem die settings.py zu finden ist. Das Arbeitsverzeichnis für den Start ist dann das Verzeichnis, in dem django-project liegt bzw. der Ordner, in dem die manage.py zu finden ist.

Natürlich bietet sich für die Nutzung in einer produktiven Umgebung eine entsprechende Einbindung in den Systemstart ein.

Management Commands als Tasks bereitstellen

Celery findet in allen INSTALLED_APPS Tasks, die im Package der App im Modul tasks.py liegen und mit dem Decorator @shared_task versehen sind.

Für juergen.rocks bot sich an, eine Reihe von Management Commands als Tasks auszuprägen. Hier ein einfaches Beispiel:

• Die Wetterdaten für die R2NSC-Wetter-Seite⁵ werden über das Management Command updater2nscweather aktualisiert
• Bisher wurde dieses durch einen Cron Job gestartet
• Damit es nun als Celery Task laufen kann, muss man es aus der tasks.py der App, die das Management Command enthält, aufrufen

Entsprechend muss also die tasks.py angepasst werden:

from io import StringIO

from celery import shared_task
from django.core.management import call_command

from weather_app.management.commands import updater2nscweather


@shared_task
def update_r2nsc_weather():
    with StringIO() as out:
        call_command(updater2nscweather.Command(), stdout=out)
        return out.getvalue().strip()

Das Management Command wird nun aufgerufen und die Ausgabe auf Standard Output wird aufgefangen und als Ergebnis des Tasks zurückgegeben, bereinigt um Leerzeichen am Anfang und am Ende. Das ist auch übrigens das, was als Ergebnis abgespeichert wird.

Das Ergebnis wird als SUCCESS gespeichert, wenn keine Exception auftritt. Tritt bei der Ausführung des Management Commands eine Exception auf, wird das Ergebnis als FAILURE gespeichert.

Nachdem die Tasks implementiert sind, sollte man Django und den Celery-Worker neu starten.

Konfiguration im Backend

Im Admin-Bereich können ab sofort Tasks zur zeitgesteuerten Ausführung eingerichtet werden. Das ist weitestgehend selbsterklärend, einfach im Admin-Bereich auf „Periodic Tasks“ klicken.

Alle erkannten Tasks sind in einem Dropdown vorbelegt, so dass man nicht die Tasks manuell erfassen muss.

Bei den Zeitpunkten kann man zwischen Intervallen, Cron-Expressions und noch ein- zwei anderen Möglichkeiten wählen. Außerdem lässt sich wählen, dass ein Task genau ein einziges Mal ausgeführt werden soll („One-Off Task“).

Ergebnisse

Wenn ein Task ausgeführt wurde, kann man das Ergebnis im Admin-Bereich unter „Task results“ betrachten oder auch für weitere Auswertungen verwenden.

Bei periodischen Tasks wächst die Ergebnistabelle schnell an. Man sollte also sicherstellen, dass man hier von Zeit zu Zeit aufräumt, zumindest die erfolgreich abgeschlossenen Ergebnisse.

Dafür bietet sich doch ein Celery-Task an, oder?

Erste Erfahrungen

Celery hat schon in der einfachen Verwendung für wiederkehrende Aufgaben bewiesen, dass es sehr viel Arbeit abnehmen und das Management von Tasks vereinfachen kann.

Die Integration ist sehr einfach, und Redis als Broker ist eine sehr hürdenarme Lösung.

Ich werde zukünftig weitere zeitintensive Operationen aus dem Kontext der Web Requests herausnehmen. So könnte ich diese Aufgaben mit Retry-Möglichkeiten versehen, wenn eine Aktion temporär nicht möglich ist oder zum Fehler führt. Insbesondere alles, was das lokale System verlässt oder auf Dienste angewiesen ist, schreit praktisch danach, auf diesem Weg sinnvoll entkoppelt zu werden.