Приложение

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

При отправке сообщения о задаче в 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).

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

tasks.py:

from celery import Celery
app = Celery()

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

if __name__ == '__main__':
    args = ['worker', '--loglevel=INFO']
    app.worker_main(argv=args)

При выполнении этого модуля имена задач будут начинаться с «__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"celeryworker-lINFO

>>> 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=1celeryworker-lINFO

Все задачи, созданные с помощью декоратора app.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'>]