Как вызвать метод в celery Django из кода?

Question

[Sabrjkee]

Добрый день. Пытаюсь создать task который должен вызывать со старта вокера и потом вызываться через определенное время. Вот код таска. При запуске вокера получаю это:

[tasks]
  . test_odoo.tasks.populate_db

[2019-04-11 09:59:24,102: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//
[2019-04-11 09:59:24,107: INFO/MainProcess] mingle: searching for neighbors
[2019-04-11 09:59:25,119: INFO/MainProcess] mingle: all alone
[2019-04-11 09:59:25,130: WARNING/MainProcess] /home/user/.virtualenvs/django/lib/python3.6/site-packages/celery/fixups/django.py:202: UserWarning: Using settings.DEBUG leads to a memory leak, never use this setting in production environments!
  warnings.warn('Using settings.DEBUG leads to a memory leak, never '
[2019-04-11 09:59:25,130: INFO/MainProcess] celery@user-System-Product-Name ready.

если нужно использовать apply_async то где его нужно вызывать

Answer 1

[FulTupFul]

celerybeat
Указываем какую задачу и с какой переодичностью запускать:

CELERY_BEAT_SCHEDULE = {
    'task_name': {
        'task': 'app.module.task_name',
        'schedule': timezone.timedelta(minutes=1),
    }
}

Создаем задачу:

from celery import shared_task


@shared_task
def task_name(*args, **kwargs):
    pass

Ну и запускаем задачу:
celery beat -A app_name -l info

https://docs.celeryproject.org/en/latest/userguide...