Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

How to pass a class based task into CELERY_BEAT_SCHEDULE

As seen in the docs class based tasks are a fair way to express complex logic.

However, the docs do not specify how to add your shiny newly created class based task into you CELERY_BEAT_SCHEDULE (using django)

Thing I tried: celery.py

app.autodiscover_tasks(lambda: settings.INSTALLED_APPS, 'task_summary')
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    from payments.tasks.generic.payeer import PayeerPaymentChecker
    from payments.tasks.generic.ok_pay import OkPayPaymentChecker

    okpay_import = OkPayPaymentChecker()
    payeer_imprt = PayeerPaymentChecker()

    sender.add_periodic_task(60.0, okpay_import.s(),
                             name='OkPay import',
                             expires=30)

    sender.add_periodic_task(60.0, payeer_imprt.s(),
                             name='Payeer import',
                             expires=30)

-- OR --

payments/task_summary.py

from tasks.generic.import import OkPayPaymentChecker, PayeerPaymentChecker
 run_okpay = OkPayPaymentChecker()
 run_payeer = PayeerPaymentChecker()

CELERY_BEAT_SCHEDULE = {
# yes, i did try referring to the class here
'check_okpay_payments': {

    'task': 'payments.tasks.task_summary.run_okpay',
    'schedule': timedelta(seconds=60),
},
'check_payeer_payments': {
    'task': 'payments.task_summary.run_payeer',
    'schedule': timedelta(seconds=60),
},
}

Really don't know what to do, restoring to something like: payments/task_summary.py/

from payments.tasks.generic.ok_pay import OkPayPaymentChecker
from payments.tasks.generic.payeer import PayeerPaymentChecker
from celery import shared_task


@shared_task
def run_payer():
    instance = PayeerPaymentChecker()
    return instance.run()


@shared_task
def run_okpay():
    instance = OkPayPaymentChecker()
    return instance.run()

Online Resources which I've checked and do not help me / solve the problem:

  • https://denibertovic.com/posts/celery-best-practices/

  • https://blog.balthazar-rouberol.com/celery-best-practices

  • http://shulhi.com/class-based-celery-task/

  • http://jsatt.com/blog/class-based-celery-tasks/

like image 507
Oleg Belousov Avatar asked Jan 30 '17 10:01

Oleg Belousov


People also ask

How do I schedule a task in Django celery?

Here are some key points: DJANGO_SETTINGS_MODULE must be set in the environment before starting a Celery process. Its presence in the environment triggers internal magic in Celery to run the Django setup at the right time. The Celery "application" must be created and configured during startup of both Django and Celery.

How do you run Celerybeat?

Firstly add the django_celery_beat module in installed apps in settings file. And then apply the django migrate command, this will create the tables in admin pannel. After completing all the process like in celery file and create task in tasks.py . you will apply the beat command as mentions in above.

How do you create a periodic task in Django?

We can configure periodic tasks either by manually adding the configurations to the celery.py module or using the django-celery-beat package which allows us to add periodic tasks from the Django Admin by extending the Admin functionality to allow scheduling tasks.


1 Answers

It took me a while to find the answer to this too, and because this question is so high up on google search results I figure I'd drop it here for people who are struggling to find the answer:

You add it just as you would a normal task but using the class name.

CELERY_BEAT_SCHEDULE = {
    'my_task_name': {
        'task': 'mymodule.tasks.MyTaskClass',
        'schedule': timedelta(seconds=60),
},

(This is assuming you have mymodule/tasks.py with:

from celery import Task

class MyTaskClass(Task):

    def run(self, *args, **kwargs):
       ... stuff ...
like image 179
ptr Avatar answered Oct 16 '22 15:10

ptr