Асинхронни задачи с Django и целина

Когато бях нов в Django, едно от най-разочароващите неща, които изпитах, беше необходимостта да изпълнявам малко код периодично. Написах хубава функция, която изпълняваше действие, което трябваше да се изпълнява ежедневно в 12 часа сутринта. Лесно, нали? Грешно. Това се оказа огромен проблем за мен, тъй като по това време бях свикнал с уеб хостинг тип „Cpanel“, където имаше приятен удобен GUI за настройка на cron работни места точно за тази цел.

След дълго проучване намерих хубаво решение — Celery, мощна опашка за асинхронни задачи, използвана за изпълнение на задачи във фонов режим. Но това доведе до допълнителни проблеми, тъй като не можах да намеря лесен набор от инструкции за интегриране на Celery в Django проект.

Разбира се, в крайна сметка успях да го разбера — това е, което ще обхване тази статия:Как да интегрирате Celery в Django проект и да създавате периодични задачи.

Безплатен бонус: Щракнете тук, за да получите достъп до безплатно ръководство за учебни ресурси на Django (PDF), което ви показва съвети и трикове, както и често срещани клопки, които трябва да избягвате, когато създавате уеб приложения на Python + Django.

Този проект използва Python 3.4, Django 1.8.2, Celery 3.1.18 и Redis 3.0.2.

Общ преглед

За ваше удобство, тъй като това е толкова голяма публикация, моля, обърнете се към тази таблица за кратка информация за всяка стъпка и за да вземете свързания код.

Какво е целина?

„Celery е асинхронна опашка от задачи/опашка със задачи, базирана на разпределено предаване на съобщения. Той е фокусиран върху работа в реално време, но поддържа и планиране." За тази публикация ще се съсредоточим върху функцията за планиране за периодично изпълнение на задание/задача.

Защо е полезно това?

• Помислете за всички случаи, когато ви се е налагало да изпълнявате определена задача в бъдеще. Може би трябва да имате достъп до API на всеки час. Или може би трябва да изпратите партида имейли в края на деня. Голям или малък, Celery прави планирането на такива периодични задачи лесно.
• Никога не искате крайните потребители да чакат ненужно зареждането на страниците или завършването на действията. Ако дълъг процес е част от работния процес на вашето приложение, можете да използвате Celery, за да изпълните този процес на заден план, когато ресурсите станат достъпни, така че приложението ви да може да продължи да отговаря на клиентски заявки. Това държи задачата извън контекста на приложението.

Настройка

Преди да се потопите в Celery, вземете стартовия проект от репозито на Github. Не забравяйте да активирате virtualenv, да инсталирате изискванията и да стартирате миграциите. След това стартирайте сървъра и отидете до http://localhost:8000/ във вашия браузър. Трябва да видите познатия текст „Поздравления за първата ви страница, задвижвана от Django“. Когато приключите, убийте сървъра.

След това нека инсталираме Celery с помощта на pip:

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

Сега можем да интегрираме Celery в нашия Django проект само в три лесни стъпки.

Стъпка 1:Добавете celery.py

В директорията „picha“ създайте нов файл, наречен 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))

Обърнете внимание на коментарите в кода.

Стъпка 2:Импортирайте новото си приложение Celery

За да сте сигурни, че приложението Celery се зарежда при стартиране на Django, добавете следния код в __init__.py файл, който се намира до вашия settings.py файл:

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

След като направите това, оформлението на вашия проект трябва да изглежда така:

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

Стъпка 3:Инсталирайте Redis като „брокер“ на Celery

Celery използва „брокери“ за предаване на съобщения между проект Django и работниците на Celery. В този урок ще използваме Redis като посредник на съобщения.

Първо инсталирайте Redis от официалната страница за изтегляне или чрез brew (brew install redis ) и след това се обърнете към вашия терминал, в нов прозорец на терминала стартирайте сървъра:

$ redis-server

Можете да тествате дали Redis работи правилно, като напишете това във вашия терминал:

$ redis-cli ping

Redis трябва да отговори с PONG - опитайте!

След като Redis стартира, добавете следния код към вашия файл settings.py:

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

Също така трябва да добавите Redis като зависимост в проекта Django:

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

Това е! Вече трябва да можете да използвате Celery с Django. За повече информация относно настройката на Celery с Django, моля, вижте официалната документация на Celery.

Преди да продължим, нека направим няколко проверки за здравина, за да се уверим, че всичко е наред...

Проверете дали работникът на Celery е готов да получава задачи:

$ 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

Убийте процеса с CTRL-C. Сега проверете дали планировчикът на задачи Celery е готов за действие:

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

Бум!

Отново убийте процеса, когато приключите.

Задачи за целина

Celery използва задачи, които могат да се разглеждат като обикновени функции на Python, които се извикват с Celery.

Например, нека превърнем тази основна функция в задача Celery:

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

Първо добавете декоратор:

from celery.decorators import task

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

След това можете да стартирате тази задача асинхронно с Celery по следния начин:

add.delay(7, 8)

Просто, нали?

Така че тези типове задачи са идеални, когато искате да заредите уеб страница, без да карате потребителя да чака завършване на някакъв фонов процес.

Нека разгледаме пример...

Връщайки се към проекта Django, вземете версия трета, която включва приложение, което приема обратна връзка от потребителите, подходящо наречена 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

Инсталирайте новите изисквания, стартирайте приложението и отидете на http://localhost:8000/feedback/. Трябва да видите:

Нека свържем задачата Celery.

Добавете задачата

По принцип, след като потребителят изпрати формуляра за обратна връзка, искаме незабавно да го оставим да продължи по веселия си път, докато обработваме обратната връзка, изпращаме имейл и т.н., всичко това на заден план.

За да постигнете това, първо добавете файл, наречен tasks.py към директорията “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)

След това актуализирайте forms.py така:

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

По същество send_feedback_email_task.delay(email, message) функцията обработва и изпраща имейла за обратна връзка във фонов режим, докато потребителят продължава да използва сайта.

ЗАБЕЛЕЖКА :success_url в views.py е настроен да пренасочва потребителя към / , който все още не съществува. Ще настроим тази крайна точка в следващия раздел.

Периодични задачи

Често ще трябва да планирате дадена задача да се изпълнява в определено време от време на време – т.е., например, може да се наложи уеб скрепер да се изпълнява ежедневно. Такива задачи, наречени периодични задачи, са лесни за настройка с Celery.

Celery използва „celery beat“, за да планира периодични задачи. Celery beat изпълнява задачи на редовни интервали, които след това се изпълняват от работниците в celery.

Например следната задача е планирана да се изпълнява на всеки петнадесет минути:

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

Нека разгледаме по-стабилен пример, като добавим тази функционалност в проекта Django...

Обратно към проекта Django, вземете версия четвърта, която включва друго ново приложение, наречено photos , който използва API на Flickr, за да получи нови снимки за показване на сайта:

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

Инсталирайте новите изисквания, стартирайте миграциите и след това стартирайте сървъра, за да сте сигурни, че всичко е наред. Опитайте отново да изпробвате формуляра за обратна връзка. Този път би трябвало да пренасочва добре.

Какво следва?

Е, тъй като ще трябва периодично да извикваме API на Flickr, за да добавим още снимки към нашия сайт, можем да добавим задача Celery.

Добавете задачата

Добавете tasks.py към photos приложение:

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

Тук стартираме save_latest_flickr_image() функция на всеки петнадесет минути, като обвиете извикването на функцията в task . @periodic_task декораторът абстрахира кода, за да изпълни задачата Celery, оставяйки tasks.py файл чист и лесен за четене!

Изпълнява се локално

Готови ли сте да стартирате това нещо?

Когато приложението ви Django и Redis работят, отворете два нови терминални прозореца/раздела. Във всеки нов прозорец отидете до директорията на вашия проект, активирайте своя virtualenv и след това изпълнете следните команди (по една във всеки прозорец):

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

Когато посетите сайта на http://127.0.0.1:8000/, сега трябва да видите едно изображение. Нашето приложение получава едно изображение от Flickr на всеки 15 минути:

Разгледайте photos/tasks.py за да видите кода. Щракването върху бутона „Обратна връзка“ ви позволява да... изпратите обратна връзка:

Това работи чрез задача за целина. Разгледайте feedback/tasks.py за повече.

Това е всичко, проектът Picha работи и работи!

Това е добре за тестване, докато разработвате вашия Django проект локално, но не работи толкова добре, когато трябва да разгръщате в производство - като например в DigitalOcean, може би. За това се препоръчва да стартирате Celery worker и планировчик във фонов режим като демон с Supervisor.

Изпълнява се отдалечено

Монтажът е лесен. Вземете версия пет от репото (ако все още го нямате). След това SSH във вашия отдалечен сървър и стартирайте:

$ sudo apt-get install supervisor

След това трябва да кажем на Supervisor за нашите работници от Celery, като добавим конфигурационни файлове към директорията “/etc/supervisor/conf.d/” на отдалечения сървър. В нашия случай се нуждаем от два такива конфигурационни файла – един за Celery worker и един за Celery Scheduler.

Локално създайте папка, наречена „supervisor“ в корена на проекта. След това добавете следните файлове...

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

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

Уверете се, че сте актуализирали пътищата в тези файлове, за да съответстват на файловата система на отдалечения сървър.

По принцип тези конфигурационни файлове на надзорника казват на супервайзора как да изпълнява и управлява нашите „програми“ (както ги нарича supervisord).

В примерите по-горе създадохме две супервайзорни програми, наречени „pichacelery“ и „pichacelerybeat“.

Сега просто копирайте тези файлове на отдалечения сървър в директорията “/etc/supervisor/conf.d/”.

Също така трябва да създадем регистрационните файлове, които са споменати в горните скриптове на отдалечения сървър:

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

И накрая, изпълнете следните команди, за да накарате Supervisor да разбере програмите - напр. pichacelery и pichacelerybeat :

$ sudo supervisorctl reread
$ sudo supervisorctl update

Изпълнете следните команди, за да спрете, стартирате и/или проверите състоянието на pichacelery програма:

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

Можете да прочетете повече за Supervisor от официалната документация.

Последни съвети

1. Не предавайте обекти на Django модел на задачи на Celery. За да избегнете случаите, в които обектът на модела вече се е променил, преди да бъде предаден на задача Celery, предайте първичния ключ на обекта на Celery. Тогава, разбира се, ще трябва да използвате първичния ключ, за да получите обекта от базата данни, преди да работите върху него.
2. Разписанието на Celery по подразбиране създава някои файлове, за да съхранява своя график локално. Тези файлове ще бъдат „celerybeat-schedule.db“ и „celerybeat.pid“. Ако използвате система за контрол на версиите като Git (която трябва!), добра идея е да игнорирате тези файлове и да не ги добавяте към вашето хранилище, тъй като те са за локално стартиране на процеси.

Следващи стъпки

Е, това е всичко за основното въведение в интегрирането на Celery в Django проект.

Искате още?

1. Потопете се в официалното ръководство за потребителя на Celery, за да научите повече.
2. Създайте Fabfile за настройка на Supervisor и конфигурационните файлове. Не забравяйте да добавите командите към reread и update Супервайзер.
3. Разклонете проекта от репо и отворете заявка за изтегляне, за да добавите нова задача Celery.

Безплатен бонус: Щракнете тук, за да получите достъп до безплатно ръководство за учебни ресурси на Django (PDF), което ви показва съвети и трикове, както и често срещани клопки, които трябва да избягвате, когато създавате уеб приложения на Python + Django.

Приятно кодиране!