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
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
|
# Copyright (C) 2015-2017 by the Free Software Foundation, Inc.
#
# This file is part of GNU Mailman.
#
# GNU Mailman is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option)
# any later version.
#
# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# GNU Mailman. If not, see <http://www.gnu.org/licenses/>.
"""Generic workflow."""
import sys
import json
import logging
from collections import deque
from mailman.interfaces.workflows import IWorkflow, IWorkflowStateManager
from mailman.utilities.modules import abstract_component
from public import public
from zope.component import getUtility
from zope.interface import implementer
COMMASPACE = ', '
log = logging.getLogger('mailman.error')
@public
@abstract_component
@implementer(IWorkflow)
class Workflow:
"""Generic workflow."""
initial_state = None
save_attributes = ()
def __init__(self):
self.token = None
self._next = deque()
self.push(self.initial_state)
self.debug = False
self._count = 0
def __iter__(self):
"""See `IWorkflow`."""
return self
def __next__(self):
"""See `IWorkflow`."""
try:
name, step = self._pop()
return step()
except IndexError:
raise StopIteration
except:
log.exception('deque: {}'.format(COMMASPACE.join(self._next)))
raise
def push(self, step):
"""See `IWorkflow`."""
self._next.append(step)
def _pop(self):
name = self._next.pop()
step = getattr(self, '_step_{}'.format(name))
self._count += 1
if self.debug: # pragma: nocover
print('[{:02d}] -> {}'.format(self._count, name), file=sys.stderr)
return name, step
def run_thru(self, stop_after):
"""See `IWorkflow`."""
results = []
while True:
try:
name, step = self._pop()
except (StopIteration, IndexError):
# We're done.
break
results.append(step())
if name == stop_after:
break
return results
def run_until(self, stop_before):
"""See `IWorkflow`."""
results = []
while True:
try:
name, step = self._pop()
except (StopIteration, IndexError):
# We're done.
break
if name == stop_before:
# Stop executing, but not before we push the last state back
# onto the deque. Otherwise, resuming the state machine would
# skip this step.
self._next.appendleft(name)
break
results.append(step())
return results
def save(self):
"""See `IWorkflow`."""
assert self.token, 'Workflow token must be set'
state_manager = getUtility(IWorkflowStateManager)
data = {attr: getattr(self, attr) for attr in self.save_attributes}
# Save the workflow stack.
if len(self._next) == 0:
steps = '[]'
else:
steps = json.dumps(list(self._next))
state_manager.save(self.token, steps, json.dumps(data))
def restore(self):
"""See `IWorkflow`."""
state_manager = getUtility(IWorkflowStateManager)
state = state_manager.restore(self.token)
if state is None:
# The token doesn't exist in the database.
raise LookupError(self.token)
self._next.clear()
if state.steps:
self._next.extend(json.loads(state.steps))
data = json.loads(state.data)
for attr in self.save_attributes:
try:
setattr(self, attr, data[attr])
except KeyError:
pass
|
