How to set up periodic tasks automatically

Question

I have a celery task in one of my django apps.

My django, celery and django-celery-beat services are ran in their own docker containers. Since this is a legacy project, I now assume they interact smoothly already with the current config.

What I want is that when the app gets deployed/started it finds my perioidc tasks automatically and registers them in the django celery beat database tables.

My task:

# project.an_app.tasks.generate_statements
from config import celery_app

@celery_app.task(...)
def generate_statements():
    print("Statment generation runs. Current date and time:", datetime.now())
    # my task logic

This is one solution I've found. But this only gets triggered when I try to access the Periodic tasks tables via admin site (probably would be triggered other ways which interact with these tables, but I don't want to "wait" for anything like that).

def schedule_task():
    interval, _ = IntervalSchedule.objects.get_or_create(
        every=30,
        period=IntervalSchedule.SECONDS,
    )

    PeriodicTask.objects.get_or_create(
        interval=interval,
        name="generate-statements",
        task=project.an_app.tasks.generate_statements,
    )
    
schedule_task()

Looking at our legacy code I've found this, this one looks

from project.an_app.tasks import generate_statements

@celery_app.on_after_finalize.connect
def register_periodic_tasks(sender, **kwargs):
    """
    Callback for Celery to register periodic tasks.
    """
    sender.add_periodic_task(
        30.0,
        task=generate_statements,
        name="generate-statements",
    )

The task of the copied code is indeed already django_celery_beat_periodictask table. So I would say this works. However this version did not create my task when I rebuilt the docker containers.

Also I keep finding solutions of simply adding it in the settings.py file:

CELERY_BEAT_SCHEDULER = "django_celery_beat.schedulers:DatabaseScheduler"

CELERY_BEAT_SCHEDULE = {
    "task-name": {
        "task": "project.an_app.tasks.generate_statements",
        "schedule": timedelta(seconds=30),
    },
}

This one did not work as well.

Can someone clear up which solution is the most suitable for what I want and what am I missing in each cases?

Answer 1

Looking at a Django app I have, there is first code in the project/_init.py

from .celery import app as celery_app

__all__ = ("celery_app",)

Then I have a celery.py:

import os

from celery import Celery


os.environ.setdefault("DJANGO_SETTINGS_MODULE", "core.settings")

app = Celery("core")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

That should ensure that celery is loaded at project start and that the tasks is autodiscovered.

Don’t know if it helps in your case though. But might be worth checking.