src/mailman/app/workflow.py

   1 # Copyright (C) 2015 by the Free Software Foundation, Inc.
   2 #
   3 # This file is part of GNU Mailman.
   4 #
   5 # GNU Mailman is free software: you can redistribute it and/or modify it under
   6 # the terms of the GNU General Public License as published by the Free
   7 # Software Foundation, either version 3 of the License, or (at your option)
   8 # any later version.
   9 #
  10 # GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
  11 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  13 # more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along with
  16 # GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
  17
  18 """Generic workflow."""
  19
  20 __all__ = [
  21     'Workflow',
  22     ]
  23
  24
  25 import sys
  26 import json
  27 import logging
  28
  29 from collections import deque
  30 from mailman.interfaces.workflow import IWorkflowStateManager
  31 from zope.component import getUtility
  32
  33
  34 COMMASPACE = ', '
  35 log = logging.getLogger('mailman.error')
  36
  37
  38 \f
  39 class Workflow:
  40     """Generic workflow."""
  41
  42     SAVE_ATTRIBUTES = ()
  43     INITIAL_STATE = None
  44
  45     def __init__(self):
  46         self.token = None
  47         self._next = deque()
  48         self.push(self.INITIAL_STATE)
  49         self.debug = False
  50         self._count = 0
  51
  52     def __iter__(self):
  53         return self
  54
  55     def push(self, step):
  56         self._next.append(step)
  57
  58     def _pop(self):
  59         name = self._next.popleft()
  60         step = getattr(self, '_step_{}'.format(name))
  61         self._count += 1
  62         if self.debug:
  63             print('[{:02d}] -> {}'.format(self._count, name), file=sys.stderr)
  64         return name, step
  65
  66     def __next__(self):
  67         try:
  68             name, step = self._pop()
  69             return step()
  70         except IndexError:
  71             raise StopIteration
  72         except:
  73             log.exception('deque: {}'.format(COMMASPACE.join(self._next)))
  74             raise
  75
  76     def run_thru(self, stop_after):
  77         """Run the state machine through and including the given step.
  78
  79         :param stop_after: Name of method, sans prefix to run the
  80             state machine through.  In other words, the state machine runs
  81             until the named method completes.
  82         """
  83         results = []
  84         while True:
  85             try:
  86                 name, step = self._pop()
  87             except (StopIteration, IndexError):
  88                 # We're done.
  89                 break
  90             results.append(step())
  91             if name == stop_after:
  92                 break
  93         return results
  94
  95     def run_until(self, stop_before):
  96         """Trun the state machine until (not including) the given step.
  97
  98         :param stop_before: Name of method, sans prefix that the
  99             state machine is run until the method is reached.  Unlike
 100             `run_thru()` the named method is not run.
 101         """
 102         results = []
 103         while True:
 104             try:
 105                 name, step = self._pop()
 106             except (StopIteration, IndexError):
 107                 # We're done.
 108                 break
 109             if name == stop_before:
 110                 # Stop executing, but not before we push the last state back
 111                 # onto the deque.  Otherwise, resuming the state machine would
 112                 # skip this step.
 113                 self._next.appendleft(step)
 114                 break
 115             results.append(step())
 116         return results
 117
 118     def save(self):
 119         assert self.token, 'Workflow token must be set'
 120         state_manager = getUtility(IWorkflowStateManager)
 121         data = {attr: getattr(self, attr) for attr in self.SAVE_ATTRIBUTES}
 122         # Note: only the next step is saved, not the whole stack.  This is not
 123         # an issue in practice, since there's never more than a single step in
 124         # the queue anyway.  If we want to support more than a single step in
 125         # the queue *and* want to support state saving/restoring, change this
 126         # method and the restore() method.
 127         if len(self._next) == 0:
 128             step = None
 129         elif len(self._next) == 1:
 130             step = self._next[0]
 131         else:
 132             raise AssertionError(
 133                 "Can't save a workflow state with more than one step "
 134                 "in the queue")
 135         state_manager.save(
 136             self.__class__.__name__,
 137             self.token,
 138             step,
 139             json.dumps(data))
 140
 141     def restore(self):
 142         state_manager = getUtility(IWorkflowStateManager)
 143         state = state_manager.restore(self.__class__.__name__, self.token)
 144         if state is not None:
 145             self._next.clear()
 146             if state.step:
 147                 self._next.append(state.step)
 148             if state.data is not None:
 149                 for attr, value in json.loads(state.data).items():
 150                     setattr(self, attr, value)