Asynchrone taken met Django en Celery

Toen ik nieuw was bij Django, was een van de meest frustrerende dingen die ik ervoer de noodzaak om periodiek een beetje code uit te voeren. Ik schreef een leuke functie die een actie uitvoerde die dagelijks om 12.00 uur moest worden uitgevoerd. Makkelijk, toch? Mis. Dit bleek een enorm probleem voor mij, aangezien ik in die tijd gewend was aan "Cpanel-type" webhosting waar er een mooie handige GUI was om cron-jobs voor dit doel in te stellen.

Na veel onderzoek vond ik een mooie oplossing:Celery, een krachtige asynchrone taakwachtrij die wordt gebruikt om taken op de achtergrond uit te voeren. Maar dit leidde tot extra problemen, omdat ik geen gemakkelijke set instructies kon vinden om Celery in een Django-project te integreren.

Natuurlijk lukte het me uiteindelijk om erachter te komen, wat dit artikel zal behandelen:Hoe Celery te integreren in een Django-project en periodieke taken te creëren.

Gratis bonus: Klik hier om toegang te krijgen tot een gratis Django Learning Resources Guide (PDF) met tips en trucs en veelvoorkomende valkuilen die je moet vermijden bij het bouwen van Python + Django-webapplicaties.

Dit project maakt gebruik van Python 3.4, Django 1.8.2, Celery 3.1.18 en Redis 3.0.2.

Overzicht

Voor uw gemak, aangezien dit zo'n groot bericht is, verwijzen wij u naar deze tabel voor korte informatie over elke stap en om de bijbehorende code te pakken.

Wat is selderij?

“Celery is een asynchrone taakwachtrij/taakwachtrij op basis van gedistribueerde berichtenoverdracht. Het is gericht op realtime werking, maar ondersteunt ook planning.” Voor dit bericht zullen we ons concentreren op de planningsfunctie om periodiek een taak/taak uit te voeren.

Waarom is dit nuttig?

• Denk aan alle keren dat je in de toekomst een bepaalde taak hebt moeten uitvoeren. Misschien moest u elk uur toegang krijgen tot een API. Of misschien moest u aan het eind van de dag een reeks e-mails verzenden. Groot of klein, Celery maakt het plannen van dergelijke periodieke taken eenvoudig.
• Je wilt nooit dat eindgebruikers onnodig hoeven te wachten tot pagina's zijn geladen of acties zijn voltooid. Als een lang proces deel uitmaakt van de workflow van uw applicatie, kunt u Celery gebruiken om dat proces op de achtergrond uit te voeren, zodra resources beschikbaar komen, zodat uw applicatie kan blijven reageren op verzoeken van klanten. Dit houdt de taak buiten de context van de toepassing.

Instellen

Voordat je in Celery duikt, pak je het startersproject uit de Github-repo. Zorg ervoor dat u een virtualenv activeert, de vereisten installeert en de migraties uitvoert. Start vervolgens de server en navigeer naar http://localhost:8000/ in uw browser. U zou de bekende tekst "Gefeliciteerd met uw eerste door Django aangedreven pagina" moeten zien. Als je klaar bent, dood je de server.

Laten we vervolgens Celery installeren met pip:

$ pip install celery==3.1.18
$ pip freeze > requirements.txt

Nu kunnen we Celery in slechts drie eenvoudige stappen in ons Django-project integreren.

Stap 1:celery.py toevoegen

Maak in de map "picha" een nieuw bestand met de naam celery.py :

from __future__ import absolute_import
import os
from celery import Celery
from django.conf import settings

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'picha.settings')
app = Celery('picha')

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

Let op de opmerkingen in de code.

Stap 2:Importeer je nieuwe Celery-app

Om ervoor te zorgen dat de Celery-app wordt geladen wanneer Django start, voegt u de volgende code toe aan de __init__.py bestand dat naast uw settings.py . staat bestand:

from __future__ import absolute_import

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

Als je dat hebt gedaan, zou je projectlay-out er nu als volgt uit moeten zien:

├── manage.py
├── picha
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
└── requirements.txt

Stap 3:Installeer Redis als een Celery “Broker”

Celery gebruikt 'makelaars' om berichten door te geven tussen een Django-project en de medewerkers van Celery. In deze zelfstudie gebruiken we Redis als de berichtenmakelaar.

Installeer eerst Redis vanaf de officiële downloadpagina of via brew (brew install redis ) en ga dan naar uw terminal, in een nieuw terminalvenster, start de server op:

$ redis-server

U kunt testen of Redis goed werkt door dit in uw terminal te typen:

$ redis-cli ping

Redis zou moeten antwoorden met PONG - probeer het!

Zodra Redis actief is, voegt u de volgende code toe aan uw settings.py-bestand:

# CELERY STUFF
BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Africa/Nairobi'

Je moet ook Redis toevoegen als een afhankelijkheid in het Django-project:

$ pip install redis==2.10.3
$ pip freeze > requirements.txt

Dat is het! Je zou nu Celery met Django moeten kunnen gebruiken. Raadpleeg de officiële Celery-documentatie voor meer informatie over het instellen van Celery met Django.

Laten we, voordat we verder gaan, een paar gezondheidschecks uitvoeren om er zeker van te zijn dat alles in orde is...

Test of de medewerker van Celery klaar is om taken te ontvangen:

$ celery -A picha worker -l info
...
[2015-07-07 14:07:07,398: INFO/MainProcess] Connected to redis://localhost:6379//
[2015-07-07 14:07:07,410: INFO/MainProcess] mingle: searching for neighbors
[2015-07-07 14:07:08,419: INFO/MainProcess] mingle: all alone

Dood het proces met CTRL-C. Test nu of de Celery-taakplanner klaar is voor actie:

$ celery -A picha beat -l info
...
[2015-07-07 14:08:23,054: INFO/MainProcess] beat: Starting...

Boem!

Nogmaals, stop het proces als je klaar bent.

Selderietaken

Celery maakt gebruik van taken, die kunnen worden gezien als normale Python-functies die worden aangeroepen met Celery.

Laten we bijvoorbeeld van deze basisfunctie een Celery-taak maken:

def add(x, y):
    return x + y

Voeg eerst een decorateur toe:

from celery.decorators import task

@task(name="sum_two_numbers")
def add(x, y):
    return x + y

Dan kunt u deze taak als volgt asynchroon uitvoeren met Celery:

add.delay(7, 8)

Simpel, toch?

Dit soort taken is dus perfect voor wanneer u een webpagina wilt laden zonder de gebruiker te laten wachten tot een achtergrondproces is voltooid.

Laten we een voorbeeld bekijken...

Ga terug naar het Django-project en pak versie drie, die een app bevat die feedback van gebruikers accepteert, toepasselijk feedback genoemd :

├── feedback
│   ├── __init__.py
│   ├── admin.py
│   ├── emails.py
│   ├── forms.py
│   ├── models.py
│   ├── tests.py
│   └── views.py
├── manage.py
├── picha
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── requirements.txt
└── templates
    ├── base.html
    └── feedback
        ├── contact.html
        └── email
            ├── feedback_email_body.txt
            └── feedback_email_subject.txt

Installeer de nieuwe vereisten, start de app en ga naar http://localhost:8000/feedback/. Je zou moeten zien:

Laten we de Celery-taak afspreken.

Voeg de taak toe

Kortom, nadat de gebruiker het feedbackformulier heeft ingediend, willen we hem onmiddellijk zijn vrolijke gang laten gaan terwijl we de feedback verwerken, een e-mail verzenden, enz., Allemaal op de achtergrond.

Om dit te bereiken, voegt u eerst een bestand toe met de naam tasks.py naar de map "feedback":

from celery.decorators import task
from celery.utils.log import get_task_logger

from feedback.emails import send_feedback_email

logger = get_task_logger(__name__)


@task(name="send_feedback_email_task")
def send_feedback_email_task(email, message):
    """sends an email when feedback form is filled successfully"""
    logger.info("Sent feedback email")
    return send_feedback_email(email, message)

Update vervolgens forms.py zoals zo:

from django import forms
from feedback.tasks import send_feedback_email_task


class FeedbackForm(forms.Form):
    email = forms.EmailField(label="Email Address")
    message = forms.CharField(
        label="Message", widget=forms.Textarea(attrs={'rows': 5}))
    honeypot = forms.CharField(widget=forms.HiddenInput(), required=False)

    def send_email(self):
        # try to trick spammers by checking whether the honeypot field is
        # filled in; not super complicated/effective but it works
        if self.cleaned_data['honeypot']:
            return False
        send_feedback_email_task.delay(
            self.cleaned_data['email'], self.cleaned_data['message'])

In wezen is de send_feedback_email_task.delay(email, message) functie verwerkt en verzendt de feedback-e-mail op de achtergrond terwijl de gebruiker de site blijft gebruiken.

OPMERKING :De success_url in views.py is ingesteld om de gebruiker om te leiden naar / , die nog niet bestaat. We zullen dit eindpunt in de volgende sectie instellen.

Periodieke taken

Vaak moet u een taak plannen om af en toe op een specifiek tijdstip uit te voeren, dat wil zeggen dat een webschraper bijvoorbeeld dagelijks moet worden uitgevoerd. Dergelijke taken, periodieke taken genoemd, zijn eenvoudig in te stellen met Celery.

Selderij gebruikt "selderijritme" om periodieke taken in te plannen. Celery beat voert taken uit met regelmatige tussenpozen, die vervolgens worden uitgevoerd door selderijwerkers.

De volgende taak is bijvoorbeeld gepland om elk kwartier te worden uitgevoerd:

from celery.task.schedules import crontab
from celery.decorators import periodic_task


@periodic_task(run_every=(crontab(minute='*/15')), name="some_task", ignore_result=True)
def some_task():
    # do something

Laten we een robuuster voorbeeld bekijken door deze functionaliteit toe te voegen aan het Django-project...

Terug naar het Django-project, pak versie vier, die een andere nieuwe app bevat, genaamd photos , die de Flickr API gebruikt om nieuwe foto's te krijgen voor weergave op de site:

├── feedback
│   ├── __init__.py
│   ├── admin.py
│   ├── emails.py
│   ├── forms.py
│   ├── models.py
│   ├── tasks.py
│   ├── tests.py
│   └── views.py
├── manage.py
├── photos
│   ├── __init__.py
│   ├── admin.py
│   ├── models.py
│   ├── settings.py
│   ├── tests.py
│   ├── utils.py
│   └── views.py
├── picha
│   ├── __init__.py
│   ├── celery.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── requirements.txt
└── templates
    ├── base.html
    ├── feedback
    │   ├── contact.html
    │   └── email
    │       ├── feedback_email_body.txt
    │       └── feedback_email_subject.txt
    └── photos
        └── photo_list.html

Installeer de nieuwe vereisten, voer de migraties uit en start vervolgens de server om te controleren of alles in orde is. Probeer het feedbackformulier opnieuw uit. Deze keer zou het prima moeten omleiden.

Wat nu?

Omdat we de Flickr-API regelmatig moeten aanroepen om meer foto's aan onze site toe te voegen, kunnen we een Celery-taak toevoegen.

Voeg de taak toe

Voeg een tasks.py toe naar de photos app:

from celery.task.schedules import crontab
from celery.decorators import periodic_task
from celery.utils.log import get_task_logger

from photos.utils import save_latest_flickr_image

logger = get_task_logger(__name__)


@periodic_task(
    run_every=(crontab(minute='*/15')),
    name="task_save_latest_flickr_image",
    ignore_result=True
)
def task_save_latest_flickr_image():
    """
    Saves latest image from Flickr
    """
    save_latest_flickr_image()
    logger.info("Saved image from Flickr")

Hier voeren we de save_latest_flickr_image() functie elke vijftien minuten door de functieaanroep in een task te verpakken . De @periodic_task decorateur abstraheert de code om de Celery-taak uit te voeren, en laat de tasks.py bestand schoon en gemakkelijk te lezen!

Lokaal hardlopen

Klaar om dit ding te gebruiken?

Open twee nieuwe terminalvensters/tabbladen terwijl uw Django-app en Redis actief zijn. Navigeer in elk nieuw venster naar uw projectdirectory, activeer uw virtualenv en voer de volgende opdrachten uit (één in elk venster):

$ celery -A picha worker -l info
$ celery -A picha beat -l info

Als je de site op http://127.0.0.1:8000/ bezoekt, zou je nu één afbeelding moeten zien. Onze app krijgt elke 15 minuten één afbeelding van Flickr:

Bekijk photos/tasks.py om de code te zien. Door op de knop "Feedback" te klikken, kunt u... wat feedback sturen:

Dit werkt via een bleekselderij. Bekijk feedback/tasks.py voor meer.

Dat is alles, u heeft het Picha-project in de lucht!

Dit is goed om te testen terwijl u uw Django-project lokaal ontwikkelt, maar werkt niet zo goed wanneer u het moet implementeren in productie, zoals op DigitalOcean misschien. Daarvoor is het aan te raden om de Celery worker en scheduler op de achtergrond te draaien als een daemon met Supervisor.

Op afstand hardlopen

Installatie is eenvoudig. Pak versie vijf uit de repo (als je die nog niet hebt). SSH vervolgens naar uw externe server en voer het volgende uit:

$ sudo apt-get install supervisor

We moeten de Supervisor dan informeren over onze Celery-werknemers door configuratiebestanden toe te voegen aan de map "/etc/supervisor/conf.d/" op de externe server. In ons geval hebben we twee van dergelijke configuratiebestanden nodig:een voor de Celery-werker en een voor de Celery-planner.

Maak lokaal een map met de naam "supervisor" in de hoofdmap van het project. Voeg dan de volgende bestanden toe...

Selderiewerker:picha_celery.conf

; ==================================
;  celery worker supervisor example
; ==================================

; the name of your supervisord program
[program:pichacelery]

; Set full path to celery program if using virtualenv
command=/home/mosh/.virtualenvs/picha/bin/celery worker -A picha --loglevel=INFO

; The directory to your Django project
directory=/home/mosh/sites/picha

; If supervisord is run as the root user, switch users to this UNIX user account
; before doing any processing.
user=mosh

; Supervisor will start as many instances of this program as named by numprocs
numprocs=1

; Put process stdout output in this file
stdout_logfile=/var/log/celery/picha_worker.log

; Put process stderr output in this file
stderr_logfile=/var/log/celery/picha_worker.log

; If true, this program will start automatically when supervisord is started
autostart=true

; May be one of false, unexpected, or true. If false, the process will never
; be autorestarted. If unexpected, the process will be restart when the program
; exits with an exit code that is not one of the exit codes associated with this
; process’ configuration (see exitcodes). If true, the process will be
; unconditionally restarted when it exits, without regard to its exit code.
autorestart=true

; The total number of seconds which the program needs to stay running after
; a startup to consider the start successful.
startsecs=10

; Need to wait for currently executing tasks to finish at shutdown.
; Increase this if you have very long running tasks.
stopwaitsecs = 600

; When resorting to send SIGKILL to the program to terminate it
; send SIGKILL to its whole process group instead,
; taking care of its children as well.
killasgroup=true

; if your broker is supervised, set its priority higher
; so it starts first
priority=998

Selderijplanner:picha_celerybeat.conf

; ================================
;  celery beat supervisor example
; ================================

; the name of your supervisord program
[program:pichacelerybeat]

; Set full path to celery program if using virtualenv
command=/home/mosh/.virtualenvs/picha/bin/celerybeat -A picha --loglevel=INFO

; The directory to your Django project
directory=/home/mosh/sites/picha

; If supervisord is run as the root user, switch users to this UNIX user account
; before doing any processing.
user=mosh

; Supervisor will start as many instances of this program as named by numprocs
numprocs=1

; Put process stdout output in this file
stdout_logfile=/var/log/celery/picha_beat.log

; Put process stderr output in this file
stderr_logfile=/var/log/celery/picha_beat.log

; If true, this program will start automatically when supervisord is started
autostart=true

; May be one of false, unexpected, or true. If false, the process will never
; be autorestarted. If unexpected, the process will be restart when the program
; exits with an exit code that is not one of the exit codes associated with this
; process’ configuration (see exitcodes). If true, the process will be
; unconditionally restarted when it exits, without regard to its exit code.
autorestart=true

; The total number of seconds which the program needs to stay running after
; a startup to consider the start successful.
startsecs=10

; if your broker is supervised, set its priority higher
; so it starts first
priority=999

Zorg ervoor dat u de paden in deze bestanden bijwerkt zodat ze overeenkomen met het bestandssysteem van de externe server.

Deze configuratiebestanden voor supervisors vertellen supervisors hoe ze onze 'programma's' (zoals ze door supervisord worden genoemd) moeten uitvoeren en beheren.

In de bovenstaande voorbeelden hebben we twee begeleide programma's gemaakt met de namen "pichacelery" en "pichacelerybeat".

Kopieer deze bestanden nu naar de externe server in de map "/etc/supervisor/conf.d/".

We moeten ook de logbestanden maken die worden genoemd in de bovenstaande scripts op de externe server:

$ touch /var/log/celery/picha_worker.log
$ touch /var/log/celery/picha_beat.log

Voer ten slotte de volgende opdrachten uit om de Supervisor bewust te maken van de programma's - bijv. pichacelery en pichacelerybeat :

$ sudo supervisorctl reread
$ sudo supervisorctl update

Voer de volgende opdrachten uit om te stoppen, te starten en/of de status van de pichacelery te controleren programma:

$ sudo supervisorctl stop pichacelery
$ sudo supervisorctl start pichacelery
$ sudo supervisorctl status pichacelery

U kunt meer over Supervisor lezen in de officiële documentatie.

Laatste tips

1. Geef Django-modelobjecten niet door aan Celery-taken. Om gevallen te voorkomen waarin het modelobject al is gewijzigd voordat het wordt doorgegeven aan een Celery-taak, geeft u de primaire sleutel van het object door aan Celery. U moet dan natuurlijk de primaire sleutel gebruiken om het object uit de database te halen voordat u eraan gaat werken.
2. De standaard Celery-planner maakt enkele bestanden aan om de planning lokaal op te slaan. Deze bestanden zijn "celerybeat-schedule.db" en "celerybeat.pid". Als je een versiebeheersysteem zoals Git gebruikt (wat je zou moeten doen!), is het een goed idee om deze bestanden te negeren en ze niet toe te voegen aan je repository, aangezien ze bedoeld zijn om processen lokaal uit te voeren.

Volgende stappen

Nou, dat was het voor de basisinleiding tot het integreren van Celery in een Django-project.

Wil je meer?

1. Duik in de officiële gebruikershandleiding van Celery voor meer informatie.
2. Maak een Fabfile om Supervisor en de configuratiebestanden in te stellen. Zorg ervoor dat u de opdrachten toevoegt aan reread en update Toezichthouder.
3. Fork het project uit de repo en open een Pull Request om een ​​nieuwe Celery-taak toe te voegen.

Gratis bonus: Klik hier om toegang te krijgen tot een gratis Django Learning Resources Guide (PDF) met tips en trucs en veelvoorkomende valkuilen die je moet vermijden bij het bouwen van Python + Django-webapplicaties.

Veel plezier met coderen!