src/mailman/app/workflow.py

   1 # Copyright (C) 2015-2016 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 import sys
  21 import json
  22 import logging
  23
  24 from collections import deque
  25 from mailman import public
  26 from mailman.interfaces.workflow import IWorkflowStateManager
  27 from zope.component import getUtility
  28
  29
  30 COMMASPACE = ', '
  31 log = logging.getLogger('mailman.error')
  32
  33
  34 @public
  35 class Workflow:
  36     """Generic workflow."""
  37
  38     SAVE_ATTRIBUTES = ()
  39     INITIAL_STATE = None
  40
  41     def __init__(self):
  42         self.token = None
  43         self._next = deque()
  44         self.push(self.INITIAL_STATE)
  45         self.debug = False
  46         self._count = 0
  47
  48     @property
  49     def name(self):
  50         return self.__class__.__name__
  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:                              # pragma: no cover
  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 None:
 145             # The token doesn't exist in the database.
 146             raise LookupError(self.token)
 147         self._next.clear()
 148         if state.step:
 149             self._next.append(state.step)
 150         for attr, value in json.loads(state.data).items():
 151             setattr(self, attr, value)