How to setup periodic task in Celery, Django?

The TLDR answer:

If you using celery for years, then you must have got used to the @periodic_task decorator to help you set up any periodic task. But as soon as you update your celery, you get a deprecation warning, or else if you start using the latest version of celery, you will no more find the periodic task in the package guide. There are a lot of weird (both good and bad) ways to set up periodic tasks. But none of them give you the comfort of using a decorator, you were using in the past.

If you need help to set up Celery with Django, head on to the following blog.

I found the following solution closed to decorator usage.

# task.py, this file can be present in any of the apps.
from celery.schedules import crontab

from backend.celery import app
from backend.celery import task


@app.on_after_finalize.connect
def setup_periodic_tasks(sender, **kwargs):
    # Call every 2 hours 52 min.
    sender.add_periodic_task(
        crontab(minute=52, hour="*/2"),
        test.s("world")
    )


@task
def test(arg):
    print(arg)

# celery.py, this file should be present in the setting.py folder level

from celery import Celery

app = Celery("backend") # "backend" is your main application name, i.e. your folder name where you have this file
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

task = app.task

You can read more about this from the official documentation.

Note: For someone who doesn't know how to set up periodic tasks, please keep on reading, where I have explained what does celery beat mean and how to integrate it into your project.

The Detailed Answer:

I am assuming you already have a basic understanding of celery and how to set up Celery with the Django project.

For running any periodic task we need a celery beat service that will maintain all the schedules and then pass them on to the main celery worker for execution. Do remember that we can have only one instance celery beat worker running, else we will see duplicate periodic tasks running in our project.

One interesting fact which most beginners might miss is, celery beat would only initiate the task at the specified time, i.e. it would put it in the queue when the scheduled time for execution arrives. You need your celery worker running to pick the task from the queue and then execute it.

If you already have celery setup in the project using docker, then all you need to do is, add a new service in your docker-compose.yaml

celery-beat:
    build: ./backend
    command: celery -A backend beat -l info
    volumes:
      - ./backend/:/app/
    env_file:
      - ./.env
    depends_on:
      - postgresql_db
      - redis_server

You can get the full reference of the setup from the example project.

Once you have the setup, all you need to do is have the periodic task setup. Use the below snippet to register the tasks.

# task.py, this file can be present in any of the apps.
from celery.schedules import crontab

from backend.celery import app
from backend.celery import task


@app.on_after_finalize.connect
def setup_periodic_tasks(sender, **kwargs):
    # Call every 2 hours 52 min.
    sender.add_periodic_task(
        crontab(minute=52, hour="*/2"),
        test.s("world")
    )

@task
def test(arg):
    print(arg)

The most important thing in the whole setup is @app.on_after_finalize.connect. What does this mean?

As per the documentation, it mentions that we should use @app.on_after_configure.connect but for Django projects, this doesn't work. You need to use a different signal @app.on_after_finalize.connect.  Do make a note of the subtle difference between configure and finalize.

As per the documentation

on_after_configure
Signal sent after app has prepared the configuration.
on_after_finalize
Signal sent after app has been finalized.

For any Django project modules get loaded only when the app is ready. This signal will be fired after all tasks.py are imported and all possible receivers are already subscribed.