Asynchronní úlohy s Django a celerem

Když jsem byl u Djanga nový, jednou z nejvíce frustrujících věcí, které jsem zažil, byla potřeba pravidelně spouštět trochu kódu. Napsal jsem pěknou funkci, která provedla akci, která musela běžet denně ve 12 hodin. Snadné, že? Špatně. To se pro mě ukázalo jako obrovský problém, protože v té době jsem byl zvyklý na webhosting typu „Cpanel“, kde bylo pěkné praktické GUI pro nastavení úloh cron právě pro tento účel.

Po dlouhém zkoumání jsem našel pěkné řešení – Celery, výkonnou asynchronní frontu úloh, která se používá pro spouštění úloh na pozadí. To však vedlo k dalším problémům, protože jsem nemohl najít snadnou sadu pokynů k integraci celery do projektu Django.

Samozřejmě se mi nakonec podařilo na to přijít – čemuž se bude věnovat tento článek:Jak integrovat celer do projektu Django a vytvářet pravidelné úlohy.

Box zdarma: Kliknutím sem získáte přístup k bezplatné příručce Django Learning Resources Guide (PDF), která vám ukáže tipy a triky a také běžné nástrahy, kterým je třeba se vyhnout při vytváření webových aplikací Python + Django.

Tento projekt využívá Python 3.4, Django 1.8.2, Celery 3.1.18 a Redis 3.0.2.

Přehled

Protože se jedná o tak velký příspěvek, pro vaše pohodlí se vraťte k této tabulce, kde najdete stručné informace o každém kroku a uchopte příslušný kód.

Co je celer?

„Celery je asynchronní fronta úloh/fronta úloh založená na distribuovaném předávání zpráv. Je zaměřen na provoz v reálném čase, ale podporuje také plánování.“ V tomto příspěvku se zaměříme na funkci plánování pro pravidelné spouštění úlohy/úkolu.

Proč je to užitečné?

• Přemýšlejte o tom, kolikrát jste museli v budoucnu spouštět určitou úlohu. Možná jste potřebovali přistupovat k API každou hodinu. Nebo jste možná na konci dne potřebovali poslat dávku e-mailů. Velký nebo malý Celery usnadňuje plánování takových pravidelných úkolů.
• Nikdy nechcete, aby koncoví uživatelé museli zbytečně čekat na načtení stránek nebo dokončení akcí. Pokud je součástí pracovního postupu vaší aplikace dlouhý proces, můžete Celery použít ke spuštění tohoto procesu na pozadí, jakmile budou dostupné prostředky, aby vaše aplikace mohla i nadále reagovat na požadavky klientů. To udrží úkol mimo kontext aplikace.

Nastavení

Než se ponoříte do Celery, stáhněte si startovací projekt z úložiště Github. Ujistěte se, že jste aktivovali virtualenv, nainstalovali požadavky a spustili migrace. Poté spusťte server a přejděte ve svém prohlížeči na http://localhost:8000/. Měli byste vidět známý text „Blahopřejeme k vaší první stránce s technologií Django“. Až budete hotovi, zabijte server.

Dále nainstalujme celer pomocí pip:

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

Nyní můžeme celer integrovat do našeho projektu Django ve třech snadných krocích.

Krok 1:Přidejte celery.py

V adresáři „picha“ vytvořte nový soubor s názvem 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))

Všimněte si komentářů v kódu.

Krok 2:Importujte svou novou aplikaci Celery

Chcete-li zajistit, aby se aplikace Celery načetla při spuštění Django, přidejte následující kód do __init__.py soubor, který se nachází vedle vašeho settings.py soubor:

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

Když to uděláte, rozvržení projektu by nyní mělo vypadat takto:

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

Krok 3:Nainstalujte Redis jako celerového „brokera“

Celery používá „zprostředkovatele“ k předávání zpráv mezi projektem Django a pracovníky Celery. V tomto tutoriálu použijeme Redis jako zprostředkovatele zpráv.

Nejprve nainstalujte Redis z oficiální stránky ke stažení nebo přes brew (brew install redis ) a poté se obraťte na svůj terminál, v novém okně terminálu spusťte server:

$ redis-server

Správnou funkci Redis můžete otestovat zadáním tohoto do terminálu:

$ redis-cli ping

Redis by měl odpovědět PONG - zkuste to!

Jakmile bude Redis spuštěn, přidejte do souboru settings.py následující kód:

# 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'

Také musíte přidat Redis jako závislost v projektu Django:

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

A je to! Nyní byste měli být schopni používat celer s Django. Další informace o nastavení Celery s Django najdete v oficiální dokumentaci Celery.

Než budeme pokračovat, proveďte několik kontrol zdravého rozumu, abyste se ujistili, že je vše v pořádku...

Otestujte, zda je pracovník Celery připraven přijímat úkoly:

$ 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

Zabijte proces pomocí CTRL-C. Nyní vyzkoušejte, zda je plánovač úloh Celery připraven k akci:

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

Bum!

Po dokončení znovu proces ukončete.

Celer Tasks

Celery využívá úlohy, které lze považovat za běžné funkce Pythonu, které se volají s Celery.

Udělejme například tuto základní funkci v úkol s celerem:

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

Nejprve přidejte dekoratér:

from celery.decorators import task

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

Potom můžete tuto úlohu spustit asynchronně s Celery takto:

add.delay(7, 8)

Jednoduché, že?

Tyto typy úloh jsou tedy ideální, když chcete načíst webovou stránku, aniž by uživatel musel čekat na dokončení nějakého procesu na pozadí.

Podívejme se na příklad…

Vraťte se k projektu Django, vezměte si verzi tři, která obsahuje aplikaci, která přijímá zpětnou vazbu od uživatelů, příhodně nazvanou feedback :

├── 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

Nainstalujte nové požadavky, spusťte aplikaci a přejděte na http://localhost:8000/feedback/. Měli byste vidět:

Pojďme zapojit úkol Celer.

Přidat úkol

V podstatě poté, co uživatel odešle formulář zpětné vazby, jej chceme okamžitě nechat pokračovat ve své veselé cestě, zatímco my zpracujeme zpětnou vazbu, pošleme e-mail atd., to vše na pozadí.

Chcete-li to provést, nejprve přidejte soubor s názvem tasks.py do adresáře „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)

Poté aktualizujte forms.py takhle:

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'])

V podstatě send_feedback_email_task.delay(email, message) funkce zpracuje a odešle e-mail se zpětnou vazbou na pozadí, když uživatel pokračuje v používání webu.

POZNÁMKA :success_url v views.py je nastaveno na přesměrování uživatele na / , který zatím neexistuje. Tento koncový bod nastavíme v další sekci.

Pravidelné úkoly

Často budete muset naplánovat spuštění úlohy v určitou dobu – například webový škrabák může být třeba spouštět denně. Takové úlohy, nazývané periodické úlohy, lze snadno nastavit pomocí Celery.

Celer používá „celer beat“ k plánování pravidelných úkolů. Celery beat spouští úkoly v pravidelných intervalech, které pak provádějí pracovníci celeru.

Například následující úloha je naplánována na spuštění každých patnáct minut:

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

Podívejme se na robustnější příklad přidáním této funkce do projektu Django…

Zpět k projektu Django, stáhněte si verzi čtyři, která obsahuje další novou aplikaci nazvanou photos , která používá Flickr API k získání nových fotografií pro zobrazení na webu:

├── 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

Nainstalujte nové požadavky, spusťte migraci a poté spusťte server, abyste se ujistili, že je vše v pořádku. Zkuste formulář zpětné vazby otestovat znovu. Tentokrát by mělo přesměrovat v pořádku.

Co bude dál?

Protože bychom museli pravidelně volat rozhraní Flickr API, abychom na naše stránky přidali další fotografie, můžeme přidat úkol Celery.

Přidat úkol

Přidejte tasks.py k photos aplikace:

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")

Zde spustíme save_latest_flickr_image() funkci každých patnáct minut zabalením volání funkce do task . @periodic_task decorator odebere kód pro spuštění úlohy Celery a ponechá tasks.py soubor čistý a snadno čitelný!

Spuštění lokálně

Jste připraveni spustit tuto věc?

Se spuštěnou aplikací Django a Redis otevřete dvě nová okna/karty terminálu. V každém novém okně přejděte do adresáře svého projektu, aktivujte virtuální env a poté spusťte následující příkazy (v každém okně jeden):

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

Když navštívíte stránku na http://127.0.0.1:8000/, měli byste nyní vidět jeden obrázek. Naše aplikace získá jeden obrázek z Flickru každých 15 minut:

Podívejte se na photos/tasks.py abyste viděli kód. Kliknutím na tlačítko „Zpětná vazba“ můžete... odeslat zpětnou vazbu:

Funguje to přes celerovou úlohu. Podívejte se na feedback/tasks.py pro více.

To je vše, máte spuštěný projekt Picha!

To je dobré pro testování při místním vývoji vašeho projektu Django, ale nefunguje to tak dobře, když potřebujete nasadit do produkce – třeba na DigitalOcean. Za tímto účelem se doporučuje spouštět Celery worker a plánovač na pozadí jako démona se správcem.

Vzdálené spouštění

Instalace je jednoduchá. Získejte verzi pět z repozitáře (pokud ji ještě nemáte). Poté SSH na váš vzdálený server a spusťte:

$ sudo apt-get install supervisor

Poté musíme nadřízenému říci o našich pracovnících Celery přidáním konfiguračních souborů do adresáře „/etc/supervisor/conf.d/“ na vzdáleném serveru. V našem případě potřebujeme dva takové konfigurační soubory – jeden pro Celery worker a jeden pro Celery plánovač.

Lokálně vytvořte složku s názvem „supervisor“ v kořenovém adresáři projektu. Poté přidejte následující soubory…

Celery Worker: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

Plánovač Celery: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

Ujistěte se, že jste aktualizovali cesty v těchto souborech, aby odpovídaly souborovému systému vzdáleného serveru.

Tyto konfigurační soubory supervizora v podstatě říkají supervizorovi, jak spouštět a spravovat naše „programy“ (jak je nazývá supervizor).

Ve výše uvedených příkladech jsme vytvořili dva supervizní programy nazvané „pichacelery“ a „pichacelerybeat“.

Nyní stačí zkopírovat tyto soubory na vzdálený server do adresáře „/etc/supervisor/conf.d/“.

Potřebujeme také vytvořit soubory protokolu, které jsou uvedeny ve výše uvedených skriptech na vzdáleném serveru:

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

Nakonec spusťte následující příkazy, aby správce informoval o programech – např. pichacelery a pichacelerybeat :

$ sudo supervisorctl reread
$ sudo supervisorctl update

Spuštěním následujících příkazů zastavíte, spustíte a/nebo zkontrolujete stav pichacelery program:

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

Více o Supervisor si můžete přečíst v oficiální dokumentaci.

Poslední tipy

1. Nepředávejte objekty modelu Django úkolům Celery. Abyste se vyhnuli případům, kdy se objekt modelu změnil již předtím, než byl předán úloze Celery, předejte primární klíč objektu Celery. Pak byste samozřejmě museli použít primární klíč k získání objektu z databáze, než na něm budete pracovat.
2. Výchozí plánovač Celery vytváří nějaké soubory pro místní uložení plánu. Tyto soubory by byly „celerybeat-schedule.db“ a „celerybeat.pid“. Pokud používáte systém správy verzí, jako je Git (což byste měli!), je dobré tyto soubory ignorovat a nepřidávat je do svého úložiště, protože jsou pro spouštění procesů lokálně.

Další kroky

No, to je vše pro základní úvod k integraci Celery do projektu Django.

Chcete více?

1. Ponořte se do oficiální uživatelské příručky Celery, kde se dozvíte více.
2. Vytvořte soubor Fabfile pro nastavení správce a konfiguračních souborů. Nezapomeňte přidat příkazy k reread a update Vedoucí.
3. Rozdělte projekt z repozitáře a otevřete žádost o stažení pro přidání nového úkolu Celer.

Box zdarma: Kliknutím sem získáte přístup k bezplatné příručce Django Learning Resources Guide (PDF), která vám ukáže tipy a triky a také běžné nástrahy, kterým je třeba se vyhnout při vytváření webových aplikací Python + Django.

Hodně štěstí při kódování!