1 #

     2 # Copyright (c) 2008-2015 Thierry Florac <tflorac AT ulthar.net>

     3 # All Rights Reserved.

     4 #

     5 # This software is subject to the provisions of the Zope Public License,

     6 # Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.

     7 # THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED

     8 # WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

     9 # WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS

    10 # FOR A PARTICULAR PURPOSE.

    11 #

    12 from pyams_utils.traversing import get_parent

    13 

    14 __docformat__ = 'restructuredtext'

    15 

    16 

    17 # import standard library

    18 

    19 # import interfaces

    20 from pyams_workflow.interfaces import MANUAL, IWorkflow, InvalidTransitionError, IWorkflowState, IWorkflowVersions, \

    21     IWorkflowInfo, ConditionFailedError, WorkflowVersionTransitionEvent, WorkflowTransitionEvent, \

    22     NoTransitionAvailableError, AmbiguousTransitionError, SYSTEM, AUTOMATIC, IWorkflowManagedContent, IWorkflowVersion

    23 

    24 # import packages

    25 from pyams_utils.adapter import adapter_config

    26 from pyams_utils.registry import get_utility

    27 from pyams_utils.request import check_request

    28 from pyramid.httpexceptions import HTTPUnauthorized

    29 from pyramid.threadlocal import get_current_registry

    30 from zope.interface import implementer

    31 from zope.lifecycleevent import ObjectModifiedEvent

    32 

    33 

    34 def NullCondition(wf, context):

    35     """Null condition"""

    36     return True

    37 

    38 

    39 def NullAction(wf, context):

    40     """Null action"""

    41     pass

    42 

    43 

    44 def granted_permission(permission, context):

    45     return True

    46 

    47 

    48 class Transition(object):

    49     """Transition object

    50 

    51     A transition doesn't make anything by itself.

    52     Everything is handled by the workflow utility

    53     """

    54 

    55     def __init__(self, transition_id, title, source, destination,

    56                  condition=NullCondition,

    57                  action=NullAction,

    58                  trigger=MANUAL,

    59                  permission=None,

    60                  order=0,

    61                  **user_data):

    62         self.transition_id = transition_id

    63         self.title = title

    64         self.source = source

    65         self.destination = destination

    66         self.condition = condition

    67         self.action = action

    68         self.trigger = trigger

    69         self.permission = permission

    70         self.order = order

    71         self.user_data = user_data

    72 

    73 

    74 @implementer(IWorkflow)

    75 class Workflow(object):

    76     """Workflow utility"""

    77 

    78     def __init__(self, transitions, states, published_states=None):

    79         self.refresh(transitions)

    80         self.states = states

    81         self.published_states = published_states or set()

    82 

    83     def _register(self, transition):

    84         transitions = self._sources.setdefault(transition.source, {})

    85         transitions[transition.transition_id] = transition

    86         self._id_transitions[transition.transition_id] = transition

    87 

    88     def refresh(self, transitions):

    89         self._sources = {}

    90         self._id_transitions = {}

    91         for transition in transitions:

    92             self._register(transition)

    93 

    94     def get_transitions(self, source):

    95         try:

    96             return self._sources[source].values()

    97         except KeyError:

    98             return []

    99 

   100     def get_transition(self, source, transition_id):

   101         transition = self._id_transitions[transition_id]

   102         if transition.source != source:

   103             raise InvalidTransitionError(source)

   104         return transition

   105 

   106     def get_transition_by_id(self, transition_id):

   107         return self._id_transitions[transition_id]

   108 

   109 

   110 WORKFLOW_STATE_KEY = 'pyams_workflow.state'

   111 

   112 

   113 @adapter_config(context=IWorkflowVersion, provides=IWorkflowInfo)

   114 class WorkflowInfo(object):

   115     """Workflow info adapter"""

   116 

   117     def __init__(self, context):

   118         self.context = context

   119         self.wf = get_utility(IWorkflow, name=self.name)

   120 

   121     @property

   122     def parent(self):

   123         return get_parent(self.context, IWorkflowManagedContent)

   124 

   125     @property

   126     def name(self):

   127         return self.parent.workflow_name

   128 

   129     def fire_transition(self, transition_id, comment=None, side_effect=None, check_security=True):

   130         versions = IWorkflowVersions(self.parent)

   131         state = IWorkflowState(self.context)

   132         # this raises InvalidTransitionError if id is invalid for current state

   133         transition = self.wf.get_transition(state.state, transition_id)

   134         # check whether we may execute this workflow transition

   135         if check_security and transition.permission:

   136             request = check_request()

   137             if not request.has_permission(transition.permission):

   138                 raise HTTPUnauthorized()

   139         # now make sure transition can still work in this context

   140         if not transition.condition(self, self.context):

   141             raise ConditionFailedError()

   142         # perform action, return any result as new version

   143         result = transition.action(self, self.context)

   144         if result is not None:

   145             # clear result history

   146             IWorkflowState(result).history.clear()

   147             # stamp it with version

   148             versions.add_version(result, transition.destination)

   149             # execute any side effect:

   150             if side_effect is not None:

   151                 side_effect(result)

   152             event = WorkflowVersionTransitionEvent(result, self.context,

   153                                                    transition.source,

   154                                                    transition.destination,

   155                                                    transition, comment)

   156         else:

   157             versions.set_state(state.version_id, transition.destination)

   158             # execute any side effect

   159             if side_effect is not None:

   160                 side_effect(self.context)

   161             event = WorkflowTransitionEvent(self.context,

   162                                             transition.source,

   163                                             transition.destination,

   164                                             transition, comment)

   165         # change state of context or new object

   166         registry = get_current_registry()

   167         registry.notify(event)

   168         # send modified event for original or new object

   169         if result is None:

   170             registry.notify(ObjectModifiedEvent(self.context))

   171         else:

   172             registry.notify(ObjectModifiedEvent(result))

   173         return result

   174 

   175     def fire_transition_toward(self, state, comment=None, side_effect=None, check_security=True):

   176         transition_ids = self.get_fireable_transition_ids_toward(state, check_security)

   177         if not transition_ids:

   178             raise NoTransitionAvailableError(self.state(self.context).get_state(), state)

   179         if len(transition_ids) != 1:

   180             raise AmbiguousTransitionError(self.state(self.context).get_state(), state)

   181         return self.fire_transition(transition_ids[0], comment, side_effect, check_security)

   182 

   183     def fire_transition_for_versions(self, state, transition_id, comment=None):

   184         versions = IWorkflowVersions(self.parent)

   185         for version in versions.get_versions(state):

   186             IWorkflowInfo(version).fire_transition(transition_id, comment)

   187 

   188     def fire_automatic(self):

   189         for transition_id in self.get_automatic_transition_ids():

   190             try:

   191                 self.fire_transition(transition_id)

   192             except ConditionFailedError:

   193                 # if condition failed, that's fine, then we weren't

   194                 # ready to fire yet

   195                 pass

   196             else:

   197                 # if we actually managed to fire a transition,

   198                 # we're done with this one now.

   199                 return

   200 

   201     def has_version(self, state):

   202         wf_versions = IWorkflowVersions(self.parent)

   203         return wf_versions.has_version(state)

   204 

   205     def get_manual_transition_ids(self, check_security=True):

   206         if check_security:

   207             request = check_request()

   208             permission_checker = request.has_permission

   209         else:

   210             permission_checker = granted_permission

   211         return [transition.transition_id

   212                 for transition in sorted(self._get_transitions(MANUAL),

   213                                          key=lambda x: x.user_data.get('order', 999))

   214                 if transition.condition(self, self.context) and

   215                     permission_checker(transition.permission, self.context)]

   216 

   217     def get_system_transition_ids(self):

   218         # ignore permission checks

   219         return [transition.transition_id

   220                 for transition in sorted(self._get_transitions(SYSTEM),

   221                                          key=lambda x: x.user_data.get('order', 999))

   222                 if transition.condition(self, self.context)]

   223 

   224     def get_fireable_transition_ids(self, check_security=True):

   225         return (self.get_manual_transition_ids(check_security) +

   226                 self.get_system_transition_ids())

   227 

   228     def get_fireable_transition_ids_toward(self, state, check_security=True):

   229         result = []

   230         for transition_id in self.get_fireable_transition_ids(check_security):

   231             transition = self.wf.get_transition_by_id(transition_id)

   232             if transition.destination == state:

   233                 result.append(transition_id)

   234         return result

   235 

   236     def get_automatic_transition_ids(self):

   237         return [transition.transition_id for transition in

   238                 self._get_transitions(AUTOMATIC)]

   239 

   240     def has_automatic_transitions(self):

   241         return bool(self.get_automatic_transition_ids())

   242 

   243     def _get_transitions(self, trigger):

   244         # retrieve all possible transitions from workflow utility

   245         state = IWorkflowState(self.context)

   246         transitions = self.wf.get_transitions(state.state)

   247         # now filter these transitions to retrieve all possible

   248         # transitions in this context, and return their ids

   249         return [transition for transition in transitions if transition.trigger == trigger]