Приложение¶

Основное название¶

Только одно из них важно, и это имя главного модуля. Давайте рассмотрим, почему это так.

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

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

>>> @app.task
... def add(x, y):
...     return x + y

>>> add
<@task: __main__.add>

>>> add.name
__main__.add

>>> app.tasks['__main__.add']
<@task: __main__.add>

и здесь вы снова видите __main__; всякий раз, когда Celery не может определить, к какому модулю принадлежит функция, он использует имя главного модуля для генерации начала имени задачи.

Это является проблемой только в ограниченном наборе случаев использования:

1. Если модуль, в котором определена задача, выполняется как программа.

2. Если приложение создается в оболочке Python (REPL).

Например, здесь, где модуль задач также используется для запуска рабочего с app.worker_main():

tasks.py:

from celery import Celery
app = Celery()

@app.task
def add(x, y): return x + y

if __name__ == '__main__':
    app.worker_main()

При выполнении этого модуля задачи будут называться, начиная с «__main__», но когда модуль импортируется другим процессом, скажем, для вызова задачи, задачи будут называться, начиная с «tasks». (настоящее имя модуля):

>>> from tasks import add
>>> add.name
tasks.add

Вы можете задать другое имя для главного модуля:

>>> app = Celery('tasks')
>>> app.main
'tasks'

>>> @app.task
... def add(x, y):
...     return x + y

>>> add.name
tasks.add

Конфигурация¶

Существует несколько опций, которые могут изменить работу Celery. Эти параметры можно установить непосредственно на экземпляре приложения или использовать специальный модуль конфигурации.

Конфигурация доступна в виде app.conf:

>>> app.conf.timezone
'Europe/London'

где вы также можете напрямую задать значения конфигурации:

>>> app.conf.enable_utc = True

или обновить несколько ключей одновременно, используя метод update:

>>> app.conf.update(
...     enable_utc=True,
...     timezone='Europe/London',
...)

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

1. Изменения, внесенные во время выполнения.

2. Модуль конфигурации (при наличии)

3. Конфигурация по умолчанию (celery.app.defaults).

Вы можете даже добавить новые источники по умолчанию, используя метод app.add_defaults().

from celery import Celery

app = Celery()
app.config_from_object('celeryconfig')

enable_utc = True
timezone = 'Europe/London'

import celeryconfig

from celery import Celery

app = Celery()
app.config_from_object(celeryconfig)

from celery import Celery

app = Celery()

class Config:
    enable_utc = True
    timezone = 'Europe/London'

app.config_from_object(Config)
# or using the fully qualified name of the object:
#   app.config_from_object('module:Config')

import os
from celery import Celery

#: Set default configuration module name
os.environ.setdefault('CELERY_CONFIG_MODULE', 'celeryconfig')

app = Celery()
app.config_from_envvar('CELERY_CONFIG_MODULE')

$ CELERY_CONFIG_MODULE="celeryconfig.prod" celery worker -l INFO

>>> app.conf.humanize(with_defaults=False, censored=True)

>>> app.conf.table(with_defaults=False, censored=True)

Лень¶

Экземпляр приложения является «ленивым», то есть он не будет оцениваться до тех пор, пока он действительно не понадобится.

Создание экземпляра Celery приведет только к следующему:

1. Создайте логический экземпляр часов, используемый для событий.

2. Создайте реестр задач.

3. Установить себя в качестве текущего приложения (но не в том случае, если аргумент set_as_current был отключен)

4. Вызовите обратный вызов app.on_init() (по умолчанию ничего не делает).

Декораторы app.task() не создают задачи в момент определения задачи, вместо этого они откладывают создание задачи либо на момент использования задачи, либо после финализации приложения,

Этот пример показывает, что задача не создается до тех пор, пока вы не используете задачу или не получите доступ к атрибуту (в данном случае repr()):

>>> @app.task
>>> def add(x, y):
...    return x + y

>>> type(add)
<class 'celery.local.PromiseProxy'>

>>> add.__evaluated__()
False

>>> add        # <-- causes repr(add) to happen
<@task: __main__.add>

>>> add.__evaluated__()
True

Финализация приложения происходит либо явно путем вызова app.finalize() – либо неявно путем обращения к атрибуту app.tasks.

Доработка объекта будет:

1. Копирование задач, которые должны быть разделены между приложениями

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

2. Оцените все находящиеся на рассмотрении декораторы задач.

3. Убедитесь, что все задачи привязаны к текущему приложению.

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

Разрыв цепи¶

Хотя можно зависеть от установки текущего приложения, лучшей практикой является постоянная передача экземпляра приложения всему, что в нем нуждается.

Я называю это «цепочкой приложений», поскольку она создает цепочку экземпляров в зависимости от передаваемого приложения.

Следующий пример считается плохой практикой:

from celery import current_app

class Scheduler:

    def run(self):
        app = current_app

Вместо этого он должен принимать в качестве аргумента app:

class Scheduler:

    def __init__(self, app):
        self.app = app

Внутри Celery использует функцию celery.app.app_or_default(), чтобы все работало и в API совместимости на основе модулей

from celery.app import app_or_default

class Scheduler:
    def __init__(self, app=None):
        self.app = app_or_default(app)

В процессе разработки вы можете установить переменную окружения CELERY_TRACE_APP, чтобы вызвать исключение при разрыве цепочки приложений:

$ CELERY_TRACE_APP=1 celery worker -l INFO

Абстрактные задачи¶

Все задачи, созданные с помощью декоратора task(), будут наследоваться от базового класса приложения Task.

Вы можете указать другой базовый класс, используя аргумент base:

@app.task(base=OtherTask):
def add(x, y):
    return x + y

Для создания пользовательского класса задачи необходимо наследоваться от нейтрального базового класса: celery.Task.

from celery import Task

class DebugTask(Task):

    def __call__(self, *args, **kwargs):
        print('TASK STARTING: {0.name}[{0.request.id}]'.format(self))
        return self.run(*args, **kwargs)

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

Для реализации базового класса необходимо создать задачу, используя декоратор app.task():

@app.task(base=DebugTask)
def add(x, y):
    return x + y

Можно даже изменить базовый класс по умолчанию для приложения, изменив его атрибут app.Task():

>>> from celery import Celery, Task

>>> app = Celery()

>>> class MyBaseTask(Task):
...    queue = 'hipri'

>>> app.Task = MyBaseTask
>>> app.Task
<unbound MyBaseTask>

>>> @app.task
... def add(x, y):
...     return x + y

>>> add
<@task: __main__.add>

>>> add.__class__.mro()
[<class add of <Celery __main__:0x1012b4410>>,
 <unbound MyBaseTask>,
 <unbound Task>,
 <type 'object'>]