src/shoobx/wfmc/xpdl.txt
===========
XPDL Import
===========
We can import process definitions from files in the XML Process
Definition Language (XPDL) format. An XPDL file contains multiple
process definitions arranged in a package. When we load the file, we
get a package containing some number of process definitions.
Let's look at an example. The file `publication.xpdl`
contains a definition for the publication example developed in the
"README.txt" file. We can read it using the xpdl module:
>>> from shoobx.wfmc import xpdl
>>> import os
>>> package = xpdl.read(open(os.path.join(this_directory,
... 'publication-1.0.xpdl')))
This package contains a single definition:
>>> package
{'Publication': ProcessDefinition('Publication')}
>>> pd = package['Publication']
>>> from shoobx.wfmc.attributeintegration import AttributeIntegration
>>> integration = AttributeIntegration()
>>> pd.integration = integration
Now, having read the process definition, we can use it as we did
before (in "README.txt"). As before, we'll create an event subscriber
so that we can see what's going on:
>>> def log_workflow(event):
... print(event)
>>> import zope.event
>>> zope.event.subscribers.append(log_workflow)
and we'll register the process definition as a utility:
>>> import zope.component
>>> from shoobx.wfmc.process import StaticProcessDefinitionFactory
>>> pdfactory = StaticProcessDefinitionFactory()
>>> zope.component.provideUtility(pdfactory)
>>> pdfactory.register(pd)
and we'll define and register participant and application adapters:
>>> import zope.interface
>>> from shoobx.wfmc import interfaces
>>> @zope.interface.implementer(interfaces.IParticipant)
... class Participant(object):
... zope.component.adapts(interfaces.IActivity)
...
... def __init__(self, activity, process):
... self.activity = activity
>>> class User:
... def __init__(self):
... self.work_list = []
>>> authors = {'bob': User(), 'ted': User(), 'sally': User()}
>>> reviewer = User()
>>> tech1 = User()
>>> tech2 = User()
>>> class Author(Participant):
... def __init__(self, activity, process):
... Participant.__init__(self, activity, process)
... author_name = activity.process.workflowRelevantData.author
... print("Author `%s` selected" % author_name)
... self.user = authors[author_name]
>>> integration.authorParticipant = Author
>>> class Reviewer(Participant):
... user = reviewer
>>> integration.reviewerParticipant = Reviewer
>>> class Tech1(Participant):
... user = tech1
>>> integration.tech1Participant = Tech1
>>> class Tech2(Participant):
... user = tech2
>>> integration.tech2Participant = Tech2
>>> integration.SystemParticipant = Participant
>>> @zope.interface.implementer(interfaces.IWorkItem)
... class ApplicationBase(object):
... zope.component.adapts(interfaces.IParticipant)
...
... def __init__(self, participant, process, activity):
... self.participant = participant
... self.activity = participant.activity
... participant.user.work_list.append(self)
...
... def start(self, args):
... pass
...
... def finish(self):
... self.participant.activity.workItemFinished(self)
>>> class Prepare(ApplicationBase):
...
... def summary(self):
... process = self.activity.process
... doc = getattr(process.applicationRelevantData, 'doc', '')
... if doc:
... print('Previous draft:')
... print(doc)
... print('Changes we need to make:')
... for change in process.workflowRelevantData.tech_changes:
... print(change)
... else:
... print('Please write the initial draft')
...
... def finish(self, doc):
... self.activity.process.applicationRelevantData.doc = doc
... super(Prepare, self).finish()
>>> integration.prepareWorkItem = Prepare
>>> class TechReview(ApplicationBase):
...
... def getDoc(self):
... return self.activity.process.applicationRelevantData.doc
...
... def finish(self, decision, changes):
... output = {'publish': decision, 'tech_changes': changes}
... self.activity.workItemFinished(self, output)
>>> integration.tech_reviewWorkItem = TechReview
>>> class Review(TechReview):
...
... def start(self, args):
... publish1 = args['publish1']
... publish2 = args['publish2']
... changes1 = args['tech_changes1']
... changes2 = args['tech_changes2']
... if not (publish1 and publish2):
... output = {'publish': False,
... 'tech_changes': changes1 + changes2,
... 'ed_changes': ()}
... # Reject if either tech reviewer rejects
... self.activity.workItemFinished(
... self, output)
...
... if changes1 or changes2:
... output = {'publish': True,
... 'tech_changes': changes1 + changes2,
... 'ed_changes': ()}
... # we won't do anyting if there are tech changes
... self.activity.workItemFinished(
... self, output)
...
... def finish(self, ed_changes):
... output = {'publish': True, 'tech_changes': (), 'ed_changes': ed_changes}
... self.activity.workItemFinished(self, output)
>>> integration.ed_reviewWorkItem = Review
>>> class Final(ApplicationBase):
...
... def summary(self):
... process = self.activity.process
... doc = getattr(process.applicationRelevantData, 'doc', '')
... print('Previous draft:')
... print(self.activity.process.applicationRelevantData.doc)
... print('Changes we need to make:')
... for change in process.workflowRelevantData.ed_changes:
... print(change)
...
... def finish(self, doc):
... self.activity.process.applicationRelevantData.doc = doc
... super(Final, self).finish()
>>> integration.finalWorkItem = Final
>>> class ReviewFinal(TechReview):
...
... def finish(self, ed_changes):
... output = {'publish': True, 'tech_changes': (), 'ed_changes': ed_changes}
... self.activity.workItemFinished(self, output)
>>> integration.rfinalWorkItem = ReviewFinal
>>> @zope.interface.implementer(interfaces.IWorkItem)
... class Publish:
... zope.component.adapts(interfaces.IParticipant)
...
... def __init__(self, participant, process, activity):
... self.participant = participant
...
... def start(self, args):
... print("Published")
... self.finish()
...
... def finish(self):
... self.participant.activity.workItemFinished(self)
>>> integration.publishWorkItem = Publish
>>> class Reject(Publish):
... def start(self, args):
... print("Rejected")
... self.finish()
>>> integration.rejectWorkItem = Reject
and a process context, so we can pass parameters:
>>> @zope.interface.implementer(interfaces.IProcessContext)
... class PublicationContext:
...
... def processFinished(self, process, decision):
... self.decision = decision
Now, let's try out our process. We'll follow the same steps we did in
"README.txt", getting the same results:
>>> context = PublicationContext()
>>> proc = pd(context)
>>> proc.start('bob')
ProcessStarted(Process('Publication'))
Transition(None, Activity('Publication.start'))
ActivityStarted(Activity('Publication.start'))
ActivityFinished(Activity('Publication.start'))
Author `bob` selected
Transition(Activity('Publication.start'),
Activity('Publication.prepare'))
ActivityStarted(Activity('Publication.prepare'))
WorkItemStarting('prepare')
WorkItemStarted('prepare')
>>> item = authors['bob'].work_list.pop()
>>> item.finish("I give my pledge, as an American\n"
... "to save, and faithfully to defend from waste\n"
... "the natural resources of my Country.")
WorkItemFinished('prepare')
ActivityFinished(Activity('Publication.prepare'))
Transition(Activity('Publication.prepare'),
Activity('Publication.tech1'))
ActivityStarted(Activity('Publication.tech1'))
WorkItemStarting('tech_review')
WorkItemStarted('tech_review')
Transition(Activity('Publication.prepare'),
Activity('Publication.tech2'))
ActivityStarted(Activity('Publication.tech2'))
WorkItemStarting('tech_review')
WorkItemStarted('tech_review')
>>> item = tech1.work_list.pop()
>>> print(item.getDoc())
I give my pledge, as an American
to save, and faithfully to defend from waste
the natural resources of my Country.
>>> item.finish(True, ['Change "American" to "human"'])
WorkItemFinished('tech_review')
ActivityFinished(Activity('Publication.tech1'))
Transition(Activity('Publication.tech1'),
Activity('Publication.review'))
>>> item = tech2.work_list.pop()
>>> item.finish(True, ['Change "Country" to "planet"'])
WorkItemFinished('tech_review')
ActivityFinished(Activity('Publication.tech2'))
Transition(Activity('Publication.tech2'),
Activity('Publication.review'))
ActivityStarted(Activity('Publication.review'))
WorkItemStarting('ed_review')
WorkItemFinished('ed_review')
ActivityFinished(Activity('Publication.review'))
Author `bob` selected
Transition(Activity('Publication.review'),
Activity('Publication.prepare'))
ActivityStarted(Activity('Publication.prepare'))
WorkItemStarting('prepare')
WorkItemStarted('prepare')
WorkItemStarted('ed_review')
>>> item = authors['bob'].work_list.pop()
>>> item.summary()
Previous draft:
I give my pledge, as an American
to save, and faithfully to defend from waste
the natural resources of my Country.
Changes we need to make:
Change "American" to "human"
Change "Country" to "planet"
>>> item.finish("I give my pledge, as an human\n"
... "to save, and faithfully to defend from waste\n"
... "the natural resources of my planet.")
WorkItemFinished('prepare')
ActivityFinished(Activity('Publication.prepare'))
Transition(Activity('Publication.prepare'),
Activity('Publication.tech1'))
ActivityStarted(Activity('Publication.tech1'))
WorkItemStarting('tech_review')
WorkItemStarted('tech_review')
Transition(Activity('Publication.prepare'),
Activity('Publication.tech2'))
ActivityStarted(Activity('Publication.tech2'))
WorkItemStarting('tech_review')
WorkItemStarted('tech_review')
>>> item = tech1.work_list.pop()
>>> item.finish(True, [])
WorkItemFinished('tech_review')
ActivityFinished(Activity('Publication.tech1'))
Transition(Activity('Publication.tech1'),
Activity('Publication.review'))
>>> item = tech2.work_list.pop()
>>> item.finish(True, [])
WorkItemFinished('tech_review')
ActivityFinished(Activity('Publication.tech2'))
Transition(Activity('Publication.tech2'),
Activity('Publication.review'))
ActivityStarted(Activity('Publication.review'))
WorkItemStarting('ed_review')
WorkItemStarted('ed_review')
>>> item = reviewer.work_list.pop()
>>> print(item.getDoc())
I give my pledge, as an human
to save, and faithfully to defend from waste
the natural resources of my planet.
>>> item.finish(['change "an" to "a"'])
WorkItemFinished('ed_review')
ActivityFinished(Activity('Publication.review'))
Author `bob` selected
Transition(Activity('Publication.review'),
Activity('Publication.final'))
ActivityStarted(Activity('Publication.final'))
WorkItemStarting('final')
WorkItemStarted('final')
>>> item = authors['bob'].work_list.pop()
>>> item.summary()
Previous draft:
I give my pledge, as an human
to save, and faithfully to defend from waste
the natural resources of my planet.
Changes we need to make:
change "an" to "a"
>>> item.finish("I give my pledge, as a human\n"
... "to save, and faithfully to defend from waste\n"
... "the natural resources of my planet.")
WorkItemFinished('final')
ActivityFinished(Activity('Publication.final'))
Transition(Activity('Publication.final'),
Activity('Publication.rfinal'))
ActivityStarted(Activity('Publication.rfinal'))
WorkItemStarting('rfinal')
WorkItemStarted('rfinal')
>>> item = reviewer.work_list.pop()
>>> print(item.getDoc())
I give my pledge, as a human
to save, and faithfully to defend from waste
the natural resources of my planet.
>>> item.finish([])
WorkItemFinished('rfinal')
ActivityFinished(Activity('Publication.rfinal'))
Transition(Activity('Publication.rfinal'),
Activity('Publication.publish'))
ActivityStarted(Activity('Publication.publish'))
WorkItemStarting('publish')
Published
WorkItemFinished('publish')
ActivityFinished(Activity('Publication.publish'))
ProcessFinished(Process('Publication'))
WorkItemStarted('publish')
>>> proc.workflowRelevantData.publish
True
Descriptions
------------
Most process elements can have names and descriptions.
>>> pd.__name__
'Publication'
>>> pd.description
'This is the sample process'
>>> pd.applications['prepare'].__name__
'Prepare'
>>> pd.applications['prepare'].description
'Prepare the initial draft'
>>> pd.activities['tech1'].__name__
'Technical Review 1'
>>> pd.activities['tech1'].description
'This is the first Technical Review.'
>>> pd.participants['tech1'].__name__
'Technical Reviewer 1'
>>> pd.participants['tech1'].description
'He is a smart guy.'
>>> sorted([item.__name__ for item in pd.transitions])
['Transition', 'Transition', 'Transition', 'Transition',
'Transition', 'Transition', 'Transition', 'Transition',
'Transition', 'Transition', 'Transition to Tech Review 1',
'Transition to Tech Review 2']
>>> descriptions = [item.description for item in pd.transitions if item.description]
>>> 'Use this transition if there are editorial changes required.' in descriptions
True