Merge branch 'release/0.1.31'

Author: Oleiade

swf/__init__.py

@@ -1,7 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
-version = (0, 1, 29)
+version = (0, 1, 31)
 
 __title__ = "python-simple-workflow"
 __author__ = "Oleiade"

swf/actors/decider.py

@@ -35,11 +35,23 @@ def complete(self, task_token,
                            made by the decider while processing this decision task
         :type   decisions: list (of swf.models.decision.Decision)
         """
-        self.connection.respond_decision_task_completed(
-            task_token,
-            decisions,
-            execution_context,
-        )
+        try:
+            self.connection.respond_decision_task_completed(
+                task_token,
+                decisions,
+                execution_context,
+            )
+        except SWFResponseError as e:
+            if e.error_code == 'UnknownResourceFault':
+                raise DoesNotExistError(
+                    "Unable to complete decision task with token: {}.\n"
+                    "Possible reasons: decision already completed or "
+                    "workflow execution is closed.\n"
+                    "Details: {}".format(task_token, e.body['message'])
+                )
+
+            raise ResponseError(e.body['message'])
+
 
     def poll(self, task_list=None,
              identity=None,
@@ -76,13 +88,23 @@ def poll(self, task_list=None,
 
         next_page = task.get('nextPageToken')
         while next_page:
-            task = self.connection.poll_for_decision_task(
-                self.domain.name,
-                task_list=task_list,
-                identity=identity,
-                next_page_token=next_page,
-                **kwargs
-            )
+            try:
+                task = self.connection.poll_for_decision_task(
+                    self.domain.name,
+                    task_list=task_list,
+                    identity=identity,
+                    next_page_token=next_page,
+                    **kwargs
+                )
+            except SWFResponseError as e:
+                if e.error_code == 'UnknownResourceFault':
+                    raise DoesNotExistError(
+                        "Unable to poll decision task.\n"
+                        "Reason: workflow execution is probably closed.\n"
+                        "Details: {}".format(task_token, e.body['message'])
+                    )
+
+                raise ResponseError(e.body['message'])
 
             token = task.get('taskToken')
             if token is None:

swf/actors/worker.py

@@ -42,7 +42,18 @@ def cancel(self, task_token, details=None):
         :param  details: provided details about cancel
         :type   details: string
         """
-        return self.connection.respond_activity_task_canceled(task_token)
+        try:
+            return self.connection.respond_activity_task_canceled(task_token)
+        except SWFResponseError as e:
+            if e.error_code == 'UnknownResourceFault':
+                raise DoesNotExistError(
+                    "Unable to cancel activity task with token: {}.\n"
+                    "Possible reasons: task already canceled or "
+                    "workflow execution is closed.\n"
+                    "Details: {}".format(task_token)
+                )
+
+                raise ResponseError(e.body['message'])
 
     def complete(self, task_token, result=None):
         """Responds to ``swf` that the activity task is completed
@@ -53,10 +64,21 @@ def complete(self, task_token, result=None):
         :param  result: The result of the activity task.
         :type   result: string
         """
-        return self.connection.respond_activity_task_completed(
-            task_token,
-            result
-        )
+        try:
+            return self.connection.respond_activity_task_completed(
+                task_token,
+                result
+            )
+        except SWFResponseError as e:
+            if e.error_code == 'UnknownResourceFault':
+                raise DoesNotExistError(
+                    "Unable to complete activity task with token: {}.\n"
+                    "Possible reasons: task already completed or "
+                    "workflow execution is closed.\n"
+                    "Details: {}".format(task_token, e.body['message'])
+                )
+
+            raise ResponseError(e.body['message'])
 
     def fail(self, task_token, details=None, reason=None):
         """Replies to ``swf`` that the activity task failed
@@ -70,11 +92,22 @@ def fail(self, task_token, details=None, reason=None):
         :param  reason: Description of the error that may assist in diagnostics
         :type   reason: string
         """
-        return self.connection.respond_activity_task_failed(
-            task_token,
-            details,
-            reason
-        )
+        try:
+            return self.connection.respond_activity_task_failed(
+                task_token,
+                details,
+                reason
+            )
+        except SWFResponseError as e:
+            if e.error_code == 'UnknownResourceFault':
+                raise DoesNotExistError(
+                    "Unable to fail activity task with token: {}.\n"
+                    "Possible reasons: task already failed or "
+                    "workflow execution is closed"
+                    "Details: {}".format(task_token, e.body['message'])
+                )
+
+            raise ResponseError(e.body['message'])
 
     def heartbeat(self, task_token, details=None):
         """Records activity task heartbeat
@@ -85,10 +118,20 @@ def heartbeat(self, task_token, details=None):
         :param  details: provided details about cancel
         :type   details: string
         """
-        return self.connection.record_activity_task_heartbeat(
-            task_token,
-            details
-        )
+        try:
+            return self.connection.record_activity_task_heartbeat(
+                task_token,
+                details
+            )
+        except SWFResponseError as e:
+            if e.error_code == 'UnknownResourceFault':
+                raise DoesNotExistError(
+                    "Unable to send activity task {} heartbeat.\n"
+                    "Possible reason: workflow execution is closed.\n"
+                    "Details: {}".format(task_token, e.body['message'])
+                )
+
+            raise ResponseError(e.body['message'])
 
     def poll(self, task_list=None, identity=None):
         """Polls for an activity task to process from current
@@ -115,11 +158,21 @@ def poll(self, task_list=None, identity=None):
         task_list = task_list or self.task_list
         identity = identity or self.identity
 
-        polled_activity_data = self.connection.poll_for_activity_task(
-            self.domain.name,
-            task_list,
-            identity=identity
-        )
+        try:
+            polled_activity_data = self.connection.poll_for_activity_task(
+                self.domain.name,
+                task_list,
+                identity=identity
+            )
+        except SWFResponseError as e:
+            if e.error_code == 'UnknownResourceFault':
+                raise DoesNotExistError(
+                    "Unable to poll activity task.\n"
+                    "Possible reason: workflow execution is probably closed.\n"
+                    "Details: {}".format(e.body['message'])
+                )
+
+            raise ResponseError(e.body['message'])
 
         if not 'taskToken' in polled_activity_data:
             raise PollTimeout("Activity Worker poll timed out")