Merge pull request #46 from zenaton/retry-context

Add runtime context

Author: antoinereyt

CHANGELOG.md

@@ -1,18 +1,13 @@
 # Changelog
 
-## [Unreleased]
+## [0.4.1] - 2019-09-25
 
 ### Added
-
-### Changed
-
-### Deprecated
-
-### Removed
-
-### Fixed
-
-### Security
+- Added a `intent_id` property when dispatching workflows and tasks, sending events to workflows, and
+  pausing/resuming/killing workflows.
+- Execution context for tasks and workflows
+- Optional `on_error_retry_delay` method handling task failures and specifying
+  how many seconds to wait before retrying.
 
 ## [0.4.0] - 2019-08-26
 

zenaton/__init__.py

@@ -1,2 +1,2 @@
-__version__ = '0.4.0'
-__version_id__ = 304
+__version__ = '0.4.1'
+__version_id__ = 401

zenaton/abstracts/task.py

@@ -1,10 +1,42 @@
 import abc
 
 from .job import Job
+from ..contexts.task_context import TaskContext
 
 
 class Task(Job):
 
     @abc.abstractmethod
     def handle(self):
         pass
+
+    """
+        (Optional) Implement this method for automatic retrial of task in case of failures.
+
+        :params error Error
+        :raises Exception when the return type is not falsy or is not positive integer.
+        :return [int, false, None] the amount of seconds to wait
+        before automatically retrying this task. Falsy values will avoid
+        retrial. Other values will raise.
+    """
+    def on_error_retry_delay(self, error):
+        None
+
+    """
+        :return TaskContext
+    """
+    def get_context(self):
+        return self._context or TaskContext()
+
+    """
+        Sets a new context if none has been set yet.
+        This is called from the zenaton agent and will raise if called twice.
+
+        :params context TaskContext
+        :raises Exception when the context was already set.
+    """
+    def set_context(self, context):
+        if hasattr(self, '_context') and self._context != None:
+            raise Exception('Context is already set and cannot be mutated.')
+
+        self._context = context

zenaton/abstracts/workflow.py

@@ -2,6 +2,7 @@
 
 from .job import Job
 
+from ..contexts.workflow_context import WorkflowContext
 
 class Workflow(Job):
 
@@ -11,3 +12,23 @@ def handle(self):
 
     def id(self):
         return None
+
+    
+    """
+        :return TaskContext
+    """
+    def get_context(self):
+        return self._context or WorkflowContext()
+
+    """
+        Sets a new context if none has been set yet.
+        This is called from the zenaton agent and will raise if called twice.
+
+        :context WorkflowContext
+        :raises Exception when the context was already set.
+    """
+    def set_context(self, context):
+        if hasattr(self, '_context') and self._context != None:
+            raise Exception('Context is already set and cannot be mutated.')
+
+        self._context = context

zenaton/contexts/__init__.py

@@ -0,0 +1,27 @@
+class TaskContext():
+    """
+    Represents the current runtime context of a Task.
+
+    The information provided by the context can be useful to alter the
+    behaviour of the task.
+
+    For example, you can use the retry_index to know if a task has been
+    automatically retried or not and how many times, and decide to do
+    something when you did not expect the task to be retried more than X
+    times.
+
+    You can also use the retry_index in the `on_error_retry_delay` method
+    of a task in order to implement complex retry strategies.
+
+    Attributes
+    ----------
+    id : str
+        The UUID identifying the current task.
+
+    retry_index : int
+        The number of times this task has been automatically retried.
+        This counter is reset if you issue a manual retry from your dashboard
+    """
+    def __init__(self, **kwargs):
+        self.id = kwargs.get('id', None)
+        self.retry_index = kwargs.get('retry_index', None)

zenaton/contexts/task_context.py

@@ -0,0 +1,11 @@
+class WorkflowContext():
+    """
+    Represents the current runtime context of a Workflow.
+
+    Attributes
+    ----------
+    id : str
+        The UUID identifying the current workflow
+    """
+    def __init__(self, **kwargs):
+        self.id = kwargs.get('id', None)

zenaton/contexts/workflow_context.py

@@ -29,6 +29,10 @@ def encode(self, data):
         self._encoded = []
         self._decoded = []
         value = dict()
+
+        if isinstance(data, dict) and data.get('_context'):
+            del data['_context']
+
         if self.__is_basic_type(data):
             value[self.KEY_DATA] = data
         else: