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     @property
  53     def name(self):
  54         return self.__class__.__name__
  55
  56     def __iter__(self):
  57         return self
  58
  59     def push(self, step):
  60         self._next.append(step)
  61
  62     def _pop(self):
  63         name = self._next.popleft()
  64         step = getattr(self, '_step_{}'.format(name))
  65         self._count += 1
  66         if self.debug:
  67             print('[{:02d}] -> {}'.format(self._count, name), file=sys.stderr)
  68         return name, step
  69
  70     def __next__(self):
  71         try:
  72             name, step = self._pop()
  73             return step()
  74         except IndexError:
  75             raise StopIteration
  76         except:
  77             log.exception('deque: {}'.format(COMMASPACE.join(self._next)))
  78             raise
  79
  80     def run_thru(self, stop_after):
  81         """Run the state machine through and including the given step.
  82
  83         :param stop_after: Name of method, sans prefix to run the
  84             state machine through.  In other words, the state machine runs
  85             until the named method completes.
  86         """
  87         results = []
  88         while True:
  89             try:
  90                 name, step = self._pop()
  91             except (StopIteration, IndexError):
  92                 # We're done.
  93                 break
  94             results.append(step())
  95             if name == stop_after:
  96                 break
  97         return results
  98
  99     def run_until(self, stop_before):
 100         """Trun the state machine until (not including) the given step.
 101
 102         :param stop_before: Name of method, sans prefix that the
 103             state machine is run until the method is reached.  Unlike
 104             `run_thru()` the named method is not run.
 105         """
 106         results = []
 107         while True:
 108             try:
 109                 name, step = self._pop()
 110             except (StopIteration, IndexError):
 111                 # We're done.
 112                 break
 113             if name == stop_before:
 114                 # Stop executing, but not before we push the last state back
 115                 # onto the deque.  Otherwise, resuming the state machine would
 116                 # skip this step.
 117                 self._next.appendleft(step)
 118                 break
 119             results.append(step())
 120         return results
 121
 122     def save(self):
 123         assert self.token, 'Workflow token must be set'
 124         state_manager = getUtility(IWorkflowStateManager)
 125         data = {attr: getattr(self, attr) for attr in self.SAVE_ATTRIBUTES}
 126         # Note: only the next step is saved, not the whole stack.  This is not
 127         # an issue in practice, since there's never more than a single step in
 128         # the queue anyway.  If we want to support more than a single step in
 129         # the queue *and* want to support state saving/restoring, change this
 130         # method and the restore() method.
 131         if len(self._next) == 0:
 132             step = None
 133         elif len(self._next) == 1:
 134             step = self._next[0]
 135         else:
 136             raise AssertionError(
 137                 "Can't save a workflow state with more than one step "
 138                 "in the queue")
 139         state_manager.save(
 140             self.__class__.__name__,
 141             self.token,
 142             step,
 143             json.dumps(data))
 144
 145     def restore(self):
 146         state_manager = getUtility(IWorkflowStateManager)
 147         state = state_manager.restore(self.__class__.__name__, self.token)
 148         if state is not None:
 149             self._next.clear()
 150             if state.step:
 151                 self._next.append(state.step)
 152             if state.data is not None:
 153                 for attr, value in json.loads(state.data).items():
 154                     setattr(self, attr, value)