"""Core hooks: workflow related hooks
:organization: Logilab
:copyright: 2001-2009 LOGILAB S.A. (Paris, FRANCE), license is LGPL v2.
:contact: http://www.logilab.fr/ -- mailto:contact@logilab.fr
:license: GNU Lesser General Public License, v2.1 - http://www.gnu.org/licenses
"""
__docformat__ = "restructuredtext en"
from datetime import datetime
from cubicweb import RepositoryError, ValidationError
from cubicweb.interfaces import IWorkflowable
from cubicweb.selectors import entity_implements
from cubicweb.server import hook
def _change_state(session, x, oldstate, newstate):
nocheck = session.transaction_data.setdefault('skip-security', set())
nocheck.add((x, 'in_state', oldstate))
nocheck.add((x, 'in_state', newstate))
# delete previous state first in case we're using a super session
session.delete_relation(x, 'in_state', oldstate)
session.add_relation(x, 'in_state', newstate)
class _SetInitialStateOp(hook.Operation):
"""make initial state be a default state"""
def precommit_event(self):
session = self.session
entity = self.entity
# if there is an initial state and the entity's state is not set,
# use the initial state as a default state
pendingeids = session.transaction_data.get('pendingeids', ())
if not (session.deleted_in_transaction(entity.eid) or entity.in_state) \
and entity.current_workflow:
state = entity.current_workflow.initial
if state:
# use super session to by-pass security checks
session.super_session.add_relation(entity.eid, 'in_state',
state.eid)
class _WorkflowChangedOp(hook.Operation):
"""fix entity current state when changing its workflow"""
def precommit_event(self):
session = self.session
if session.deleted_in_transaction(self.eid):
return
entity = session.entity_from_eid(self.eid)
# notice that enforcment that new workflow apply to the entity's type is
# done by schema rule, no need to check it here
if entity.current_workflow.eid == self.wfeid:
deststate = entity.current_workflow.initial
if not deststate:
msg = session._('workflow has no initial state')
raise ValidationError(entity.eid, {'custom_workflow': msg})
if entity.current_workflow.state_by_eid(entity.current_state.eid):
# nothing to do
return
# if there are no history, simply go to new workflow's initial state
if not entity.workflow_history:
if entity.current_state.eid != deststate.eid:
_change_state(session, entity.eid,
entity.current_state.eid, deststate.eid)
return
msg = session._('workflow changed to "%s"')
msg %= entity.current_workflow.name
entity.change_state(deststate.name, msg)
class WorkflowHook(hook.Hook):
__abstract__ = True
category = 'worfklow'
class SetInitialStateHook(WorkflowHook):
__id__ = 'wfsetinitial'
__select__ = WorkflowHook.__select__ & entity_implements(IWorkflowable)
events = ('after_add_entity',)
def __call__(self):
_SetInitialStateOp(self._cw, entity=self.entity)
class PrepareStateChangeHook(WorkflowHook):
"""record previous state information"""
__id__ = 'cwdelstate'
__select__ = WorkflowHook.__select__ & hook.match_rtype('in_state')
events = ('before_delete_relation',)
def __call__(self):
self._cw.transaction_data.setdefault('pendingrelations', []).append(
(self.eidfrom, self.rtype, self.eidto))
class FireTransitionHook(WorkflowHook):
"""check the transition is allowed, add missing information. Expect that:
* wf_info_for inlined relation is set
* by_transition or to_state (managers only) inlined relation is set
"""
__id__ = 'wffiretransition'
__select__ = WorkflowHook.__select__ & entity_implements('TrInfo')
events = ('before_add_entity',)
def __call__(self):
session = self._cw
entity = self.entity
# first retreive entity to which the state change apply
try:
foreid = entity['wf_info_for']
except KeyError:
msg = session._('mandatory relation')
raise ValidationError(entity.eid, {'wf_info_for': msg})
forentity = session.entity_from_eid(foreid)
# then check it has a workflow set
wf = forentity.current_workflow
if wf is None:
msg = session._('related entity has no workflow set')
raise ValidationError(entity.eid, {None: msg})
# then check it has a state set
fromstate = forentity.current_state
if fromstate is None:
msg = session._('related entity has no state')
raise ValidationError(entity.eid, {None: msg})
# no investigate the requested state change...
try:
treid = entity['by_transition']
except KeyError:
# no transition set, check user is a manager and destination state is
# specified (and valid)
if not (session.is_super_session or 'managers' in session.user.groups):
msg = session._('mandatory relation')
raise ValidationError(entity.eid, {'by_transition': msg})
deststateeid = entity.get('to_state')
if not deststateeid:
msg = session._('mandatory relation')
raise ValidationError(entity.eid, {'by_transition': msg})
deststate = wf.state_by_eid(deststateeid)
if deststate is None:
msg = session._("state doesn't belong to entity's workflow")
raise ValidationError(entity.eid, {'to_state': msg})
else:
# check transition is valid and allowed
tr = wf.transition_by_eid(treid)
if tr is None:
msg = session._("transition doesn't belong to entity's workflow")
raise ValidationError(entity.eid, {'by_transition': msg})
if not tr.has_input_state(fromstate):
msg = session._("transition isn't allowed")
raise ValidationError(entity.eid, {'by_transition': msg})
if not tr.may_be_fired(foreid):
msg = session._("transition may not be fired")
raise ValidationError(entity.eid, {'by_transition': msg})
deststateeid = tr.destination().eid
# everything is ok, add missing information on the trinfo entity
entity['from_state'] = fromstate.eid
entity['to_state'] = deststateeid
nocheck = session.transaction_data.setdefault('skip-security', set())
nocheck.add((entity.eid, 'from_state', fromstate.eid))
nocheck.add((entity.eid, 'to_state', deststateeid))
class FiredTransitionHook(WorkflowHook):
"""change related entity state"""
__id__ = 'wffiretransition'
__select__ = WorkflowHook.__select__ & entity_implements('TrInfo')
events = ('after_add_entity',)
def __call__(self):
_change_state(self._cw, self.entity['wf_info_for'],
self.entity['from_state'], self.entity['to_state'])
class SetModificationDateOnStateChange(WorkflowHook):
"""update entity's modification date after changing its state"""
__id__ = 'wfsyncmdate'
__select__ = WorkflowHook.__select__ & hook.match_rtype('in_state')
events = ('after_add_relation',)
def __call__(self):
if self._cw.added_in_transaction(self.eidfrom):
# new entity, not needed
return
entity = self._cw.entity_from_eid(self.eidfrom)
try:
entity.set_attributes(modification_date=datetime.now(),
_cw_unsafe=True)
except RepositoryError, ex:
# usually occurs if entity is coming from a read-only source
# (eg ldap user)
self.warning('cant change modification date for %s: %s', entity, ex)
class SetCustomWorkflow(WorkflowHook):
__id__ = 'wfsetcustom'
__select__ = WorkflowHook.__select__ & hook.match_rtype('custom_workflow')
events = ('after_add_relation',)
def __call__(self):
_WorkflowChangedOp(self._cw, eid=self.eidfrom, wfeid=self.eidto)
class DelCustomWorkflow(SetCustomWorkflow):
__id__ = 'wfdelcustom'
events = ('after_delete_relation',)
def __call__(self):
entity = self._cw.entity_from_eid(self.eidfrom)
typewf = entity.cwetype_workflow()
if typewf is not None:
_WorkflowChangedOp(self._cw, eid=self.eidfrom, wfeid=typewf.eid)
class DelWorkflowHook(WorkflowHook):
__id__ = 'wfdel'
__select__ = WorkflowHook.__select__ & entity_implements('Workflow')
events = ('after_delete_entity',)
def __call__(self):
# cleanup unused state and transition
self._cw.execute('DELETE State X WHERE NOT X state_of Y')
self._cw.execute('DELETE Transition X WHERE NOT X transition_of Y')