Skip to content

erp5
erp5
Commit 8e10c692 authored by Nicolas Dumazet's avatar Nicolas Dumazet
Browse files
Options

identation: use 2-spaces everywhere instead of a 2/4/6 mix 


git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@42722 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent a38c1884
Hide whitespace changes
Inline Side-by-side
Showing with 285 additions and 285 deletions
product/ERP5/InteractionWorkflow.py
View file @ 8e10c692
......@@ -34,320 +34,320 @@ from Products.ERP5Type import Permissions
_MARKER = []
class InteractionWorkflowDefinition (DCWorkflowDefinition, ActiveObject):
"""
The InteractionTool implements portal object
interaction policies.
"""
The InteractionTool implements portal object
interaction policies.
An interaction is defined by
a domain and a behaviour:
The domain is defined as:
- the meta_type it applies to
- the portal_type it applies to
- the conditions of application (category membership, value range,
security, function, etc.)
An interaction is defined by
a domain and a behaviour:
The transformation template is defined as:
The domain is defined as:
- pre method executed before
- the meta_type it applies to
- pre async executed anyway
- the portal_type it applies to
- post method executed after success before return
- the conditions of application (category membership, value range,
security, function, etc.)
- post method executed after success anyway
The transformation template is defined as:
This is similar to signals and slots except is applies to classes
rather than instances. Similar to
stateless workflow methods with more options. Similar to ZSQL scipts
but in more cases.
- pre method executed before
Examples of applications:
- pre async executed anyway
- when movement is updated, apply transformation rules to movement
- post method executed after success before return
- when stock is 0, post an event of stock empty
- post method executed after success anyway
- when birthday is called, call the happy birthday script
This is similar to signals and slots except is applies to classes
rather than instances. Similar to
stateless workflow methods with more options. Similar to ZSQL scipts
but in more cases.
ERP5 main application: specialize behaviour of classes "on the fly".
Make the architecture as modular as possible. Implement connections
a la Qt.
Examples of applications:
Try to mimic: Workflow...
- when movement is updated, apply transformation rules to movement
Question: should be use it for values ? or use a global value model ?
- when stock is 0, post an event of stock empty
Status : OK
- when birthday is called, call the happy birthday script
ERP5 main application: specialize behaviour of classes "on the fly".
Make the architecture as modular as possible. Implement connections
a la Qt.
Try to mimic: Workflow...
Implementation:
Question: should be use it for values ? or use a global value model ?
A new kind of workflow (stateless). Follow the DCWorkflow class.
Provide filters (per portal_type, etc.). Allow inspection of objects ?
"""
meta_type = 'Workflow'
title = 'Interaction Workflow Definition'
Status : OK
interactions = None
security = ClassSecurityInfo()
Implementation:
manage_options = (
{'label': 'Properties', 'action': 'manage_properties'},
{'label': 'Interactions', 'action': 'interactions/manage_main'},
{'label': 'Variables', 'action': 'variables/manage_main'},
{'label': 'Scripts', 'action': 'scripts/manage_main'},
) + App.Undo.UndoSupport.manage_options
A new kind of workflow (stateless). Follow the DCWorkflow class.
Provide filters (per portal_type, etc.). Allow inspection of objects ?
def __init__(self, id):
self.id = id
from Interaction import Interaction
self._addObject(Interaction('interactions'))
from Products.DCWorkflow.Variables import Variables
self._addObject(Variables('variables'))
from Products.DCWorkflow.Worklists import Worklists
self._addObject(Worklists('worklists'))
from Products.DCWorkflow.Scripts import Scripts
self._addObject(Scripts('scripts'))
security.declareProtected(Permissions.View, 'getChainedPortalTypeList')
def getChainedPortalTypeList(self):
"""Returns the list of portal types that are chained to this
interaction workflow."""
chained_ptype_list = []
wf_tool = getToolByName(self, 'portal_workflow')
types_tool = getToolByName(self, 'portal_types')
for ptype in types_tool.objectIds():
if self.getId() in wf_tool._chains_by_type.get(ptype, []) :
chained_ptype_list.append(ptype)
return chained_ptype_list
security.declarePrivate('listObjectActions')
def listObjectActions(self, info):
return []
security.declarePrivate('_changeStateOf')
def _changeStateOf(self, ob, tdef=None, kwargs=None) :
"""
InteractionWorkflow is stateless. Thus, this function should do nothing.
"""
return
security.declarePrivate('isInfoSupported')
def isInfoSupported(self, ob, name):
'''
Returns a true value if the given info name is supported.
'''
vdef = self.variables.get(name, None)
if vdef is None:
return 0
return 1
security.declarePrivate('getInfoFor')
def getInfoFor(self, ob, name, default):
'''
Allows the user to request information provided by the
workflow. This method must perform its own security checks.
'''
vdef = self.variables.get(name, _MARKER)
if vdef is _MARKER:
return default
if vdef.info_guard is not None and not vdef.info_guard.check(
getSecurityManager(), self, ob):
return default
status = self._getStatusOf(ob)
if status is not None and status.has_key(name):
value = status[name]
# Not set yet. Use a default.
elif vdef.default_expr is not None:
ec = createExprContext(StateChangeInfo(ob, self, status))
value = vdef.default_expr(ec)
else:
value = vdef.default_value
return value
security.declarePrivate('isWorkflowMethodSupported')
def isWorkflowMethodSupported(self, ob, method_id):
'''
Returns a true value if the given workflow is
automatic with the propper method_id
NOTE: this method is not used in ERP5 because
of transition_list approach
'''
tdef = self.interactions.get(method_id, None)
if (tdef is not None and
tdef.trigger_type == TRIGGER_WORKFLOW_METHOD and
self._checkTransitionGuard(tdef, ob)):
return 1
return 0
security.declarePrivate('wrapWorkflowMethod')
def wrapWorkflowMethod(self, ob, method_id, func, args, kw):
'''
Allows the user to request a workflow action. This method
must perform its own security checks.
'''
return
security.declarePrivate('notifyWorkflowMethod')
def notifyWorkflowMethod(self, ob, transition_list, args=None, kw=None):
"""
meta_type = 'Workflow'
title = 'Interaction Workflow Definition'
interactions = None
security = ClassSecurityInfo()
manage_options = (
{'label': 'Properties', 'action': 'manage_properties'},
{'label': 'Interactions', 'action': 'interactions/manage_main'},
{'label': 'Variables', 'action': 'variables/manage_main'},
{'label': 'Scripts', 'action': 'scripts/manage_main'},
) + App.Undo.UndoSupport.manage_options
def __init__(self, id):
self.id = id
from Interaction import Interaction
self._addObject(Interaction('interactions'))
from Products.DCWorkflow.Variables import Variables
self._addObject(Variables('variables'))
from Products.DCWorkflow.Worklists import Worklists
self._addObject(Worklists('worklists'))
from Products.DCWorkflow.Scripts import Scripts
self._addObject(Scripts('scripts'))
security.declareProtected(Permissions.View, 'getChainedPortalTypeList')
def getChainedPortalTypeList(self):
"""Returns the list of portal types that are chained to this
interaction workflow."""
chained_ptype_list = []
wf_tool = getToolByName(self, 'portal_workflow')
types_tool = getToolByName(self, 'portal_types')
for ptype in types_tool.objectIds():
if self.getId() in wf_tool._chains_by_type.get(ptype, []) :
chained_ptype_list.append(ptype)
return chained_ptype_list
security.declarePrivate('listObjectActions')
def listObjectActions(self, info):
return []
security.declarePrivate('_changeStateOf')
def _changeStateOf(self, ob, tdef=None, kwargs=None) :
"""
InteractionWorkflow is stateless. Thus, this function should do nothing.
"""
InteractionWorkflow is stateless. Thus, this function should do nothing.
"""
return
security.declarePrivate('notifyBefore')
def notifyBefore(self, ob, transition_list, args=None, kw=None):
'''
Notifies this workflow of an action before it happens,
allowing veto by exception. Unless an exception is thrown, either
a notifySuccess() or notifyException() can be expected later on.
The action usually corresponds to a method name.
'''
if type(transition_list) in StringTypes:
return
security.declarePrivate('isInfoSupported')
def isInfoSupported(self, ob, name):
'''
Returns a true value if the given info name is supported.
'''
vdef = self.variables.get(name, None)
if vdef is None:
return 0
return 1
security.declarePrivate('getInfoFor')
def getInfoFor(self, ob, name, default):
'''
Allows the user to request information provided by the
workflow. This method must perform its own security checks.
'''
vdef = self.variables.get(name, _MARKER)
if vdef is _MARKER:
return default
if vdef.info_guard is not None and not vdef.info_guard.check(
getSecurityManager(), self, ob):
return default
status = self._getStatusOf(ob)
if status is not None and status.has_key(name):
value = status[name]
# Not set yet. Use a default.
elif vdef.default_expr is not None:
ec = createExprContext(StateChangeInfo(ob, self, status))
value = vdef.default_expr(ec)
else:
value = vdef.default_value
# Wrap args into kw since this is the only way
# to be compatible with DCWorkflow
# A better approach consists in extending DCWorkflow
if kw is None:
kw = {'workflow_method_args' : args}
else:
kw = kw.copy()
kw['workflow_method_args'] = args
filtered_transition_list = []
for t_id in transition_list:
tdef = self.interactions[t_id]
assert tdef.trigger_type == TRIGGER_WORKFLOW_METHOD
if (tdef.portal_type_filter is None or \
ob.getPortalType() in tdef.portal_type_filter) and \
self._checkTransitionGuard(tdef, ob, **kw):
filtered_transition_list.append(tdef.id)
former_status = self._getStatusOf(ob)
# Execute the "before" script.
for script_name in tdef.script_name:
script = self.scripts[script_name]
# Pass lots of info to the script in a single parameter.
sci = StateChangeInfo(
ob, self, former_status, tdef, None, None, kwargs=kw)
script(sci) # May throw an exception
return filtered_transition_list
return value
security.declarePrivate('isWorkflowMethodSupported')
def isWorkflowMethodSupported(self, ob, method_id):
'''
Returns a true value if the given workflow is
automatic with the propper method_id
NOTE: this method is not used in ERP5 because
of transition_list approach
'''
tdef = self.interactions.get(method_id, None)
if (tdef is not None and
tdef.trigger_type == TRIGGER_WORKFLOW_METHOD and
self._checkTransitionGuard(tdef, ob)):
return 1
return 0
security.declarePrivate('wrapWorkflowMethod')
def wrapWorkflowMethod(self, ob, method_id, func, args, kw):
'''
Allows the user to request a workflow action. This method
must perform its own security checks.
'''
return
security.declarePrivate('notifyWorkflowMethod')
def notifyWorkflowMethod(self, ob, transition_list, args=None, kw=None):
"""
InteractionWorkflow is stateless. Thus, this function should do nothing.
"""
security.declarePrivate('notifySuccess')
def notifySuccess(self, ob, transition_list, result, args=None, kw=None):
'''
Notifies this workflow that an action has taken place.
'''
if type(transition_list) in StringTypes:
return
security.declarePrivate('notifyBefore')
def notifyBefore(self, ob, transition_list, args=None, kw=None):
'''
Notifies this workflow of an action before it happens,
allowing veto by exception. Unless an exception is thrown, either
a notifySuccess() or notifyException() can be expected later on.
The action usually corresponds to a method name.
'''
if type(transition_list) in StringTypes:
return
# Wrap args into kw since this is the only way
# to be compatible with DCWorkflow
# A better approach consists in extending DCWorkflow
if kw is None:
kw = {'workflow_method_args' : args}
kw = kw.copy()
kw['workflow_method_args'] = args
kw['workflow_method_result'] = result
for t_id in transition_list:
tdef = self.interactions[t_id]
assert tdef.trigger_type == TRIGGER_WORKFLOW_METHOD
if (tdef.portal_type_filter is not None and \
ob.getPortalType() not in tdef.portal_type_filter):
continue
# Initialize variables
former_status = self._getStatusOf(ob)
econtext = None
sci = None
# Update variables.
tdef_exprs = tdef.var_exprs
if tdef_exprs is None: tdef_exprs = {}
status = {}
for id, vdef in self.variables.items():
if not vdef.for_status:
continue
expr = None
if tdef_exprs.has_key(id):
expr = tdef_exprs[id]
elif not vdef.update_always and former_status.has_key(id):
# Preserve former value
value = former_status[id]
else:
kw = kw.copy()
kw['workflow_method_args'] = args
filtered_transition_list = []
for t_id in transition_list:
tdef = self.interactions[t_id]
assert tdef.trigger_type == TRIGGER_WORKFLOW_METHOD
if (tdef.portal_type_filter is None or \
ob.getPortalType() in tdef.portal_type_filter) and \
self._checkTransitionGuard(tdef, ob, **kw):
filtered_transition_list.append(tdef.id)
former_status = self._getStatusOf(ob)
# Execute the "before" script.
for script_name in tdef.script_name:
script = self.scripts[script_name]
# Pass lots of info to the script in a single parameter.
if vdef.default_expr is not None:
expr = vdef.default_expr
else:
value = vdef.default_value
if expr is not None:
# Evaluate an expression.
if econtext is None:
# Lazily create the expression context.
if sci is None:
sci = StateChangeInfo(
ob, self, former_status, tdef, None, None, kwargs=kw)
script(sci) # May throw an exception
return filtered_transition_list
security.declarePrivate('notifySuccess')
def notifySuccess(self, ob, transition_list, result, args=None, kw=None):
'''
Notifies this workflow that an action has taken place.
'''
if type(transition_list) in StringTypes:
return
kw = kw.copy()
kw['workflow_method_args'] = args
kw['workflow_method_result'] = result
for t_id in transition_list:
tdef = self.interactions[t_id]
assert tdef.trigger_type == TRIGGER_WORKFLOW_METHOD
if (tdef.portal_type_filter is not None and \
ob.getPortalType() not in tdef.portal_type_filter):
continue
# Initialize variables
former_status = self._getStatusOf(ob)
econtext = None
sci = None
# Update variables.
tdef_exprs = tdef.var_exprs
if tdef_exprs is None: tdef_exprs = {}
status = {}
for id, vdef in self.variables.items():
if not vdef.for_status:
continue
expr = None
if tdef_exprs.has_key(id):
expr = tdef_exprs[id]
elif not vdef.update_always and former_status.has_key(id):
# Preserve former value
value = former_status[id]
else:
if vdef.default_expr is not None:
expr = vdef.default_expr
else:
value = vdef.default_value
if expr is not None:
# Evaluate an expression.
if econtext is None:
# Lazily create the expression context.
if sci is None:
sci = StateChangeInfo(
ob, self, former_status, tdef,
None, None, None)
econtext = createExprContext(sci)
value = expr(econtext)
status[id] = value
sci = StateChangeInfo(
ob, self, former_status, tdef, None, None, kwargs=kw)
# Execute the "after" script.
for script_name in tdef.after_script_name:
script = self.scripts[script_name]
# Pass lots of info to the script in a single parameter.
script(sci) # May throw an exception
# Execute Before Commit
for script_name in tdef.before_commit_script_name:
transaction.get().addBeforeCommitHook(self._before_commit,
(sci, script_name))
# Execute "activity" scripts
for script_name in tdef.activate_script_name:
self.activate(activity='SQLQueue')\
.activeScript(script_name, ob.getRelativeUrl(),
status, tdef.id)
def _before_commit(self, sci, script_name):
# check the object still exists before calling the script
ob = sci.object
while ob.isTempObject():
ob = ob.getParentValue()
if aq_base(self.unrestrictedTraverse(ob.getPhysicalPath(), None)) is \
aq_base(ob):
self.scripts[script_name](sci)
security.declarePrivate('activeScript')
def activeScript(self, script_name, ob_url, status, tdef_id):
script = self.scripts[script_name]
ob = self.unrestrictedTraverse(ob_url)
tdef = self.interactions.get(tdef_id)
sci = StateChangeInfo(
ob, self, status, tdef, None, None, None)
script(sci)
def _getWorkflowStateOf(self, ob, id_only=0):
return None
def _checkTransitionGuard(self, t, ob, **kw):
# This check can be implemented with a guard expression, but
# it has a lot of overhead to use a TALES, so we make a special
# treatment for the frequent case, that is, disallow the trigger
# on a temporary document.
if t.temporary_document_disallowed:
isTempDocument = getattr(ob, 'isTempDocument', None)
if isTempDocument is not None:
if isTempDocument():
return 0
return DCWorkflowDefinition._checkTransitionGuard(self, t, ob, **kw)
ob, self, former_status, tdef,
None, None, None)
econtext = createExprContext(sci)
value = expr(econtext)
status[id] = value
sci = StateChangeInfo(
ob, self, former_status, tdef, None, None, kwargs=kw)
# Execute the "after" script.
for script_name in tdef.after_script_name:
script = self.scripts[script_name]
# Pass lots of info to the script in a single parameter.
script(sci) # May throw an exception
# Execute Before Commit
for script_name in tdef.before_commit_script_name:
transaction.get().addBeforeCommitHook(self._before_commit,
(sci, script_name))
# Execute "activity" scripts
for script_name in tdef.activate_script_name:
self.activate(activity='SQLQueue')\
.activeScript(script_name, ob.getRelativeUrl(),
status, tdef.id)
def _before_commit(self, sci, script_name):
# check the object still exists before calling the script
ob = sci.object
while ob.isTempObject():
ob = ob.getParentValue()
if aq_base(self.unrestrictedTraverse(ob.getPhysicalPath(), None)) is \
aq_base(ob):
self.scripts[script_name](sci)
security.declarePrivate('activeScript')
def activeScript(self, script_name, ob_url, status, tdef_id):
script = self.scripts[script_name]
ob = self.unrestrictedTraverse(ob_url)
tdef = self.interactions.get(tdef_id)
sci = StateChangeInfo(
ob, self, status, tdef, None, None, None)
script(sci)
def _getWorkflowStateOf(self, ob, id_only=0):
return None
def _checkTransitionGuard(self, t, ob, **kw):
# This check can be implemented with a guard expression, but
# it has a lot of overhead to use a TALES, so we make a special
# treatment for the frequent case, that is, disallow the trigger
# on a temporary document.
if t.temporary_document_disallowed:
isTempDocument = getattr(ob, 'isTempDocument', None)
if isTempDocument is not None:
if isTempDocument():
return 0
return DCWorkflowDefinition._checkTransitionGuard(self, t, ob, **kw)
Globals.InitializeClass(InteractionWorkflowDefinition)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7