django-celery-beat

# myproject/settings.py INSTALLED_APPS = [ # ... 'django.contrib.admin', 'django_celery_beat', # 'myapp' where your tasks are defined ] # Configure Celery Beat scheduler to use the database CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler' # myproject/celery.py import os from celery import Celery os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') app = Celery('myproject') # Using a string here means the worker doesn't have to serialize # the configuration object to child processes. # - namespace='CELERY' means all celery-related configuration keys # should have a `CELERY_` prefix. app.config_from_object('django.conf:settings', namespace='CELERY') # Auto-discover tasks in all installed apps app.autodiscover_tasks() @app.task(bind=True, ignore_result=True) def debug_task(self): print(f'Request: {self.request!r}') # myapp/tasks.py from celery import shared_task from celery.utils.log import get_task_logger logger = get_task_logger(__name__) @shared_task def my_periodic_task(): logger.info('Executing my_periodic_task!') # To run: # 1. Apply migrations: python manage.py migrate django_celery_beat # 2. Start Celery worker: celery -A myproject worker -l info # 3. Start Celery Beat: celery -A myproject beat -l info --scheduler django_celery_beat.schedulers:DatabaseScheduler # 4. In Django Admin (e.g., http://127.0.0.1:8000/admin/django_celery_beat/periodictask/add/): # - Create a Crontab Schedule (e.g., 'Every minute': minute='*', hour='*', day_of_week='*', day_of_month='*', month_of_year='*') # - Create a Periodic Task: # - Name: 'Run My Periodic Task' # - Task: 'myapp.tasks.my_periodic_task' (full path to your task) # - Schedule: Choose the Crontab Schedule created above # - Enabled: True