cubicweb: comparison entities/wfobjs.py

equal deleted inserted replaced

-:a64f48dd5fe4
+:9ab2b4c74baf
 # FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
 # details.
 #
 # You should have received a copy of the GNU Lesser General Public License along
 # with CubicWeb.  If not, see <http://www.gnu.org/licenses/>.
-"""workflow definition and history related entities
+"""workflow handling:
+* entity types defining workflow (Workflow, State, Transition...)
+* workflow history (TrInfo)
+* adapter for workflowable entities (IWorkflowableAdapter)
 """
 __docformat__ = "restructuredtext en"
 from warnings import warn
 from logilab.common.decorators import cached, clear_cache
 from logilab.common.deprecation import deprecated
 from logilab.common.compat import any
 from cubicweb.entities import AnyEntity, fetch_config
-from cubicweb.interfaces import IWorkflowable
+from cubicweb.view import EntityAdapter
+from cubicweb.selectors import relation_possible
 from cubicweb.mixins import MI_REL_TRIGGERS
 class WorkflowException(Exception): pass
 class Workflow(AnyEntity):
 """return True if this workflow is the default workflow for the given
 entity type
 """
 return any(et for et in self.reverse_default_workflow
 if et.name == etype)
-# XXX define parent() instead? what if workflow of multiple types?
-def after_deletion_path(self):
-"""return (path, parameters) which should be used as redirect
-information when this entity is being deleted
-"""
-if self.workflow_of:
-return self.workflow_of[0].rest_path(), {'vid': 'workflow'}
-return super(Workflow, self).after_deletion_path()
 def iter_workflows(self, _done=None):
 """return an iterator on actual workflows, eg this workflow and its
 subworkflows
 """
 if rqlexpr.check_expression(self._cw, eid):
 return True
 if self.condition or groups:
 return False
 return True
-def after_deletion_path(self):
-"""return (path, parameters) which should be used as redirect
-information when this entity is being deleted
-"""
-if self.transition_of:
-return self.transition_of[0].rest_path(), {}
-return super(BaseTransition, self).after_deletion_path()
 def set_permissions(self, requiredgroups=(), conditions=(), reset=True):
 """set or add (if `reset` is False) groups and conditions for this
 transition
 """
 def destination(self, entity):
 try:
 return self.destination_state[0]
 except IndexError:
-return entity.latest_trinfo().previous_state
+return entity.cw_adapt_to('IWorkflowable').latest_trinfo().previous_state
 def potential_destinations(self):
 try:
 yield self.destination_state[0]
 except IndexError:
 for incomingstate in self.reverse_allowed_transition:
 for tr in incomingstate.reverse_destination_state:
 for previousstate in tr.reverse_allowed_transition:
 yield previousstate
-def parent(self):
-return self.workflow
 class WorkflowTransition(BaseTransition):
 """customized class for WorkflowTransition entities"""
 __regid__ = 'WorkflowTransition'
 tostateeid = self.exit_points()[stateeid]
 except KeyError:
 return None
 if tostateeid is None:
 # go back to state from which we've entered the subworkflow
-return entity.subworkflow_input_trinfo().previous_state
+return entity.cw_adapt_to('IWorkflowable').subworkflow_input_trinfo().previous_state
 return self._cw.entity_from_eid(tostateeid)
 @cached
 def exit_points(self):
 result = {}
 @property
 def destination(self):
 return self.destination_state and self.destination_state[0] or None
-def parent(self):
-return self.reverse_subworkflow_exit[0]
 class State(AnyEntity):
 """customized class for State entities"""
 __regid__ = 'State'
 fetch_attrs, fetch_order = fetch_config(['name'])
 rest_attr = 'eid'
 @property
 def workflow(self):
 # take care, may be missing in multi-sources configuration
-return self.state_of and self.state_of[0]
+return self.state_of and self.state_of[0] or None
-def parent(self):
-return self.workflow
 class TrInfo(AnyEntity):
 """customized class for Transition information entities
 """
 @property
 def transition(self):
 return self.by_transition and self.by_transition[0] or None
-def parent(self):
-return self.for_entity
 class WorkflowableMixIn(object):
 """base mixin providing workflow helper methods for workflowable entities.
 This mixin will be automatically set on class supporting the 'in_state'
 relation (which implies supporting 'wf_info_for' as well)
 """
-__implements__ = (IWorkflowable,)
+@property
+@deprecated('[3.5] use printable_state')
+def displayable_state(self):
+return self._cw._(self.state)
+@property
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').main_workflow")
+def main_workflow(self):
+return self.cw_adapt_to('IWorkflowable').main_workflow
+@property
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').current_workflow")
+def current_workflow(self):
+return self.cw_adapt_to('IWorkflowable').current_workflow
+@property
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').current_state")
+def current_state(self):
+return self.cw_adapt_to('IWorkflowable').current_state
+@property
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').state")
+def state(self):
+return self.cw_adapt_to('IWorkflowable').state
+@property
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').printable_state")
+def printable_state(self):
+return self.cw_adapt_to('IWorkflowable').printable_state
+@property
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').workflow_history")
+def workflow_history(self):
+return self.cw_adapt_to('IWorkflowable').workflow_history
+@deprecated('[3.5] get transition from current workflow and use its may_be_fired method')
+def can_pass_transition(self, trname):
+"""return the Transition instance if the current user can fire the
+transition with the given name, else None
+"""
+tr = self.current_workflow and self.current_workflow.transition_by_name(trname)
+if tr and tr.may_be_fired(self.eid):
+return tr
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').cwetype_workflow()")
+def cwetype_workflow(self):
+return self.cw_adapt_to('IWorkflowable').main_workflow()
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').latest_trinfo()")
+def latest_trinfo(self):
+return self.cw_adapt_to('IWorkflowable').latest_trinfo()
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').possible_transitions()")
+def possible_transitions(self, type='normal'):
+return self.cw_adapt_to('IWorkflowable').possible_transitions(type)
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').fire_transition()")
+def fire_transition(self, tr, comment=None, commentformat=None):
+return self.cw_adapt_to('IWorkflowable').fire_transition(tr, comment, commentformat)
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').change_state()")
+def change_state(self, statename, comment=None, commentformat=None, tr=None):
+return self.cw_adapt_to('IWorkflowable').change_state(statename, comment, commentformat, tr)
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').subworkflow_input_trinfo()")
+def subworkflow_input_trinfo(self):
+return self.cw_adapt_to('IWorkflowable').subworkflow_input_trinfo()
+@deprecated("[3.9] use entity.cw_adapt_to('IWorkflowable').subworkflow_input_transition()")
+def subworkflow_input_transition(self):
+return self.cw_adapt_to('IWorkflowable').subworkflow_input_transition()
+MI_REL_TRIGGERS[('in_state', 'subject')] = WorkflowableMixIn
+class IWorkflowableAdapter(WorkflowableMixIn, EntityAdapter):
+"""base adapter providing workflow helper methods for workflowable entities.
+"""
+__regid__ = 'IWorkflowable'
+__select__ = relation_possible('in_state')
+@cached
+def cwetype_workflow(self):
+"""return the default workflow for entities of this type"""
+# XXX CWEType method
+wfrset = self._cw.execute('Any WF WHERE ET default_workflow WF, '
+'ET name %(et)s', {'et': self.entity.__regid__})
+if wfrset:
+return wfrset.get_entity(0, 0)
+self.warning("can't find any workflow for %s", self.entity.__regid__)
+return None
 @property
 def main_workflow(self):
 """return current workflow applied to this entity"""
-if self.custom_workflow:
+if self.entity.custom_workflow:
-return self.custom_workflow[0]
+return self.entity.custom_workflow[0]
 return self.cwetype_workflow()
 @property
 def current_workflow(self):
 """return current workflow applied to this entity"""
 return self.current_state and self.current_state.workflow or self.main_workflow
 @property
 def current_state(self):
 """return current state entity"""
-return self.in_state and self.in_state[0] or None
+return self.entity.in_state and self.entity.in_state[0] or None
 @property
 def state(self):
 """return current state name"""
 try:
-return self.in_state[0].name
+return self.current_state.name
-except IndexError:
+except AttributeError:
 self.warning('entity %s has no state', self)
 return None
 @property
 def printable_state(self):
 @property
 def workflow_history(self):
 """return the workflow history for this entity (eg ordered list of
 TrInfo entities)
 """
-return self.reverse_wf_info_for
+return self.entity.reverse_wf_info_for
 def latest_trinfo(self):
 """return the latest transition information for this entity"""
 try:
-return self.reverse_wf_info_for[-1]
+return self.workflow_history[-1]
 except IndexError:
 return None
-@cached
-def cwetype_workflow(self):
-"""return the default workflow for entities of this type"""
-# XXX CWEType method
-wfrset = self._cw.execute('Any WF WHERE ET default_workflow WF, '
-'ET name %(et)s', {'et': self.__regid__})
-if wfrset:
-return wfrset.get_entity(0, 0)
-self.warning("can't find any workflow for %s", self.__regid__)
-return None
 def possible_transitions(self, type='normal'):
 """generates transition that MAY be fired for the given entity,
 expected to be in this state
 used only by the UI
 'T type TT, T type %(type)s, '
 'T name TN, T transition_of WF, WF eid %(wfeid)s',
 {'x': self.current_state.eid, 'type': type,
 'wfeid': self.current_workflow.eid})
 for tr in rset.entities():
-if tr.may_be_fired(self.eid):
+if tr.may_be_fired(self.entity.eid):
 yield tr
-def _add_trinfo(self, comment, commentformat, treid=None, tseid=None):
-kwargs = {}
-if comment is not None:
-kwargs['comment'] = comment
-if commentformat is not None:
-kwargs['comment_format'] = commentformat
-kwargs['wf_info_for'] = self
-if treid is not None:
-kwargs['by_transition'] = self._cw.entity_from_eid(treid)
-if tseid is not None:
-kwargs['to_state'] = self._cw.entity_from_eid(tseid)
-return self._cw.create_entity('TrInfo', **kwargs)
-def fire_transition(self, tr, comment=None, commentformat=None):
-"""change the entity's state by firing transition of the given name in
-entity's workflow
-"""
-assert self.current_workflow
-if isinstance(tr, basestring):
-_tr = self.current_workflow.transition_by_name(tr)
-assert _tr is not None, 'not a %s transition: %s' % (
-self.__regid__, tr)
-tr = _tr
-return self._add_trinfo(comment, commentformat, tr.eid)
-def change_state(self, statename, comment=None, commentformat=None, tr=None):
-"""change the entity's state to the given state (name or entity) in
-entity's workflow. This method should only by used by manager to fix an
-entity's state when their is no matching transition, otherwise
-fire_transition should be used.
-"""
-assert self.current_workflow
-if hasattr(statename, 'eid'):
-stateeid = statename.eid
-else:
-if not isinstance(statename, basestring):
-warn('[3.5] give a state name', DeprecationWarning)
-state = self.current_workflow.state_by_eid(statename)
-else:
-state = self.current_workflow.state_by_name(statename)
-if state is None:
-raise WorkflowException('not a %s state: %s' % (self.__regid__,
-statename))
-stateeid = state.eid
-# XXX try to find matching transition?
-return self._add_trinfo(comment, commentformat, tr and tr.eid, stateeid)
 def subworkflow_input_trinfo(self):
 """return the TrInfo which has be recorded when this entity went into
 the current sub-workflow
 """
 def subworkflow_input_transition(self):
 """return the transition which has went through the current sub-workflow
 """
 return getattr(self.subworkflow_input_trinfo(), 'transition', None)
-def clear_all_caches(self):
+def _add_trinfo(self, comment, commentformat, treid=None, tseid=None):
-super(WorkflowableMixIn, self).clear_all_caches()
+kwargs = {}
-clear_cache(self, 'cwetype_workflow')
+if comment is not None:
+kwargs['comment'] = comment
-@deprecated('[3.5] get transition from current workflow and use its may_be_fired method')
+if commentformat is not None:
-def can_pass_transition(self, trname):
+kwargs['comment_format'] = commentformat
-"""return the Transition instance if the current user can fire the
+kwargs['wf_info_for'] = self.entity
-transition with the given name, else None
+if treid is not None:
-"""
+kwargs['by_transition'] = self._cw.entity_from_eid(treid)
-tr = self.current_workflow and self.current_workflow.transition_by_name(trname)
+if tseid is not None:
-if tr and tr.may_be_fired(self.eid):
+kwargs['to_state'] = self._cw.entity_from_eid(tseid)
-return tr
+return self._cw.create_entity('TrInfo', **kwargs)
-@property
+def fire_transition(self, tr, comment=None, commentformat=None):
-@deprecated('[3.5] use printable_state')
+"""change the entity's state by firing transition of the given name in
-def displayable_state(self):
+entity's workflow
-return self._cw._(self.state)
+"""
+assert self.current_workflow
-MI_REL_TRIGGERS[('in_state', 'subject')] = WorkflowableMixIn
+if isinstance(tr, basestring):
+_tr = self.current_workflow.transition_by_name(tr)
+assert _tr is not None, 'not a %s transition: %s' % (
+self.__regid__, tr)
+tr = _tr
+return self._add_trinfo(comment, commentformat, tr.eid)
+def change_state(self, statename, comment=None, commentformat=None, tr=None):
+"""change the entity's state to the given state (name or entity) in
+entity's workflow. This method should only by used by manager to fix an
+entity's state when their is no matching transition, otherwise
+fire_transition should be used.
+"""
+assert self.current_workflow
+if hasattr(statename, 'eid'):
+stateeid = statename.eid
+else:
+if not isinstance(statename, basestring):
+warn('[3.5] give a state name', DeprecationWarning)
+state = self.current_workflow.state_by_eid(statename)
+else:
+state = self.current_workflow.state_by_name(statename)
+if state is None:
+raise WorkflowException('not a %s state: %s' % (self.__regid__,
+statename))
+stateeid = state.eid
+# XXX try to find matching transition?
+return self._add_trinfo(comment, commentformat, tr and tr.eid, stateeid)

changeset 5556	9ab2b4c74baf
parent 5426	0d4853a6e5ee
child 5557	1a534c596bff