Регистрация задачи на основе класса Celery ⇐ Python
-
Гость
Регистрация задачи на основе класса Celery
Python 3.x, Celery 4.x...
У меня есть задача на основе класса.
myproj/celery.py
из импорта сельдерея Сельдерей # настройки django... приложение = Сельдерей('myproj') app.autodiscover_tasks() app1/tasks.py
импортировать сельдерей класс EmailTask(celery.Task): def run(self, *args, **kwargs): self.do_something() Если да:
$ celery worker -A myproj -l info [задания] . app2.tasks.debug_task . app2.tasks.test Итак, декораторы сельдерея работают над регистрацией задач, но задача на основе класса не регистрируется.
Как мне зарегистрировать задачи на основе классов?
Обновление 1:
Если я добавлю следующие строки в app1/tasks.py
из приложения импорта myproj.celery email_task = app.tasks[EmailTask.name] .
$ celery worker -A myproj -l info Файл «myproj/app1/tasks.py», строка 405, в email_task = app.tasks[EmailTask.name] Файл «/usr/local/lib/python3.5/site-packages/celery/app/registry.py», строка 19, в __missing__ поднять self.NotRegistered(ключ) сельдерей.Exceptions.NotRegistered Обновление 2:
Я могу выполнить свою задачу синхронно (запустить) через оболочку. Однако я не могу запустить задачу асинхронно, т. е. через delay.
app1/tasks.py
@app.task защита email_task(): """ Оболочка для вызова задачи на основе класса """ задача = Задача электронной почты() # Task.delay() # Не сработает!!! задача.запуск() .
Оболочка $./manage.py > из app1.tasks импортируйте EmailTask > задача1 = Задача по электронной почте() > Task1.run() # ладно > задача2 = Задача по электронной почте() > Task2.delay() # нет # И на рабочем... [2017-01-22 08:07:28,120: INFO/PoolWorker-1] Задача app1.tasks.email_task[41e5bc7d-058a-400e-9f73-c853c0f60a2a] выполнена успешно в 0,0701281649817247с: нет [2017-01-22 08:10:31,909: ОШИБКА/MainProcess] Получена незарегистрированная задача типа Нет. Сообщение было проигнорировано и удалено.
Python 3.x, Celery 4.x...
У меня есть задача на основе класса.
myproj/celery.py
из импорта сельдерея Сельдерей # настройки django... приложение = Сельдерей('myproj') app.autodiscover_tasks() app1/tasks.py
импортировать сельдерей класс EmailTask(celery.Task): def run(self, *args, **kwargs): self.do_something() Если да:
$ celery worker -A myproj -l info [задания] . app2.tasks.debug_task . app2.tasks.test Итак, декораторы сельдерея работают над регистрацией задач, но задача на основе класса не регистрируется.
Как мне зарегистрировать задачи на основе классов?
Обновление 1:
Если я добавлю следующие строки в app1/tasks.py
из приложения импорта myproj.celery email_task = app.tasks[EmailTask.name] .
$ celery worker -A myproj -l info Файл «myproj/app1/tasks.py», строка 405, в email_task = app.tasks[EmailTask.name] Файл «/usr/local/lib/python3.5/site-packages/celery/app/registry.py», строка 19, в __missing__ поднять self.NotRegistered(ключ) сельдерей.Exceptions.NotRegistered Обновление 2:
Я могу выполнить свою задачу синхронно (запустить) через оболочку. Однако я не могу запустить задачу асинхронно, т. е. через delay.
app1/tasks.py
@app.task защита email_task(): """ Оболочка для вызова задачи на основе класса """ задача = Задача электронной почты() # Task.delay() # Не сработает!!! задача.запуск() .
Оболочка $./manage.py > из app1.tasks импортируйте EmailTask > задача1 = Задача по электронной почте() > Task1.run() # ладно > задача2 = Задача по электронной почте() > Task2.delay() # нет # И на рабочем... [2017-01-22 08:07:28,120: INFO/PoolWorker-1] Задача app1.tasks.email_task[41e5bc7d-058a-400e-9f73-c853c0f60a2a] выполнена успешно в 0,0701281649817247с: нет [2017-01-22 08:10:31,909: ОШИБКА/MainProcess] Получена незарегистрированная задача типа Нет. Сообщение было проигнорировано и удалено.
