Использование Celery с Django для обработки фоновых задач

Russian (Pусский) translation by Marat Amerov (you can also view the original English article)

Веб-приложения обычно начинаются разрабатываться просто, но могут стать довольно сложными, и большинство из них быстро перерастает область ответственности только за HTTP-запросы.

Когда это происходит, нужно провести различие между тем, что должно выполняться мгновенно (обычно в жизненном цикле HTTP-запроса), и что может произойти в конечном итоге, после запроса. Почему это необходимо? Ну, потому что, когда ваше приложение становится перегруженным трафиком, решения, подобные этому, начинают иметь значение.

Операции в веб-приложении могут быть классифицированы как критические или request-time операции и фоновые задачи, которые происходят за пределами цикла запрос-ответ. Они сопоставляются с описанными выше:

Должно произойти мгновенно: request-time операция
Должно произойти после ответа: фоновые задачи

Request-time операции могут выполняться в течение одного цикла запрос/ответ, не беспокоясь о том, что операция может получить тайм-аут или что у пользователя может быть плохое соединение. Общие примеры включают CRUD (создание, чтение, обновление, удаление) операций с базой данных и управление пользователями (процедуры входа/выхода).

Фоновые задачи отличаются, поскольку они обычно довольно трудоемки и склонны к сбою, главным образом из-за внешних зависимостей. Некоторые распространенные сценарии среди сложных веб-приложений включают:

Отправка уведомлений о подтверждении или рассылка сообщений
Ежедневное сканирование и скрапинг некоторой информации из разных источников и сохранение их
Анализ данных
Удаление ненужных ресурсов
экспорт документов/фотографии в различных форматах

Фоновые задачи являются основным направлением данного руководства. Наиболее распространенным шаблоном программирования, используемым для этого сценария, является Producer-Consumer архитектура.

Проще говоря, эта архитектура может быть описана следующим образом:

Producers создают данные или задачи.
Задачи помещаются в очередь, которая называется очередью задач.
Consumers несут ответственность за потребление данных или выполнение задач.

Обычно сonsumers загружают задачи из очереди в режиме first-out (FIFO) или в соответствии с их приоритетами. Потребителей также называют воркерами, и это термин, который мы будем использовать повсюду, поскольку он согласуется с терминологией, используемой обсуждаемыми технологиями.

Какие задачи можно обрабатывать в фоновом режиме? Задачи, которые:

Не являются существенными для основного функционала веб-приложения
Не могут выполняться в цикле запросо/ответ, поскольку они медленны (интенсивность ввода-вывода и.т.д.)
Зависят от внешних ресурсов, которые могут быть недоступны или не будут вести себя так, как ожидалось
Возможно, потребуют повторить попытку хотя бы один раз
Должны выполняться по графику

Celery является де-факто выбором для обработки фоновых задач в экосистеме Python/Django. Он имеет простой и понятный API, и он прекрасно сочетается с Django. Он поддерживает различные технологии для очереди задач и различные парадигмы для воркеров.

В этом уроке мы собираемся создать игровое веб-приложение Django (работающее с реальными сценариями), которое использует обработку фоновых задач.

Настройка

Предполагая, что вы уже знакомы с менеджером пакетов Python и виртуальными окружениями, давайте установим Django:

$ pip install Django

Я решил создать еще одно приложение для ведения блога. Приложение будет простым. Пользователь может просто создать учетную запись и без особых проблем может создать пост и опубликовать его на платформе.

Настройка Django проекта quick_publisher:

$ django-admin startproject quick_publisher

Давайте начнем разработку приложения:

$ cd quick_publisher 
$ ./manage.py startapp main

При запуске нового Django проекта мне нравится создавать main приложение, которое содержит, помимо прочего, кастомную модель пользователя. Чаще всего Я сталкиваюсь с ограничениями модели User по умолчанию. Наличие пользовательской модели User дает нам гибкость.

# main/models.py

from django.db import models
from django.contrib.auth.models import AbstractBaseUser, PermissionsMixin, BaseUserManager


class UserAccountManager(BaseUserManager):
    use_in_migrations = True

    def _create_user(self, email, password, **extra_fields):
        if not email:
            raise ValueError('Email address must be provided')

        if not password:
            raise ValueError('Password must be provided')

        email = self.normalize_email(email)
        user = self.model(email=email, **extra_fields)
        user.set_password(password)
        user.save(using=self._db)
        return user

    def create_user(self, email=None, password=None, **extra_fields):
        return self._create_user(email, password, **extra_fields)

    def create_superuser(self, email, password, **extra_fields):
        extra_fields['is_staff'] = True
        extra_fields['is_superuser'] = True

        return self._create_user(email, password, **extra_fields)


class User(AbstractBaseUser, PermissionsMixin):

    REQUIRED_FIELDS = []
    USERNAME_FIELD = 'email'

    objects = UserAccountManager()

    email = models.EmailField('email', unique=True, blank=False, null=False)
    full_name = models.CharField('full name', blank=True, null=True, max_length=400)
    is_staff = models.BooleanField('staff status', default=False)
    is_active = models.BooleanField('active', default=True)

    def get_short_name(self):
        return self.email

    def get_full_name(self):
        return self.email

    def __unicode__(self):
        return self.email

Обязательно ознакомьтесь с документацией Django, если вы не знакомы с тем, как работают пользовательские модели.

Теперь нам нужно указать Django, использовать эту модель пользователя вместо стандартной. Добавьте эту строку в файл quick_publisher/settings.py:

AUTH_USER_MODEL = 'main.User'

Нам также необходимо добавить main приложение в список INSTALLED_APPS в файле quick_publisher/settings.py. Теперь мы можем создавать миграции, применять их и создавать суперпользователя, чтобы иметь возможность входа в панель администратора Django:

$ ./manage.py makemigrations main 
$ ./manage.py migrate 
$ ./manage.py createsuperuser

Давайте теперь создадим отдельное Django приложение, которое отвечает за посты:

$ ./manage.py startapp publish

Давайте определим простую модель Post в publisher/models.py:

from django.db import models
from django.utils import timezone
from django.contrib.auth import get_user_model


class Post(models.Model):
    author = models.ForeignKey(get_user_model())
    created = models.DateTimeField('Created Date', default=timezone.now)
    title = models.CharField('Title', max_length=200)
    content = models.TextField('Content')
    slug = models.SlugField('Slug')

    def __str__(self):
        return '"%s" by %s' % (self.title, self.author)

Привязка модели Post с администратором Django выполняется в файле publisher/admin.py следующим образом:

from django.contrib import admin
from .models import Post


@admin.register(Post)
class PostAdmin(admin.ModelAdmin):
    pass

Наконец, давайте подключим приложение publisher к нашему проекту, добавив его в список INSTALLED_APPS.

Теперь мы можем запустить сервер и перейти к http:// localhost:8000/admin/ и создать наши первые посты, чтобы у нас было то, с чем можно взаимодействовать:

$ ./manage.py runserver

Надеюсь, что Вы сделали задание и создали посты.

Давайте двигаться дальше. Следующий очевидный шаг - создать способ просмотра опубликованных постов.

# publisher/views.py

from django.http import Http404
from django.shortcuts import render
from .models import Post


def view_post(request, slug):
    try:
        post = Post.objects.get(slug=slug)
    except Post.DoesNotExist:
        raise Http404("Poll does not exist")

    return render(request, 'post.html', context={'post': post})

Давайте свяжем наше новое представление с URL-адресом в: quick_publisher/urls.py

# quick_publisher/urls.py

from django.conf.urls import url
from django.contrib import admin

from publisher.views import view_post

urlpatterns = [
    url(r'^admin/', admin.site.urls),
    url(r'^(?P<slug>[a-zA-Z0-9\-]+)', view_post, name='view_post')
]

Наконец, давайте создадим шаблон, который отображает пост: publisher/templates/post.html

<!DOCTYPE html>
<html>
<head lang="en">
    <meta charset="UTF-8">
    <title></title>
</head>
<body>
    <h1>{{ post.title }}</h1>
    <p>{{ post.content }}</p>
    <p>Published by {{ post.author.full_name }} on {{ post.created }}</p>
</body>
</html>

Теперь мы можем перейти к http://localhost:8000/the-slug-of-the-post-you-created/ в браузере. Это не совсем чудо веб-дизайна, но создание красивых страниц постов выходит за рамки этого урока.

Отправка писем с подтверждением

Вот классический сценарий:

Вы создаете учетную запись.
Вы указываете адрес электронной почты, который будет идентифицирован, как уникальный на платформе.
Платформа проверяет, что вы действительно являетесь владельцем адреса электронной почты, отправив электронное письмо с ссылкой на подтверждение.
Пока Вы не выполните проверку, вы не сможете (полностью) использовать платформу.

Давайте добавим флаг is_verified и verify_uid в модель User:

# main/models.py
import uuid


class User(AbstractBaseUser, PermissionsMixin):

    REQUIRED_FIELDS = []
    USERNAME_FIELD = 'email'

    objects = UserAccountManager()

    email = models.EmailField('email', unique=True, blank=False, null=False)
    full_name = models.CharField('full name', blank=True, null=True, max_length=400)
    is_staff = models.BooleanField('staff status', default=False)
    is_active = models.BooleanField('active', default=True)
    is_verified = models.BooleanField('verified', default=False) # Add the `is_verified` flag
    verification_uuid = models.UUIDField('Unique Verification UUID', default=uuid.uuid4)

    def get_short_name(self):
        return self.email

    def get_full_name(self):
        return self.email

    def __unicode__(self):
        return self.email

Давайте воспользуемся этим случаем, чтобы добавить возможность администрирования модели User:

from django.contrib import admin
from .models import User


@admin.register(User)
class UserAdmin(admin.ModelAdmin):
    pass

Давайте сделаем изменения в базе данных:

$ ./manage.py makemigrations 
$ ./manage.py migrate

Теперь нам нужно написать код, который отправляет электронное письмо при создании пользователя. В таких случаях используются сигналы Django, и это хороший момент, чтобы коснуться этой темы.

Сигналы срабатывают до/после определенных событий в приложении. Мы можем определить коллбеки, которые вызываются автоматически при срабатывании сигналов. Чтобы сделать триггер для коллбека, мы должны сначала подключить его к сигналу.

Мы собираемся создать коллбек, который будет вызван после создания пользователя. Мы добавим этот код после определения модели User в: main/models.py

from django.db.models import signals
from django.core.mail import send_mail


def user_post_save(sender, instance, signal, *args, **kwargs):
    if not instance.is_verified:
        # Send verification email
        send_mail(
            'Verify your QuickPublisher account',
            'Follow this link to verify your account: '
                'http://localhost:8000%s' % reverse('verify', kwargs={'uuid': str(instance.verification_uuid)}),
            'from@quickpublisher.dev',
            [instance.email],
            fail_silently=False,
        )

signals.post_save.connect(user_post_save, sender=User)

Здесь мы определили функцию user_post_save и связали ее с сигналом post_save (который запускается после сохранения модели), отправленным моделью User.

Django не просто отправляет электронные письма самостоятельно; Его необходимо связать с почтовой службой. Для простоты вы можете добавить свои учетные данные Gmail в quick_publisher/settings.py или добавить своего любимого поставщика электронной почты.

Вот как выглядит конфигурация Gmail:

EMAIL_USE_TLS = True
EMAIL_HOST = 'smtp.gmail.com'
EMAIL_HOST_USER = '<YOUR_GMAIL_USERNAME>@gmail.com'
EMAIL_HOST_PASSWORD = '<YOUR_GMAIL_PASSWORD>'
EMAIL_PORT = 587

Чтобы проверить все, зайдите в панель администратора и создайте нового пользователя с действительным адресом электронной почты, который вы можете быстро проверить. Если все пойдет хорошо, вы получите электронное письмо со ссылкой для подтверждения. Процедура проверки еще не готова.

Вот как выполнить проверку учетной записи:

# main/views.py

from django.http import Http404
from django.shortcuts import render, redirect
from .models import User


def home(request):
    return render(request, 'home.html')


def verify(request, uuid):
    try:
        user = User.objects.get(verification_uuid=uuid, is_verified=False)
    except User.DoesNotExist:
        raise Http404("User does not exist or is already verified")

    user.is_verified = True
    user.save()

    return redirect('home')

Привязка представления в: quick_publisher/urls.py

# quick_publisher/urls.py

from django.conf.urls import url
from django.contrib import admin

from publisher.views import view_post
from main.views import home, verify

urlpatterns = [
    url(r'^$', home, name='home'),
    url(r'^admin/', admin.site.urls),
    url(r'^verify/(?P<uuid>[a-z0-9\-]+)/', verify, name='verify'),
    url(r'^(?P<slug>[a-zA-Z0-9\-]+)', view_post, name='view_post')
]

Кроме того, не забудьте создать файл home.html в каталоге main/templates/home.html. Это будет рендерится c представлением home.

Попробуйте запустить весь сценарий снова. Если все будет хорошо, вы получите электронное письмо с действительным URL-адресом подтверждения. Если пройдёте по URL-адресу, а затем проверите раздел администратора, Вы увидите, как была проверена учетная запись.

Асинхронная отправка писем

У нас есть проблема в текущей реализации. Возможно, вы заметили, что создание пользователя немного медленное. Это происходит потому, что Django отправляет письмо с проверкой во время запроса.

Вот как это работает: мы отправляем пользовательские данные в приложение Django. Приложение создает модель User, а затем создает соединение с Gmail (или другим выбранным вами сервисом). Django ждет ответа, и только после этого он возвращает ответ на наш браузер.

Вот где проявляется необходимость в Celery. Во первых, убедитесь, что он установлен:

$ pip install Celery

Теперь нам нужно создать Celery приложение в нашем Django приложении:

# quick_publisher/celery.py

import os
from celery import Celery

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

app = Celery('quick_publisher')
app.config_from_object('django.conf:settings')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

Celery - это очередь задач. Он получает задания из нашего приложения Django и запускает их в фоновом режиме. Celery должен быть сопряжен с другими сервисами, которые действуют в качестве брокеров.

Брокеры промежуточно отправляют сообщения между веб-приложением и Celery. В этом уроке мы будем использовать Redis. Redis прост в установке, и мы можем легко начать с него без особых проблем.

Вы можете установить Redis, следуя инструкциям на странице Redis Quick Start. Вам нужно будет установить библиотеку Redis Python, выполнив pip install redis и комплект, необходимый для использования Redis и Celery: pip install celery [redis].

Запустите Redis сервер в отдельной консоли следующим образом: $ redis-server

Давайте добавим связанные с Celery/Redis конфиги в quick_publisher/settings.py:

# REDIS related settings 
REDIS_HOST = 'localhost' 
REDIS_PORT = '6379' 
BROKER_URL = 'redis://' + REDIS_HOST + ':' + REDIS_PORT + '/0' 
BROKER_TRANSPORT_OPTIONS = {'visibility_timeout': 3600} 
CELERY_RESULT_BACKEND = 'redis://' + REDIS_HOST + ':' + REDIS_PORT + '/0'

Прежде чем что-либо может быть запущено в Celery, оно должно быть декларировано как задача.

Вот как это сделать:

# main/tasks.py

import logging

from django.urls import reverse
from django.core.mail import send_mail
from django.contrib.auth import get_user_model
from quick_publisher.celery import app


@app.task
def send_verification_email(user_id):
    UserModel = get_user_model()
    try:
        user = UserModel.objects.get(pk=user_id)
        send_mail(
            'Verify your QuickPublisher account',
            'Follow this link to verify your account: '
                'http://localhost:8000%s' % reverse('verify', kwargs={'uuid': str(user.verification_uuid)}),
            'from@quickpublisher.dev',
            [user.email],
            fail_silently=False,
        )
    except UserModel.DoesNotExist:
        logging.warning("Tried to send verification email to non-existing user '%s'" % user_id)

Мы здесь сделали следующее: мы переместили функцию отправки по почты в другой файл под названием tasks.py.

Несколько примечаний:

Имя файла имеет значение. Celery проходит через все приложения в INSTALLED_APPS и регистрирует задачи в файлах tasks.py.
Обратите внимание, как мы украсили декорировали send_verification_email с помощью @app.task. Это указывает Celery, что это задача, которая будет выполняться в очереди задач.
Обратите внимание, что мы ожидаем аргумент user_id, а не объект User. Это связано с тем, что при отправке задач на Celery может возникнуть проблема с сериализацией сложных объектов. Лучше использовать примитивные типы.

Возвращаясь к main/models.py, код сигнала преобразуется в:

from django.db.models import signals
from main.tasks import send_verification_email


def user_post_save(sender, instance, signal, *args, **kwargs):
    if not instance.is_verified:
        # Send verification email
        send_verification_email.delay(instance.pk)

signals.post_save.connect(user_post_save, sender=User)

Обратите внимание, как мы вызываем метод .delay объекта задачи. Это означает, что мы отправляем задание на Celery, и мы не ожидаем результата. Если бы мы использовали send_verification_email (instance.pk), мы все равно отправили бы его в Celery и ждали завершения задачи, чего мы не хотим.

Прежде чем Вы начнете создавать нового пользователя, есть есть загвоздка. Celery - это сервис, и нам нужно его запустить. Откройте новую консоль, убедитесь, что вы активируете соответствующий virtualenv, и перейдите в каталог проекта.

$ celery worker -A quick_publisher --loglevel=debug --concurrency=4

Эта команда запустит четыре процесса воркеров Celery. Да, теперь Вы можете, наконец, пойти и создать другого пользователя. Обратите внимание на то, что нет задержки, следите за логами в консоли Celery и посмотрите, правильно ли выполняются задачи. Это должно выглядеть примерно так:

[2017-04-28 15:00:09,190: DEBUG/MainProcess] Task accepted: main.tasks.send_verification_email[f1f41e1f-ca39-43d2-a37d-9de085dc99de] pid:62065
[2017-04-28 15:00:11,740: INFO/PoolWorker-2] Task main.tasks.send_verification_email[f1f41e1f-ca39-43d2-a37d-9de085dc99de] succeeded in 2.5500912349671125s: None

Периодические задачи с Celery

Вот еще один общий сценарий. Большинство зрелых веб-приложений отправляют своим пользователям электронные письма c периодичностью. Некоторые распространенные примеры электронных писем с периодической отправкой:

Ежемесячные отчеты
Уведомления о деятельности (лайки, запросы в друзья и т. д.)
Напоминания для выполнения определенных действий («Не забудьте активировать свою учетную запись»)

Вот что мы будем делать в нашем приложении. Мы посчитаем, сколько раз просматривался каждый пост и будем отправлять ежедневный отчет автору. Каждый день мы собираемся пройтись через всех пользователей, загружать их посты и отправлять электронное письмо с таблицей, содержащей посты и количество просмотров.

Давайте изменим модель Post, чтобы мы могли выполнять подсчет просмотров.

class Post(models.Model):
    author = models.ForeignKey(User)
    created = models.DateTimeField('Created Date', default=timezone.now)
    title = models.CharField('Title', max_length=200)
    content = models.TextField('Content')
    slug = models.SlugField('Slug')
    view_count = models.IntegerField("View Count", default=0)

    def __str__(self):
        return '"%s" by %s' % (self.title, self.author)

Как всегда, когда мы меняем модель, нам нужно выполнить миграцию базы данных:

$ ./manage.py makemigrations 

$ ./manage.py migrate

Давайте также изменим Django представление view_post для подсчета просмотров:

def view_post(request, slug):
    try:
        post = Post.objects.get(slug=slug)
    except Post.DoesNotExist:
        raise Http404("Poll does not exist")
    
    post.view_count += 1
    post.save()

    return render(request, 'post.html', context={'post': post})

Было бы полезно отобразить view_count в шаблоне. Добавьте <p>Viewed {{ post.view_count }} times</p> где-нибудь в файле publisher/templates/post.html. Сделайте несколько просмотров поста и посмотрите, как увеличивается счетчик.

Давайте создадим задачу Celery. Поскольку речь идет о постах, я собираюсь разместить её в publisher/tasks.py:

from django.template import Template, Context
from django.core.mail import send_mail
from django.contrib.auth import get_user_model
from quick_publisher.celery import app
from publisher.models import Post


REPORT_TEMPLATE = """
Here's how you did till now:

{% for post in posts %}
        "{{ post.title }}": viewed {{ post.view_count }} times |

{% endfor %}
"""


@app.task
def send_view_count_report():
    for user in get_user_model().objects.all():
        posts = Post.objects.filter(author=user)
        if not posts:
            continue

        template = Template(REPORT_TEMPLATE)

        send_mail(
            'Your QuickPublisher Activity',
            template.render(context=Context({'posts': posts})),
            'from@quickpublisher.dev',
            [user.email],
            fail_silently=False,
        )

Каждый раз, когда Вы вносите изменения в задачи Celery, не забудьте перезапустить процесс Celery. Celery должен обнаружить и перезагрузить задачи. Перед созданием задачи выполняющейся периодический мы должны проверить это в Django shell, чтобы убедиться, что все работает как ожидается:

$ ./manage.py shell 

In [1]: from publisher.tasks import send_view_count_report 

In [2]: send_view_count_report.delay()

Надеюсь, что Вы получили правильный отчёт в своем письме.

Давайте теперь создадим периодическую задачу. Откройте quick_publisher/celery.py и зарегистрируйте задания:

# quick_publisher/celery.py

import os
from celery import Celery
from celery.schedules import crontab

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

app = Celery('quick_publisher')
app.config_from_object('django.conf:settings')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

app.conf.beat_schedule = {
    'send-report-every-single-minute': {
        'task': 'publisher.tasks.send_view_count_report',
        'schedule': crontab(),  # change to `crontab(minute=0, hour=0)` if you want it to run daily at midnight
    },
}

До сих пор мы создавали расписание, которое запускало задачу publisher.tasks.send_view_count_report каждую минуту, как показано в нотации crontab(). Вы также можете указать различные расписания Celery Crontab.

Откройте другую консоль, активируйте соответствующее окружение и запустите службу Celery Beat.

$ celery -A quick_publisher beat

Работа службы Beat заключается в том, чтобы задавать задачи в Celery в соответствии с расписанием. Учтите, что расписание выполняет задачу send_view_count_report каждую минуту в соответствии с настройкой. Это удобно для тестирования, но не рекомендуется для реального веб-приложения.

Сделать задачи более надежными

Задачи часто используются для выполнения ненадежных операций, операций, зависящих от внешних ресурсов, или задачи, которые могут легко падать по различным причинам. Вот инструкция, чтобы сделать их более надежными:

Сделать задачи идемпотентными. Идемпотентная задача - это задача, которая, не изменяет состояние системы, если она остановлена на полпути. Задача либо полностью вносит изменения в систему, либо нет.
Повторение задачи. Если задача падает, рекомендуется попробовать выполнять ее снова и снова, пока она не будет выполнена успешно. Вы можете сделать это c помощью Celery Retry. Еще одна интересная вещь, на которую стоит обратить внимание, - это алгоритм экспоненциального отказа. Это может пригодиться, когда вы думаете об уменьшении ненужной нагрузки на сервер, возникающей из за повторных задач.

Заключение

Надеюсь, что это была интересная статья для вас и хорошее введение в использование Celery с Django.

Вот несколько выводов, которые мы можем сделать:

Хорошей практикой является сохранение ненадежных и трудоемких задач за пределами запроса.
Задачи, которые долго выполняются, должны выполняться в фоновом режиме воркерами (или другими способами).
Фоновые задачи могут использоваться для различных задач, которые не критичны для базового функционирования приложения.
Celery также может выполнять периодические задания с использованием службы celery beat.
Задачи могут быть более надежными, если они сделаны идемпотентными и повторены (возможно, с использованием алгоритма экспоненциального отказа).