Add LambdaFunctionTask

Currently SWF only; needs to be moved upward. And with some rough
edges...

Signed-off-by: Yves Bastide <yves@botify.com>

Author: ybastide

examples/lambda.py

@@ -0,0 +1,23 @@
+from simpleflow import Workflow, futures
+from simpleflow.lambda_function import LambdaFunction
+from simpleflow.swf.task import LambdaFunctionTask
+
+
+class LambdaWorkflow(Workflow):
+    name = 'basic'
+    version = 'example'
+    task_list = 'example'
+    # lambda_role = ''
+
+    def run(self):
+        future = self.submit(
+            LambdaFunctionTask(
+                LambdaFunction(
+                    'hello-world-python',
+                    idempotent=True,
+                ),
+                8,
+                foo='bar',
+            )
+        )
+        print(futures.wait(future))

simpleflow/lambda_function.py

@@ -0,0 +1,11 @@
+from simpleflow.base import Submittable
+
+
+class LambdaFunction(Submittable):
+    def __init__(self,
+                 name,
+                 start_to_close_timeout=None,
+                 idempotent=None):
+        self.name = name
+        self.start_to_close_timeout = start_to_close_timeout
+        self.idempotent = idempotent

simpleflow/swf/executor.py

@@ -36,6 +36,7 @@
     MarkerTask,
     TimerTask,
     CancelTimerTask,
+    LambdaFunctionTask,
 )
 from simpleflow.utils import (
     hex_hash,
@@ -342,7 +343,7 @@ def _get_future_from_child_workflow_event(self, event):
             future.set_finished(json_loads_or_raw(event['result']))
         elif state == 'failed':
             future.set_exception(exceptions.TaskFailed(
-                name=event['id'],
+                name=event['name'],
                 reason=event['reason'],
                 details=event.get('details'),
             ))
@@ -360,6 +361,43 @@ def _get_future_from_child_workflow_event(self, event):
 
         return future
 
+    def _get_future_from_lambda_function_event(self, event):
+        """
+
+        :param event: child workflow event
+        :type  event: dict[str, Any]
+        :return:
+        :rtype: futures.Future
+        """
+        future = futures.Future()
+        state = event['state']
+
+        if state == 'scheduled':
+            pass
+        elif state == 'schedule_failed':
+            logger.info('failed to schedule {}: {}'.format(
+                event['name'],
+                event['cause'],
+            ))
+            return None
+        elif state == 'started':
+            future.set_running()
+        elif state == 'completed':
+            future.set_finished(json_loads_or_raw(event['result']))
+        elif state == 'failed':
+            future.set_exception(exceptions.TaskFailed(
+                name=event['name'],
+                reason=event['reason'],
+                details=event.get('details'),
+            ))
+        elif state == 'timed_out':
+            future.set_exception(exceptions.TimeoutError(
+                event['timeout_type'],
+                None,
+            ))
+
+        return future
+
     def _get_future_from_marker_event(self, a_task, event):
         """Maps a marker event to a Future with the corresponding
         state.
@@ -497,6 +535,19 @@ def find_child_workflow_event(self, a_task, history):
         """
         return history.child_workflows.get(a_task.id)
 
+    def find_lambda_function_event(self, a_task, history):
+        """
+        Get the event corresponding to a lambda function, if any.
+
+        :param a_task:
+        :type a_task: LambdaFunctionTask
+        :param history:
+        :type history: simpleflow.history.History
+        :return:
+        :rtype: Optional[dict]
+        """
+        return history.lambda_functions.get(a_task.id)
+
     def find_signal_event(self, a_task, history):
         """
         Get the event corresponding to a signal, if any.
@@ -570,6 +621,7 @@ def find_timer_event(self, a_task, history):
         MarkerTask: find_marker_event,
         TimerTask: find_timer_event,
         CancelTimerTask: find_timer_event,
+        LambdaFunctionTask: find_lambda_function_event,
     }
 
     def find_event(self, a_task, history):
@@ -644,12 +696,30 @@ def resume_child_workflow(self, a_task, event):
 
         return future
 
+    def resume_lambda_function(self, a_task, event):
+        """
+        Resume a child workflow.
+
+        :param a_task:
+        :type a_task: LambdaTask
+        :param event:
+        :type event: dict
+        :return:
+        :rtype: simpleflow.futures.Future
+        """
+        future = self._get_future_from_lambda_function_event(event)
+
+        if future.finished and future.exception:
+            raise future.exception
+
+        return future
+
     def schedule_task(self, a_task, task_list=None):
         """
         Let a task schedule itself.
         If too many decisions are in flight, add a timer decision and raise ExecutionBlocked.
         :param a_task:
-        :type a_task: ActivityTask | WorkflowTask | SignalTask | MarkerTask
+        :type a_task: SwfTask
         :param task_list:
         :type task_list: Optional[str]
         :raise: exceptions.ExecutionBlocked if too many decisions waiting
@@ -710,6 +780,7 @@ def _add_start_timer_decision(self, id):
         'external_workflow': get_future_from_external_workflow_event,
         'marker': _get_future_from_marker_event,
         'timer': _get_future_from_timer_event,
+        'lambda_function': resume_lambda_function,
     }
 
     def resume(self, a_task, *args, **kwargs):

simpleflow/swf/task.py

@@ -5,6 +5,11 @@
 from simpleflow import task
 from simpleflow.utils import json_dumps
 
+
+if False:
+    from simpleflow.lambda_function import LambdaFunction
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -351,3 +356,34 @@ def schedule(self, *args, **kwargs):
             id=self.timer_id,
         )
         return [decision]
+
+
+class LambdaFunctionTask(task.Task, SwfTask):
+
+    def __init__(self, lambda_function, *args, **kwargs):
+        # type: (LambdaFunction) -> None
+        self.lambda_function = lambda_function
+        self.args = self.resolve_args(*args)
+        self.kwargs = self.resolve_kwargs(**kwargs)
+        self.idempotent = lambda_function.idempotent
+        self.id = None
+
+    @property
+    def name(self):
+        return self.lambda_function.name
+
+    def schedule(self, *args, **kwargs):
+        input = {
+            'args': self.args,
+            'kwargs': self.kwargs,
+        }
+
+        decision = swf.models.decision.LambdaFunctionDecision(
+            'schedule',
+            id=self.id,
+            name=self.name,
+            input=input,
+            start_to_close_timeout=str(self.lambda_function.start_to_close_timeout)
+            if self.lambda_function.start_to_close_timeout else None,
+        )
+        return [decision]