django-celery-beat-next 2.10.0

Example creating interval-based periodic task

To create a periodic task executing at an interval you must first create the interval object:

>>> from django_celery_beat.models import PeriodicTask, IntervalSchedule

# executes every 10 seconds.
>>> schedule, created = IntervalSchedule.objects.get_or_create(
...     every=10,
...     period=IntervalSchedule.SECONDS,
... )

That’s all the fields you need: a period type and the frequency.

You can choose between a specific set of periods:

• IntervalSchedule.DAYS

• IntervalSchedule.HOURS

• IntervalSchedule.MINUTES

• IntervalSchedule.SECONDS

• IntervalSchedule.MICROSECONDS

There’s also a “choices tuple” available should you need to present this to the user:

>>> IntervalSchedule.PERIOD_CHOICES

Now that we have defined the schedule object, we can create the periodic task entry:

>>> PeriodicTask.objects.create(
...     interval=schedule,                  # we created this above.
...     name='Importing contacts',          # simply describes this periodic task.
...     task='proj.tasks.import_contacts',  # name of task.
... )

Note that this is a very basic example, you can also specify the arguments and keyword arguments used to execute the task, the queue to send it to[*], and set an expiry time.

Here’s an example specifying the arguments, note how JSON serialization is required:

>>> import json
>>> from datetime import datetime, timedelta

>>> PeriodicTask.objects.create(
...     interval=schedule,                  # we created this above.
...     name='Importing contacts',          # simply describes this periodic task.
...     task='proj.tasks.import_contacts',  # name of task.
...     args=json.dumps(['arg1', 'arg2']),
...     kwargs=json.dumps({
...        'be_careful': True,
...     }),
...     expires=datetime.utcnow() + timedelta(seconds=30)
... )

Example creating crontab-based periodic task

A crontab schedule has the fields: minute, hour, day_of_week, day_of_month and month_of_year, so if you want the equivalent of a 30 * * * * (execute 30 minutes past every hour) crontab entry you specify:

>>> from django_celery_beat.models import CrontabSchedule, PeriodicTask
>>> schedule, _ = CrontabSchedule.objects.get_or_create(
...     minute='30',
...     hour='*',
...     day_of_week='*',
...     day_of_month='*',
...     month_of_year='*',
...     timezone=zoneinfo.ZoneInfo('Canada/Pacific')
... )

The crontab schedule is linked to a specific timezone using the ‘timezone’ input parameter.

Then to create a periodic task using this schedule, use the same approach as the interval-based periodic task earlier in this document, but instead of interval=schedule, specify crontab=schedule:

>>> PeriodicTask.objects.create(
...     crontab=schedule,
...     name='Importing contacts',
...     task='proj.tasks.import_contacts',
... )

Temporarily disable a periodic task

You can use the enabled flag to temporarily disable a periodic task:

>>> periodic_task.enabled = False
>>> periodic_task.save()

Example running periodic tasks

The periodic tasks still need ‘workers’ to execute them. So make sure the default Celery package is installed. (If not installed, please follow the installation instructions here: https://github.com/celery/celery)

Both the worker and beat services need to be running at the same time.

1. Start a Celery worker service (specify your Django project name):

   $ celery -A [project-name] worker --loglevel=info

2. As a separate process, start the beat service (specify the Django scheduler):

   $ celery -A [project-name] beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler

   OR you can use the -S (scheduler flag), for more options see celery beat --help):

   $ celery -A [project-name] beat -l info -S django

   Also, as an alternative, you can run the two steps above (worker and beat services) with only one command (recommended for development environment only):

   $ celery -A [project-name] worker --beat --scheduler django --loglevel=info

3. Now you can add and manage your periodic tasks from the Django Admin interface.

Downloading and installing from source

Download the latest version of django-celery-beat-next from https://pypi.org/project/django-celery-beat-next/

You can install it by doing the following :

$ python3 -m venv .venv
$ source .venv/bin/activate
$ pip install --upgrade build pip
$ tar xvfz django-celery-beat-next-0.0.0.tar.gz
$ cd django-celery-beat-next-0.0.0
$ python -m build
$ pip install --upgrade .

After installation, add django_celery_beat to Django’s settings module:

INSTALLED_APPS = [
    ...,
    'django_celery_beat',
]

Run the django_celery_beat migrations using:

$ python manage.py migrate django_celery_beat

Using the development version

With pip

You can install the latest main version of django-celery-beat-next using the following pip command:

$ pip install git+https://github.com/azataiot/django-celery-beat-next.git#egg=django-celery-beat-next

Developing django-celery-beat

To spin up a local development copy of django-celery-beat with Django admin at http://127.0.0.1:58000/admin/ run:

$ docker-compose up --build

Log-in as user admin with password admin.

TZ Awareness:

If you have a project that is time zone naive, you can set DJANGO_CELERY_BEAT_TZ_AWARE=False in your settings file.