[Zope3-checkins] CVS: Zope3/src/zope/app/workflow/stateful - __init__.py:1.1 configure.zcml:1.1 contentworkflow.py:1.1 definition.py:1.1 instance.py:1.1 xmlexport_template.pt:1.1 xmlimportexport.py:1.1
Ulrich Eck
ueck@net-labs.de
Thu, 8 May 2003 13:27:20 -0400
Update of /cvs-repository/Zope3/src/zope/app/workflow/stateful
In directory cvs.zope.org:/tmp/cvs-serv7538/src/zope/app/workflow/stateful
Added Files:
__init__.py configure.zcml contentworkflow.py definition.py
instance.py xmlexport_template.pt xmlimportexport.py
Log Message:
Finally got it into Zope3:
Workflow has arrived!
this is a merge of the workflow package that was seperatly developed
at /Packages3/workflow.
please to a
cvs update -dPA
to ensure that old files/directories are deleted, otherwise you'll
probably encounter errors when trying to run zope3
if you have problems .. send me an email ueck <at> net-labs.de
Ulrich
=== Added File Zope3/src/zope/app/workflow/stateful/__init__.py ===
=== Added File Zope3/src/zope/app/workflow/stateful/configure.zcml ===
<zopeConfigure
xmlns="http://namespaces.zope.org/zope"
xmlns:workflow="http://namespaces.zope.org/workflow" >
<!-- Stateful ProcessDefintion -->
<content class="zope.app.workflow.stateful.definition.StatefulProcessDefinition">
<factory
id="StatefulProcessDefinition"
permission="zope.workflow.ManageProcessDefinitions"
/>
<require
permission="zope.workflow.ManageProcessDefinitions"
interface="zope.app.interfaces.workflow.stateful.IStatefulProcessDefinition"
set_schema="zope.app.interfaces.workflow.stateful.IStatefulProcessDefinition"
/>
<require
permission="zope.workflow.ManageProcessDefinitions"
interface="zope.app.interfaces.container.IReadContainer"
/>
<implements
interface="zope.app.interfaces.annotation.IAttributeAnnotatable"
/>
<implements
interface="zope.app.interfaces.services.configuration.IUseConfigurable"
/>
</content>
<!-- States Container -->
<content class="zope.app.workflow.stateful.definition.StatesContainer">
<factory
id="StatefulStatesContainer"
permission="zope.workflow.ManageProcessDefinitions"
/>
<require
permission="zope.workflow.ManageProcessDefinitions"
interface="zope.app.interfaces.workflow.stateful.IStatefulStatesContainer"
/>
<implements
interface="zope.app.interfaces.annotation.IAttributeAnnotatable"
/>
</content>
<!-- State -->
<content class="zope.app.workflow.stateful.definition.State">
<factory
id="StatefulState"
permission="zope.workflow.ManageProcessDefinitions"
/>
<require
permission="zope.workflow.ManageProcessDefinitions"
interface="zope.app.interfaces.workflow.stateful.IState"
set_schema="zope.app.interfaces.workflow.stateful.IState"
/>
<implements
interface="zope.app.interfaces.annotation.IAttributeAnnotatable"
/>
</content>
<!-- Transitions Container -->
<content class="zope.app.workflow.stateful.definition.TransitionsContainer">
<factory
id="StatefulTransitionsContainer"
permission="zope.workflow.ManageProcessDefinitions"
/>
<require
permission="zope.workflow.ManageProcessDefinitions"
interface="zope.app.interfaces.workflow.stateful.IStatefulTransitionsContainer"
/>
<implements
interface="zope.app.interfaces.annotation.IAttributeAnnotatable"
/>
</content>
<!-- Transition -->
<content class="zope.app.workflow.stateful.definition.Transition">
<factory
id="StatefulTransition"
permission="zope.workflow.ManageProcessDefinitions"
/>
<require
permission="zope.workflow.ManageProcessDefinitions"
interface="zope.app.interfaces.workflow.stateful.ITransition"
set_schema="zope.app.interfaces.workflow.stateful.ITransition"
/>
<implements
interface="zope.app.interfaces.annotation.IAttributeAnnotatable"
/>
</content>
<!-- ContentWorkflowsUtility -->
<content class="zope.app.workflow.stateful.contentworkflow.ContentWorkflowsUtility">
<require
permission="zope.ManageServices"
interface="zope.app.interfaces.workflow.stateful.IContentWorkflowsUtility"
/>
<factory
id="ContentWorkflowsUtility"
permission="zope.ManageServices"
/>
</content>
<!-- Stateful workflow import/Export -->
<workflow:importHandler
interface="zope.app.interfaces.workflow.stateful.IStatefulProcessDefinition"
factory=".xmlimportexport.XMLImportHandler"
/>
<workflow:exportHandler
interface="zope.app.interfaces.workflow.stateful.IStatefulProcessDefinition"
factory=".xmlimportexport.XMLExportHandler"
/>
</zopeConfigure>
=== Added File Zope3/src/zope/app/workflow/stateful/contentworkflow.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Content Workflows Utility
Associates content objects with some workflow process definitions.
$Id: contentworkflow.py,v 1.1 2003/05/08 17:27:19 jack-e Exp $
"""
__metaclass__ = type
from zope.interface import Interface
from persistence import Persistent
from zope.component import getService, queryAdapter
from zope.component.exceptions import ComponentLookupError
from zope.proxy.context import ContextMethod
from zope.proxy.introspection import removeAllProxies
from zope.app.interfaces.event import ISubscriber
from zope.app.interfaces.event import IObjectCreatedEvent
from zope.app.services.servicenames import EventSubscription, Workflows
from zope.app.interfaces.workflow import IProcessInstanceContainer
from zope.app.interfaces.workflow import IProcessInstanceContainerAdaptable
from zope.app.interfaces.workflow.stateful import IContentWorkflowsUtility
class ContentWorkflowsUtility(Persistent):
__implements__ = IContentWorkflowsUtility, ISubscriber
def __init__(self):
super(ContentWorkflowsUtility, self).__init__()
self._names = ('default',) # _names should be a TypeRegistry
# ISubscriber
def notify(self, event):
"""An event occured. Perhaps register this object with the hub."""
obj = event.object
# XXX Do i need to removeAllProxies somewhere in here ???
# check if it implements IProcessInstanceContainerAdaptable
if not IProcessInstanceContainerAdaptable.isImplementedBy(obj):
return
pi_container = queryAdapter(obj, IProcessInstanceContainer)
# probably need to adapt to IZopeContainer to use pi_container with
# context.
if pi_container is None:
# Object can't have associated PIs.
return
if IObjectCreatedEvent.isImplementedBy(event):
wfs = getService(self, Workflows)
# here we will lookup the configured processdefinitions
# for the newly created compoent. For every pd_name
# returned we will create a processinstance.
for pd_name in self._names:
if pd_name in pi_container.keys():
continue
try:
pi = wfs.createProcessInstance(pd_name)
print "CREATED PROCESSINSTANCE:", str(pi)
except KeyError:
# No registered PD with that name..
continue
pi_container.setObject(pd_name, pi)
notify = ContextMethod(notify)
# IContentWorkflowsUtility
# control
currentlySubscribed = False # Default subscription state
def subscribe(self):
if self.currentlySubscribed:
raise ValueError, "already subscribed; please unsubscribe first"
channel = self._getChannel(None)
channel.subscribe(self, IObjectCreatedEvent)
self.currentlySubscribed = True
subscribe = ContextMethod(subscribe)
def unsubscribe(self):
if not self.currentlySubscribed:
raise ValueError, "not subscribed; please subscribe first"
channel = self._getChannel(None)
channel.unsubscribe(self, IObjectCreatedEvent)
self.currentlySubscribed = False
unsubscribe = ContextMethod(unsubscribe)
def isSubscribed(self):
return self.currentlySubscribed
def _getChannel(self, channel):
if channel is None:
channel = getService(self, EventSubscription)
return channel
_getChannel = ContextMethod(_getChannel)
# config
def getProcessDefinitionNames(self):
"""Get the process definition names."""
return self._names
def setProcessDefinitionNames(self, names):
"""Set the process definition names."""
self._names = tuple(names)
=== Added File Zope3/src/zope/app/workflow/stateful/definition.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Stateful workflow process definition.
$Id: definition.py,v 1.1 2003/05/08 17:27:19 jack-e Exp $
"""
__metaclass__ = type
from persistence import Persistent
from persistence.dict import PersistentDict
from zope.proxy.context import ContextMethod, ContextWrapper
from zope.proxy.context import getWrapperData, getWrapperContainer
from zope.proxy.context import getWrapperContext
from zope.app.interfaces.container import IReadContainer
from zope.app.interfaces.workflow import IProcessDefinition
from zope.app.interfaces.workflow.stateful import IStatefulProcessDefinition
from zope.app.interfaces.workflow.stateful import IState, ITransition
from zope.app.interfaces.workflow.stateful import IStatefulStatesContainer
from zope.app.interfaces.workflow.stateful import IStatefulTransitionsContainer
from zope.app.workflow.definition import ProcessDefinition
from zope.app.workflow.definition import ProcessDefinitionElementContainer
from zope.app.workflow.stateful.instance import StatefulProcessInstance
class State(Persistent):
"""State."""
__implements__ = IState
class StatesContainer(ProcessDefinitionElementContainer):
"""Container that stores States.
"""
__implements__ = IStatefulStatesContainer
class Transition(Persistent):
"""Transition."""
__implements__ = ITransition
def __init__(self, source=None, destination=None, condition=None,
script=None, permission=None, triggerMode=None):
super(Transition, self).__init__()
self.__source = source
self.__destination = destination
self.__condition = condition or None
self.__script = script or None
self.__permission = permission or None
self.__triggerMode = triggerMode
def getSourceState(self):
return self.__source
def setSourceState(self, source):
self.__source = source
def getDestinationState(self):
return self.__destination
def setDestinationState(self, destination):
self.__destination = destination
def getCondition(self):
return self.__condition
def setCondition(self, condition):
self.__condition = condition or None
def getScript(self):
return self.__script
def setScript(self, script):
self.__script = script or None
def getPermission(self):
return self.__permission
def setPermission(self, permission):
self.__permission = permission or None
def getTriggerMode(self):
return self.__triggerMode
def setTriggerMode(self, mode):
self.__triggerMode = mode
# See ITransition
sourceState = property(getSourceState, setSourceState, None,
"Source State of Transition.")
destinationState = property(getDestinationState, setDestinationState, None,
"Destination State of Transition.")
condition = property(getCondition, setCondition, None,
"Condition for Transition.")
script = property(getScript, setScript, None,
"Script for Transition.")
permission = property(getPermission, setPermission, None,
"Permission for Transition.")
triggerMode = property(getTriggerMode, setTriggerMode, None,
"TriggerMode for Transition.")
def getProcessDefinition(self):
return getWrapperContainer(self).getProcessDefinition()
getProcessDefinition = ContextMethod(getProcessDefinition)
class TransitionsContainer(ProcessDefinitionElementContainer):
"""Container that stores Transitions.
"""
__implements__ = IStatefulTransitionsContainer
class StatefulProcessDefinition(ProcessDefinition):
"""Stateful workflow process definition."""
__implements__ = IStatefulProcessDefinition, IReadContainer
def __init__(self):
super(StatefulProcessDefinition, self).__init__()
self.__states = StatesContainer()
initial = State()
self.__states.setObject(self.getInitialStateName(), initial)
self.__transitions = TransitionsContainer()
self.__schema = None
_clear = clear = __init__
############################################################
# Implementation methods for interface
# zope.app.interfaces.workflow.stateful.IStatefulProcessDefinition
def getRelevantDataSchema(self):
return self.__schema
def setRelevantDataSchema(self, schema):
self.__schema = schema
relevantDataSchema = property(getRelevantDataSchema,
setRelevantDataSchema,
None,
"Schema for RelevantData.")
states = property(lambda self: self.__states)
transitions = property(lambda self: self.__transitions)
def addState(self, name, state):
if name in self.states:
raise KeyError, name
self.states.setObject(name, state)
def getState(self, name):
return self.states[name]
getState = ContextMethod(getState)
def removeState(self, name):
del self.states[name]
def getStateNames(self):
return self.states.keys()
# XXX This shouldn't be hardcoded
def getInitialStateName(self):
return 'INITIAL'
def addTransition(self, name, transition):
if name in self.transitions:
raise KeyError, name
self.transitions.setObject(name, transition)
def getTransition(self, name):
return self.transitions[name]
getTransition = ContextMethod(getTransition)
def removeTransition(self, name):
del self.transitions[name]
def getTransitionNames(self):
return self.transitions.keys()
# IProcessDefinition
def createProcessInstance(self, definition_name):
pi_obj = StatefulProcessInstance(definition_name)
ContextWrapper(pi_obj, self).initialize()
return pi_obj
createProcessInstance = ContextMethod(createProcessInstance)
#
############################################################
############################################################
# Implementation methods for interface
# zope.app.interfaces.container.IReadContainer
def __getitem__(self, key):
"See Interface.Common.Mapping.IReadMapping"
result = self.get(key)
if result is None:
raise KeyError(key)
return result
__getitem__ = ContextMethod(__getitem__)
def get(self, key, default=None):
"See Interface.Common.Mapping.IReadMapping"
if key == 'states':
return self.states
if key == 'transitions':
return self.transitions
return default
get = ContextMethod(get)
def __contains__(self, key):
"See Interface.Common.Mapping.IReadMapping"
return self.get(key) is not None
# Enumeration methods. We'll only expose Packages for now:
def __iter__(self):
return iter(self.keys())
def keys(self):
return ['states', 'transitions']
def values(self):
return map(self.get, self.keys())
values = ContextMethod(values)
def items(self):
return [(key, self.get(key)) for key in self.keys()]
items = ContextMethod(items)
def __len__(self):
return 2
#
############################################################
=== Added File Zope3/src/zope/app/workflow/stateful/instance.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Stateful Process Instance
$Id: instance.py,v 1.1 2003/05/08 17:27:19 jack-e Exp $
"""
__metaclass__ = type
from types import StringTypes
from persistence import Persistent
from zope.schema import getFields
from zope.interface import directlyProvides
from zope.exceptions import Unauthorized
from zope.component import getService
from zope.component import getServiceManager
from zope.proxy.introspection import removeAllProxies
from zope.proxy.context import ContextMethod, getWrapperContainer
from zope.proxy.context import ContextWrapper,ContextAware
from zope.security.management import getSecurityManager
from zope.security.checker import CheckerPublic
# XXX Needed for WfrData Permission checking
# commented out for now
#from zope.security.checker import CheckerPublic, selectChecker
#from zope.security.checker import Checker
#from zope.security.proxy import getChecker, Proxy
from zope.app.security.permission import checkPermission
from zope.tales.engine import Engine
from zope.app.interfaces.workflow.stateful import IStatefulProcessInstance
from zope.app.workflow.instance import ProcessInstance
class RelevantData(ContextAware):
pass
# XXX Example of how Changes to Workflow Relevant Data would send out Events
# ToDo:
# - Define Events:
# RelevantDataChangingWorkflowEvent
# RelevantDataChangedWorkflowEvent
#
# - all this is untested !!!!
#
#class RelevantData:
#
# def __setattr__(self, key, value):
# is_schema_field = bool(key in getFields(self.__implements__).keys())
# if is_schema_field:
# # Send an Event bevor RelevantData changes
# oldvalue = getattr(self, key, None)
# print "send RelevantDataChangingWorkflowEvent(key:%s, old:%s, new:%s) here" \
# % (key, oldvalue, value)
#
# super(RelevantData, self).__setattr__(key, value)
#
# if is_schema_field:
# # Send an Event after RelevantData has changed
# print "send RelevantDataChangedWorkflowEvent(key:%s, old:%s, new:%s) here" \
# % (key, oldvalue, value)
class StateChangeInfo:
"""Immutable StateChangeInfo.
"""
def __init__(self, transition):
self.__old_state = transition.sourceState
self.__new_state = transition.destinationState
old_state = property(lambda self: self.__old_state)
new_state = property(lambda self: self.__new_state)
class StatefulProcessInstance(ProcessInstance, Persistent):
"""Stateful Workflow ProcessInstance.
"""
__implements__ = IStatefulProcessInstance
############################################################
# Implementation methods for interface
# zope.app.interfaces.workflow.IStatefulProcessInstance
data = property(lambda self: ContextWrapper(self._data, self))
# XXX this is not entirely tested nor finished
#def _getData(self):
# """getter for Workflow Relevant Data."""
#
# data = self._data
# if data is None:
# return
#
# schema = data.__implements__
#
# # XXX permissions need to be manageable TTW
# # is this too much overhead ????
# checker_getattr = {}
# checker_setattr = {}
# for name in schema.names(all=True):
#
# # XXX Just a dummy implementation for now
# checker_getattr[name] = CheckerPublic
# checker_setattr[name] = CheckerPublic
#
# checker = Checker(checker_getattr.get, checker_setattr.get)
# return Proxy(data, checker)
#
#data = property(lambda self: ContextWrapper(self._getData(), self))
def initialize(self):
pd = self._getProcessDefinition()
clean_pd = removeAllProxies(pd)
self._status = clean_pd.getInitialStateName()
# resolve schema class
schema = clean_pd.getRelevantDataSchema()
if schema:
if type(schema) in StringTypes:
sm = getServiceManager(self)
schema = sm.resolve(schema)
# create relevant-data
self._data = self._buildRelevantData(schema)
else:
self._data = None
# setup permission on data
# check for Automatic Transitions
self._checkAndFireAuto(clean_pd)
initialize = ContextMethod(initialize)
def getOutgoingTransitions(self):
pd = self._getProcessDefinition()
clean_pd = removeAllProxies(pd)
return self._outgoingTransitions(clean_pd)
getOutgoingTransitions = ContextMethod(getOutgoingTransitions)
def fireTransition(self, id):
pd = self._getProcessDefinition()
clean_pd = removeAllProxies(pd)
if not id in self._outgoingTransitions(clean_pd):
raise KeyError, 'Invalid Transition Id: %s' % id
trans = clean_pd.transitions[id]
# modify relevant-data if needed
# XXX Implement EventHandling in BaseClass as property ???
# send StatusChangingWorkflowEvent
#print "send StatusChangingWorkflowEvent(old:%s, new:%s) here" \
# % (self._status, trans.destinationState)
# change status
self._status = trans.destinationState
# send StatusChangedWorkflowEvent
#print "send StatusChangedWorkflowEvent(old:%s, new:%s) here" \
# % (trans.sourceState, self._status)
# check for automatic transitions
self._checkAndFireAuto(clean_pd)
fireTransition = ContextMethod(fireTransition)
#
############################################################
# XXX expose this method in the interface (without _) ???
def _getProcessDefinition(self):
"""Get the ProcessDefinition object from WorkflowService.
"""
svc = getService(self, "Workflows")
return svc.getProcessDefinition(self.processDefinitionName)
_getProcessDefinition = ContextMethod(_getProcessDefinition)
# XXX this is not entirely tested
def _getContext(self):
ctx = {}
# data should be readonly for condition-evaluation
ctx['data'] = self.data
ctx['principal'] = getSecurityManager().getPrincipal()
# XXX This needs to be discussed:
# how can we know if this ProcessInstance is annotated
# to a Content-Object and provide secure ***READONLY***
# Access to it for evaluating Transition Conditions ???
#content = getWrapperContainer(self)
# XXX How can i make shure that nobody modifies content
# while the condition scripts/conditions are evaluated ????
# this hack only prevents from directly setting an attribute
# using a setter-method directly is not protected :((
#try:
# checker = getChecker(content)
# checker._setattr_permission_func = lambda x: None
#except TypeError:
# # got object without Security Proxy
# checker = selectChecker(content)
# checker._setattr_permission_func = lambda x: None
# content = Proxy(content, checker)
#ctx['content'] = content
return ctx
_getContext = ContextMethod(_getContext)
def _extendContext(self, transition, ctx={}):
ctx['state_change'] = StateChangeInfo(transition)
return ctx
def _evaluateCondition(self, transition, contexts):
"""Evaluate a condition in context of relevant-data.
"""
if not transition.condition:
return True
expr = Engine.compile(transition.condition)
return expr(Engine.getContext( contexts=contexts ))
def _evaluateScript(self, transition, contexts):
script = transition.script
if not script:
return True
if type(script) in StringTypes:
sm = getServiceManager(self)
script = sm.resolve(script)
return script(contexts)
_evaluateScript = ContextMethod(_evaluateScript)
def _buildRelevantData(self, schema):
"""Create a new data object and initialize with Schema defaults.
"""
data = RelevantData()
if schema is not None:
# set schema to RelevantData Instance
directlyProvides(data, schema)
for name, field in getFields(schema).items():
setattr(data, name, field.default)
return data
def _outgoingTransitions(self, clean_pd):
sm = getSecurityManager()
ret = []
contexts = self._getContext()
for name, trans in clean_pd.transitions.items():
if self.status == trans.sourceState:
# check permissions
permission = trans.permission
#
if (permission is not None
and permission is not CheckerPublic
and not sm.checkPermission(permission, self)
):
continue
ctx = self._extendContext(trans, contexts)
# evaluate conditions
if trans.condition is not None:
try:
include = self._evaluateCondition(trans, ctx)
except Unauthorized:
include = 0
if not include:
continue
if trans.script is not None:
try:
include = self._evaluateScript(trans, ctx)
except Unauthorized:
include = 0
if not include:
continue
# append transition name
ret.append(name)
return ret
_outgoingTransitions = ContextMethod(_outgoingTransitions)
def _checkAndFireAuto(self, clean_pd):
outgoing_transitions = self.getOutgoingTransitions()
for name in outgoing_transitions:
trans = clean_pd.transitions[name]
# XXX Use Constants instead of strings
if trans.triggerMode == 'Automatic':
self.fireTransition(name)
return
_checkAndFireAuto = ContextMethod(_checkAndFireAuto)
=== Added File Zope3/src/zope/app/workflow/stateful/xmlexport_template.pt ===
<?xml version="1.0"?>
<workflow type="StatefulWorkflow"
xmlns:tal="http://xml.zope.org/namespaces/tal"
tal:define="wf view/getDefinition"
tal:attributes="title python:view.getDublinCore(wf).Title()">
<schema
name = ""
tal:attributes="name wf/getRelevantDataSchema">
</schema>
<states>
<state
title = ""
name = ""
tal:repeat="state wf/getStateNames"
tal:attributes="title python:view.getDublinCore(wf.getState(state)).Title();
name state"></state>
</states>
<transitions>
<tal:block tal:repeat="trans wf/getTransitionNames">
<transition
sourceState = ""
destinationState = ""
condition = ""
script = ""
permission = ""
triggerMode = ""
title = ""
name = ""
tal:define="transObj python:wf.getTransition(trans)"
tal:attributes="sourceState transObj/getSourceState;
destinationState transObj/getDestinationState;
condition transObj/getCondition;
script transObj/getScript;
permission python:view.getPermissionId(transObj.getPermission());
triggerMode transObj/getTriggerMode;
title python:view.getDublinCore(transObj).Title();
name trans"></transition>
</tal:block>
</transitions>
</workflow>
=== Added File Zope3/src/zope/app/workflow/stateful/xmlimportexport.py ===
##############################################################################
#
# Copyright (c) 2002 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
"""Stateful ProcessDefinition XML Import/Export handlers
$Id: xmlimportexport.py,v 1.1 2003/05/08 17:27:19 jack-e Exp $
"""
__metaclass__ = type
from zope.app.pagetemplate.viewpagetemplatefile \
import ViewPageTemplateFile
from zope.app.interfaces.workflow.stateful \
import IStatefulProcessDefinition
from zope.app.interfaces.workflow import IProcessDefinitionImportHandler
from zope.app.interfaces.workflow import IProcessDefinitionExportHandler
from zope.component import getAdapter, getServiceManager
from zope.app.interfaces.dublincore import IZopeDublinCore
from types import StringTypes
from zope.proxy.context import ContextMethod
from zope.proxy.introspection import removeAllProxies
from zope.security.checker import CheckerPublic
from xml.sax import parse
from xml.sax.handler import ContentHandler
from zope.app.workflow.stateful.definition import State, Transition
# basic implementation for a format-checker
class XMLFormatChecker(ContentHandler):
def __init__(self):
self.__valid = False
def startElement(self, name, attrs):
if name == 'workflow' and attrs.get('type',None) == 'StatefulWorkflow':
self.__valid = True
def endElement(self, name):
pass
def isValid(self):
return self.__valid
class XMLStatefulImporter(ContentHandler):
def __init__(self, context, encoding='latin-1'):
self.context = context
self.encoding = encoding
def startElement(self, name, attrs):
handler = getattr(self, 'start' + name.title().replace('-', ''), None)
if not handler:
raise ValueError, 'Unknown element %s' % name
handler(attrs)
def endElement(self, name):
handler = getattr(self, 'end' + name.title().replace('-', ''), None)
if handler:
handler()
def noop(*args):
pass
startStates = noop
startTransitions = noop
def startWorkflow(self, attrs):
dc = getAdapter(self.context, IZopeDublinCore)
dc.title = attrs.get('title', u'')
def startSchema(self, attrs):
name = attrs['name'].encode(self.encoding)
self.context.setRelevantDataSchema(name)
def startState(self, attrs):
encoding = self.encoding
name = attrs['name'].encode(encoding)
if name == 'INITIAL':
state = self.context.getState('INITIAL')
dc = getAdapter(state, IZopeDublinCore)
dc.title = attrs.get('title', u'')
else:
state = State()
dc = getAdapter(state, IZopeDublinCore)
dc.title = attrs.get('title', u'')
self.context.addState(name, state)
def startTransition(self, attrs):
encoding = self.encoding
name = attrs['name'].encode(encoding)
permission = attrs.get('permission', '').encode(encoding)
if permission == 'zope.Public':
permission = CheckerPublic
trans = Transition(source = attrs['sourceState'].encode(encoding),
destination = attrs['destinationState'].encode(encoding),
condition = attrs.get('condition', '').encode(encoding),
script = attrs.get('script', '').encode(encoding),
permission = permission,
triggerMode = attrs['triggerMode'].encode(encoding))
dc = getAdapter(trans, IZopeDublinCore)
dc.title = attrs.get('title', u'')
self.context.addTransition(name, trans)
class XMLImportHandler:
__implements__ = IProcessDefinitionImportHandler
# XXX Implementation needs more work !!
# check if xml-data can be imported and represents a StatefulPD
def canImport(self, context, data):
checker = XMLFormatChecker()
parse(data, checker)
return bool(IStatefulProcessDefinition.isImplementedBy(context)) \
and checker.isValid()
def doImport(self, context, data):
# XXX Manually clean ProcessDefinition ??
context.clear()
parse(data, XMLStatefulImporter(context))
class XMLExportHandler:
__implements__ = IProcessDefinitionExportHandler
template = ViewPageTemplateFile('xmlexport_template.pt')
def doExport(self, context, process_definition):
# XXX Not really nice to fake a BrowserView here ....
self.request = None
self.process_definition = process_definition
self.context = context
return self.template()
def getDefinition(self):
return self.process_definition
def getDublinCore(self, obj):
return getAdapter(obj, IZopeDublinCore)
def getPermissionId(self, permission):
if isinstance(permission, str):
return permission
if permission is CheckerPublic:
return 'zope.Public'
if permission is None:
return ''
return removeAllProxies(permission).getId()