Как правильно зарегистрировать задачу в Celery 4?

Question

[nurzhannogerbek]

Здравствуйте! Столкнулся с проблемой, что Celery не видит таски из приложения profile. При попытки запустить Celery вылетает ошибка (см. ниже). В списке зарегистрированных тастов нет таска который мне нужен (в моем случаи amount_counting). При этом если поставить в celery.py файл django.setup() таск появляется в списке зарегистрированных и саму задачу можно запустить, но вот в таком случаи уже команды runserver выдает ошибку. Мог бы кто-нибудь посоветовать что делать?

Структура:

TestProject
   TestProject
      - settings.py
      - __init__.py
      - celery.py
  profile
      - tasks.py

celery.py:

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'TestProject.settings')
app = Celery('TestProject')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()

settings.py:

CELERY_BROKER_URL = 'redis://localhost:6379'
CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Almaty'
CELERY_BEAT_SCHEDULE = {
    "amount-counting": {
        "task": "profile.tasks.amount_counting",
        "schedule": 60.0,
    },
}

__init__.py:

from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app
 
__all__ = ['celery_app']

profile/tasks.py:

from __future__ import absolute_import, unicode_literals
from celery import task
 
 
@task
def amount_counting():
    # Some Code

ERROR:

[2018-06-04 04:29:22,641: INFO/Beat] Scheduler: Sending due task amount-counting (profile.tasks.amount_counting)
[2018-06-04 04:29:22,654: ERROR/MainProcess] Received unregistered task of type 'profile.tasks.amount_counting'.
The message has been ignored and discarded.

Did you remember to import the module containing this task?
Or maybe you're using relative imports?

Please see
http://docs.celeryq.org/en/latest/internals/protocol.html
for more information.

The full contents of the message body was:
b'[[], {}, {"callbacks": null, "errbacks": null, "chain": null, "chord": null}]' (77b)
Traceback (most recent call last):
  File "/Users/nurzhan_nogerbek/Virtualenvs/enjoy_jumping/lib/python3.6/site-packages/celery/worker/consumer/consumer.py", line 557, in on_task_received
    strategy = strategies[type_]
KeyError: 'profile.tasks.amount_counting'

Comments

[Марат Мхитарян]

Ооооо Asia/Almaty, вы не из пониэкспресс случайно?

[nurzhannogerbek]

marataziat, Здравствуйте! Нет, c чего вы это взяли? Касательно вопроса, что можете сказать?

[Марат Мхитарян]

NogerbekNurzhan, Немогу, просто пониэкспресс в алматы используют django, мало кто в алматы использует django

Answer 1

[nurzhannogerbek]

Нашел проблему и надо сказать для меня это было неожиданно.

В файле tasks.py все зависимости я указал в начали файла как обычно. Так только зависимости поставил в тело функции amount_counting проблема исчезла. Был бы рад, если кто-нибудь объяснил почему все так.

tasks.py:

from __future__ import absolute_import, unicode_literals
from celery import task
 
 
@task
def amount_counting():
    from  app_name.models import ModelName
    # Some other code

Answer 2

[Артем]

жмак - нормальный пример

Comments

[nurzhannogerbek]

Здравствуйте, Артем!

Статью уже читал, но из статьи не совсем понятно какую версию Celery и Django использует автор. Сам файл celery.py у меня схожий как в статье. Поменял только что файл настроек и декоратор в файле tasks.py добавил. Но к сожалению проблему это не решило. Само приложение profile есть в INSTALLED_APPS, так же пробовал CELERY_INCLUDE добавить, но не помогло. Есть ли у вас идеи в чем может быть дело? Если судить по документации app.autodiscover_tasks() должен подтягивать все таски из приложений.

settings.py:

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'

tasks.py:

from EnjoyJumping.celery import app

@app.task
def amount_counting():
    # some code

К слову для эксперимента через shell пробовал сделать так:

from profile.tasks import amount_counting
amount_counting.delay()
<AsyncResult: e5f2207f-7a91-47ed-a58a-3a9afdbdf4d7>

Сам таск нормальный как видите, проблемы именно в том что Celery не видит таск.

[Артем]

NogerbekNurzhan, profile зарегистрированно в сетингах?

[nurzhannogerbek]

Артем, да. profile есть в INSTALLED_APPS.

INSTALLED_APPS = [
    'django.contrib.sites',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',

    'profile',  # 'profile' application
]