django_celery_beat_periodic_task_with_database_scheduler.py

1# 1. Add to your settings.py
2INSTALLED_APPS = [
3    ...,
4    'django_celery_beat',
5]
6
7# 2. Configure Celery to use the django-celery-beat scheduler in settings.py
8CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
9
10# 3. Complete example of defining a periodic task programmatically (e.g., in a signals.py or management command)
11import json
12from django_celery_beat.models import PeriodicTask, IntervalSchedule
13
14# Create an interval of every 30 seconds
15schedule, created = IntervalSchedule.objects.get_or_create(
16    every=30,
17    period=IntervalSchedule.SECONDS,
18)
19
20# Create the periodic task
21PeriodicTask.objects.create(
22    interval=schedule,                  # we created this above.
23    name='Importing contacts',          # simply describes this periodic task.
24    task='proj.tasks.import_contacts',  # name of the celery task to run.
25    args=json.dumps(['arg1', 'arg2']),
26    kwargs=json.dumps({
27       'be_careful': True,
28    }),
29)