--- /dev/null Thu Jan 01 00:00:00 1970 +0000
+++ b/src/pyams_workflow/interfaces.py Thu Jan 10 17:27:02 2019 +0100
@@ -0,0 +1,502 @@
+#
+# Copyright (c) 2008-2015 Thierry Florac <tflorac AT ulthar.net>
+# All Rights Reserved.
+#
+# This software is subject to the provisions of the Zope Public License,
+# Version 2.1 (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.
+#
+
+__docformat__ = 'restructuredtext'
+
+
+# import standard library
+
+# import interfaces
+from pyams_utils.interfaces import VIEW_PERMISSION
+from zope.annotation.interfaces import IAttributeAnnotatable
+from zope.interface.interfaces import IObjectEvent, ObjectEvent
+from zope.lifecycleevent.interfaces import IObjectCreatedEvent
+
+# import packages
+from pyams_security.schema import Principal
+from zope.interface import implementer, invariant, Interface, Attribute, Invalid
+from zope.lifecycleevent import ObjectCreatedEvent
+from zope.schema import Choice, Datetime, Set, TextLine, Text, List, Object, Int, Bool
+from zope.schema.vocabulary import SimpleVocabulary, SimpleTerm
+
+from pyams_workflow import _
+
+
+MANUAL = 0
+AUTOMATIC = 1
+SYSTEM = 2
+
+
+class InvalidTransitionError(Exception):
+ """Base transition error"""
+
+ def __init__(self, source):
+ self.source = source
+
+ def __str__(self):
+ return 'source: "%s"' % self.source
+
+
+class NoTransitionAvailableError(InvalidTransitionError):
+ """Exception raised when there is not available transition"""
+
+ def __init__(self, source, destination):
+ super(NoTransitionAvailableError, self).__init__(source)
+ self.destination = destination
+
+ def __str__(self):
+ return 'source: "%s" destination: "%s"' % (self.source, self.destination)
+
+
+class AmbiguousTransitionError(InvalidTransitionError):
+ """Exception raised when required transition is ambiguous"""
+
+ def __init__(self, source, destination):
+ super(AmbiguousTransitionError, self).__init__(source)
+ self.destination = destination
+
+ def __str__(self):
+ return 'source: "%s" destination: "%s"' % (self.source, self.destination)
+
+
+class VersionError(Exception):
+ """Versions management error"""
+
+
+class ConditionFailedError(Exception):
+ """Exception raised when transition condition failed"""
+
+
+class IWorkflowTransitionEvent(IObjectEvent):
+ """Workflow transition event interface"""
+
+ wokflow = Attribute("Workflow utility")
+
+ principal = Attribute("Event principal")
+
+ source = Attribute('Original state or None if initial state')
+
+ destination = Attribute('New state')
+
+ transition = Attribute('Transition that was fired or None if initial state')
+
+ comment = Attribute('Comment that went with state transition')
+
+
+@implementer(IWorkflowTransitionEvent)
+class WorkflowTransitionEvent(ObjectEvent):
+ """Workflow transition event"""
+
+ def __init__(self, object, workflow, principal, source, destination, transition, comment):
+ super(WorkflowTransitionEvent, self).__init__(object)
+ self.workflow = workflow
+ self.principal = principal
+ self.source = source
+ self.destination = destination
+ self.transition = transition
+ self.comment = comment
+
+
+class IWorkflowVersionTransitionEvent(IWorkflowTransitionEvent):
+ """Workflow version transition event interface"""
+
+ old_object = Attribute('Old version of object')
+
+
+@implementer(IWorkflowVersionTransitionEvent)
+class WorkflowVersionTransitionEvent(WorkflowTransitionEvent):
+ """Workflow version transition event"""
+
+ def __init__(self, object, workflow, principal, old_object, source, destination, transition, comment):
+ super(WorkflowVersionTransitionEvent, self).__init__(object, workflow, principal, source,
+ destination, transition, comment)
+ self.old_object = old_object
+
+
+class IObjectClonedEvent(IObjectCreatedEvent):
+ """Object cloned event interface"""
+
+ source = Attribute("Cloned object source")
+
+
+@implementer(IObjectClonedEvent)
+class ObjectClonedEvent(ObjectCreatedEvent):
+ """Object cloned event"""
+
+ def __init__(self, object, source):
+ super(ObjectClonedEvent, self).__init__(object)
+ self.source = source
+
+
+class IWorkflow(Interface):
+ """Defines workflow in the form of transition objects.
+
+ Defined as a utility.
+ """
+
+ initial_state = Attribute("Initial state")
+
+ update_states = Set(title="Updatable states",
+ description="States of contents which are updatable by standard contributors")
+
+ readonly_states = Set(title="Read-only states",
+ description="States of contents which can't be modified by anybody")
+
+ protected_states = Set(title="Protected states",
+ description="States of contents which can only be modified by site administrators")
+
+ manager_states = Set(title="Manager states",
+ description="States of contents which can be modified by site administrators and content "
+ "managers")
+
+ published_states = Set(title="Published states",
+ description="States of contents which are published")
+
+ visible_states = Set(title="Visible staets",
+ description="States of contents which are visible in front-office")
+
+ waiting_states = Set(title="Waiting states",
+ description="States of contents waiting for action")
+
+ retired_states = Set(title="Retired states",
+ description="States of contents which are retired but not yet archived")
+
+ auto_retired_state = Attribute("Auto-retired state")
+
+ def initialize(self):
+ """Do any needed initialization.
+
+ Such as initialization with the workflow versions system.
+ """
+
+ def refresh(self, transitions):
+ """Refresh workflow completely with new transitions."""
+
+ def get_state_label(self, state):
+ """Get given state label"""
+
+ def get_transitions(self, source):
+ """Get all transitions from source"""
+
+ def get_transition(self, source, transition_id):
+ """Get transition with transition_id given source state.
+
+ If the transition is invalid from this source state,
+ an InvalidTransitionError is raised.
+ """
+
+ def get_transition_by_id(self, transition_id):
+ """Get transition with transition_id"""
+
+
+class IWorkflowInfo(Interface):
+ """Get workflow info about workflowed object, and drive workflow.
+
+ Defined as an adapter.
+ """
+
+ def set_initial_state(self, state, comment=None):
+ """Set initial state for the context object.
+
+ Fires a transition event.
+ """
+
+ def fire_transition(self, transition_id, comment=None, side_effect=None, check_security=True, principal=None):
+ """Fire a transition for the context object.
+
+ There's an optional comment parameter that contains some
+ opaque object that offers a comment about the transition.
+ This is useful for manual transitions where users can motivate
+ their actions.
+
+ There's also an optional side effect parameter which should
+ be a callable which receives the object undergoing the transition
+ as the parameter. This could do an editing action of the newly
+ transitioned workflow object before an actual transition event is
+ fired.
+
+ If check_security is set to False, security is not checked
+ and an application can fire a transition no matter what the
+ user's permission is.
+ """
+
+ def fire_transition_toward(self, state, comment=None, side_effect=None, check_security=True, principal=None):
+ """Fire transition toward state.
+
+ Looks up a manual transition that will get to the indicated
+ state.
+
+ If no such transition is possible, NoTransitionAvailableError will
+ be raised.
+
+ If more than one manual transitions are possible,
+ AmbiguousTransitionError will be raised.
+ """
+
+ def fire_transition_for_versions(self, state, transition_id, comment=None, principal=None):
+ """Fire a transition for all versions in a state"""
+
+ def fire_automatic(self):
+ """Fire automatic transitions if possible by condition"""
+
+ def has_version(self, state):
+ """Return true if a version exists in given state"""
+
+ def get_manual_transition_ids(self):
+ """Returns list of valid manual transitions.
+
+ These transitions have to have a condition that's True.
+ """
+
+ def get_manual_transition_ids_toward(self, state):
+ """Returns list of manual transitions towards state"""
+
+ def get_automatic_transition_ids(self):
+ """Returns list of possible automatic transitions.
+
+ Condition is not checked.
+ """
+
+ def has_automatic_transitions(self):
+ """Return true if there are possible automatic outgoing transitions.
+
+ Condition is not checked.
+ """
+
+
+class IWorkflowStateHistoryItem(Interface):
+ """Workflow state history item"""
+
+ date = Datetime(title="State change datetime",
+ required=True)
+
+ source_version = Int(title="Source version ID",
+ required=False)
+
+ source_state = TextLine(title="Transition source state",
+ required=False)
+
+ target_state = TextLine(title="Transition target state",
+ required=True)
+
+ transition_id = TextLine(title="Transition ID",
+ required=True)
+
+ transition = TextLine(title="Transition name",
+ required=True)
+
+ principal = Principal(title="Transition principal",
+ required=False)
+
+ comment = Text(title="Transition comment",
+ required=False)
+
+
+class IWorkflowState(Interface):
+ """Store state on workflowed objects.
+
+ Defined as an adapter.
+ """
+
+ version_id = Int(title=_("Version ID"))
+
+ state = TextLine(title=_("Version state"))
+
+ state_date = Datetime(title=_("State date"),
+ description=_("Date at which the current state was applied"))
+
+ state_principal = Principal(title=_("State principal"),
+ description=_("ID of the principal which defined current state"))
+
+ state_urgency = Bool(title=_("Urgent request?"),
+ required=True,
+ default=False)
+
+ history = List(title="Workflow states history",
+ value_type=Object(schema=IWorkflowStateHistoryItem))
+
+ def get_first_state_date(self, states):
+ """Get first date at which given state was set"""
+
+
+class IWorkflowVersions(Interface):
+ """Interface to get information about versions of content in workflow"""
+
+ last_version_id = Attribute("Last version ID")
+
+ def get_version(self, version_id):
+ """Get version matching given id"""
+
+ def get_versions(self, states=None, sort=False, reverse=False):
+ """Get all versions of object known for this (optional) state"""
+
+ def get_last_versions(self, count=1):
+ """Get last versions of this object. Set count=0 to get all versions."""
+
+ def add_version(self, content, state, principal=None):
+ """Return new unique version id"""
+
+ def set_state(self, version_id, state, principal=None):
+ """Set new state for given version"""
+
+ def has_version(self, state):
+ """Return true if a version exists with the specific workflow state"""
+
+ def remove_version(self, version_id, state='deleted', comment=None, principal=None):
+ """Remove version with given ID"""
+
+
+class IWorkflowStateLabel(Interface):
+ """Workflow state label adapter interface"""
+
+ def get_label(self, content, request=None, format=True):
+ """Get state label for given content"""
+
+
+class IWorkflowManagedContent(IAttributeAnnotatable):
+ """Workflow managed content"""
+
+ content_class = Attribute("Content class")
+
+ workflow_name = Choice(title=_("Workflow name"),
+ description=_("Name of workflow utility managing this content"),
+ required=True,
+ vocabulary='PyAMS workflows')
+
+ view_permission = Choice(title=_("View permission"),
+ description=_("This permission will be required to display content"),
+ vocabulary='PyAMS permissions',
+ default=VIEW_PERMISSION,
+ required=False)
+
+
+class IWorkflowPublicationSupport(IAttributeAnnotatable):
+ """Workflow publication support"""
+
+
+class IWorkflowVersion(IWorkflowPublicationSupport):
+ """Workflow content version marker interface"""
+
+
+class IWorkflowTransitionInfo(Interface):
+ """Workflow transition info"""
+
+ transition_id = TextLine(title=_("Transition ID"),
+ required=True)
+
+
+DISPLAY_FIRST_VERSION = 'first'
+DISPLAY_CURRENT_VERSION = 'current'
+
+VERSION_DISPLAY = {DISPLAY_FIRST_VERSION: _("Display first version date"),
+ DISPLAY_CURRENT_VERSION: _("Display current version date")}
+
+VERSION_DISPLAY_VOCABULARY = SimpleVocabulary([SimpleTerm(v, title=t)
+ for v, t in VERSION_DISPLAY.items()])
+
+
+class IWorkflowPublicationInfo(Interface):
+ """Workflow content publication info"""
+
+ publication_date = Datetime(title=_("Publication date"),
+ description=_("Last date at which content was accepted for publication"),
+ required=False)
+
+ publisher = Principal(title=_("Publisher"),
+ description=_("Name of the manager who published the document"),
+ required=False)
+
+ publication = TextLine(title=_("Publication"),
+ description=_("Last publication date and actor"),
+ required=False,
+ readonly=True)
+
+ first_publication_date = Datetime(title=_("First publication date"),
+ description=_("First date at which content was accepted for publication"),
+ required=False)
+
+ publication_effective_date = Datetime(title=_("Publication start date"),
+ description=_("Date from which content will be visible"),
+ required=False)
+
+ push_end_date = Datetime(title=_("Push end date"),
+ description=_("Some contents can be pushed by components to front-office pages; if you "
+ "set a date here, this content will not be pushed anymore passed this "
+ "date, but will still be available via search engine or direct links"),
+ required=False)
+
+ push_end_date_index = Attribute("Push end date value used by catalog indexes")
+
+ @invariant
+ def check_push_end_date(self):
+ if self.push_end_date is not None:
+ if self.publication_effective_date is None:
+ raise Invalid(_("Can't define push end date without publication start date!"))
+ if self.publication_effective_date >= self.push_end_date:
+ raise Invalid(_("Push end date must be defined after publication start date!"))
+ if self.publication_expiration_date is not None:
+ if self.publication_expiration_date < self.push_end_date:
+ raise Invalid(_("Push end date must be null or defined before publication end date!"))
+
+ publication_expiration_date = Datetime(title=_("Publication end date"),
+ description=_("Date past which content will not be visible"),
+ required=False)
+
+ @invariant
+ def check_expiration_date(self):
+ if self.publication_expiration_date is not None:
+ if self.publication_effective_date is None:
+ raise Invalid(_("Can't define publication end date without publication start date!"))
+ if self.publication_effective_date >= self.publication_expiration_date:
+ raise Invalid(_("Publication end date must be defined after publication start date!"))
+
+ displayed_publication_date = Choice(title=_("Displayed publication date"),
+ description=_("The matching date will be displayed in front-office"),
+ vocabulary='PyAMS content publication date',
+ default=DISPLAY_FIRST_VERSION,
+ required=True)
+
+ visible_publication_date = Attribute("Visible publication date")
+
+ def reset(self, complete=True):
+ """Reset all publication info (used by clone features)
+
+ If 'complete' argument is True, all date fields are reset; otherwise, push and publication end dates are
+ preserved in new versions.
+ """
+
+ def is_published(self, check_parent=True):
+ """Is the content published?"""
+
+ def is_visible(self, request=None, check_parent=True):
+ """Is the content visible?"""
+
+
+class IWorkflowRequestUrgencyInfo(Interface):
+ """Workflow request urgency info"""
+
+ urgent_request = Bool(title=_("Urgent request?"),
+ description=_("Please use this option only when really needed..."),
+ required=True,
+ default=False)
+
+
+class IWorkflowCommentInfo(Interface):
+ """Workflow comment info"""
+
+ comment = Text(title=_("Comment"),
+ description=_("Comment associated with this operation"),
+ required=False)
+
+
+class IWorkflowManagementTask(Interface):
+ """Workflow management task marker interface"""