django_celery_quickstart_with_shared_tasks_and_autodiscover.py

1# proj/proj/celery.py
2import os
3from celery import Celery
4
5# Set the default Django settings module for the 'celery' program.
6os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings')
7
8app = Celery('proj')
9
10# Using a string here means the worker doesn't have to serialize
11# the configuration object to child processes.
12# - namespace='CELERY' means all celery-related configuration keys
13#   should have a `CELERY_` prefix.
14app.config_from_object('django.conf:settings', namespace='CELERY')
15
16# Load task modules from all registered Django apps.
17app.autodiscover_tasks()
18
19
20@app.task(bind=True, ignore_result=True)
21def debug_task(self):
22    print(f'Request: {self.request!r}')
23
24# proj/proj/__init__.py
25# This ensures the app is always imported when Django starts so that shared_task will use this app.
26from .celery import app as celery_app
27
28__all__ = ('celery_app',)
29
30# demoapp/tasks.py
31# Example of a shared task in a Django app
32from celery import shared_task
33
34@shared_task
35def add(x, y):
36    return x + y
37
38@shared_task
39def mul(x, y):
40    return x * y
41
42@shared_task
43def xsum(numbers):
44    return sum(numbers)