diff -r 69194c2ab7bc -r 509f88791c41 src/pyams_workflow/interfaces.py --- /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 +# 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"""