forked from celery/django-celery-results
-
Notifications
You must be signed in to change notification settings - Fork 0
/
celery_app.py
74 lines (57 loc) · 1.38 KB
/
celery_app.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import os
from celery import Celery, group, chain
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'reproduce.rep194.settings')
app = Celery('proj')
# Using a string here means the worker doesn't have to serialize
# the configuration object.
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
@app.task(name="task_0")
def task_0():
return "Task 0"
@app.task(name="task_1")
def task_1():
return "Task 1"
@app.task(name="task_2.1")
def task_2_1():
return "Task 2.1"
@app.task(name="task_2.2")
def task_2_2():
return "Task 2.2"
@app.task(name="task_2", bind=True)
def task_2(self):
return self.replace(group(task_2_1.si(), task_2_2.si()))
@app.task(name="task_3")
def task_3():
return "Task 3"
@app.task(name="task_4")
def task_4():
return "Task 4"
@app.task(name="task_5")
def task_5():
return "Task 5"
@app.task(name="task_6")
def task_6():
return "Task 6"
@app.task(name="task_7")
def task_7():
return "Task 7"
def run():
import django
django.setup()
group(
task_0.si(),
chain(
task_1.si(),
group(
task_2.si(),
chain(
task_3.si(),
task_4.si(),
task_5.si(),
task_6.si(),
)
),
task_7.si()
)
).delay()