-
Notifications
You must be signed in to change notification settings - Fork 90
/
backend.py
113 lines (88 loc) · 2.91 KB
/
backend.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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
from abc import ABCMeta, abstractmethod
from decimal import Decimal
from celery.result import AsyncResult, allow_join_result
PROGRESS_STATE = 'PROGRESS'
class AbstractProgressRecorder(object):
__metaclass__ = ABCMeta
@abstractmethod
def set_progress(self, current, total, description=""):
pass
@abstractmethod
def stop_task(self, current, total, exc):
pass
class ConsoleProgressRecorder(AbstractProgressRecorder):
def set_progress(self, current, total, description=""):
print('processed {} items of {}. {}'.format(current, total, description))
def stop_task(self, current, total, exc):
pass
class ProgressRecorder(AbstractProgressRecorder):
def __init__(self, task):
self.task = task
def set_progress(self, current, total, description=""):
percent = 0
if total > 0:
percent = (Decimal(current) / Decimal(total)) * Decimal(100)
percent = float(round(percent, 2))
self.task.update_state(
state=PROGRESS_STATE,
meta={
'pending': False,
'current': current,
'total': total,
'percent': percent,
'description': description
}
)
def stop_task(self, current, total, exc):
self.task.update_state(
state='FAILURE',
meta={
'pending': False,
'current': current,
'total': total,
'percent': 100.0,
'exc_message': str(exc),
'exc_type': str(type(exc))
}
)
class Progress(object):
def __init__(self, task_id):
self.task_id = task_id
self.result = AsyncResult(task_id)
def get_info(self):
if self.result.ready():
success = self.result.successful()
with allow_join_result():
return {
'complete': True,
'success': success,
'progress': _get_completed_progress(),
'result': self.result.get(self.task_id) if success else str(self.result.info),
}
elif self.result.state == PROGRESS_STATE:
return {
'complete': False,
'success': None,
'progress': self.result.info,
}
elif self.result.state in ['PENDING', 'STARTED']:
return {
'complete': False,
'success': None,
'progress': _get_unknown_progress(self.result.state),
}
return self.result.info
def _get_completed_progress():
return {
'pending': False,
'current': 100,
'total': 100,
'percent': 100,
}
def _get_unknown_progress(state):
return {
'pending': state == 'PENDING',
'current': 0,
'total': 100,
'percent': 0,
}