simpleflow

import os import simpleflow from simpleflow import workflow, activity, futures from simpleflow.local.executor import Executor # Define an activity @activity.with_attributes(task_list='my_activity_task_list', version='1.0') def say_hello(name): """An activity that returns a greeting.""" print(f"Activity 'say_hello' received: {name}") return f"Hello, {name}!" # Define a workflow @workflow.with_attributes(task_list='my_workflow_task_list', version='1.0') class GreetingWorkflow(workflow.Workflow): """A simple workflow that uses the say_hello activity.""" def __init__(self): super(GreetingWorkflow, self).__init__() # Bind the activity to the workflow instance self.say_hello_activity = say_hello def run(self, name): print(f"Workflow 'GreetingWorkflow' started with input: {name}") # Schedule the activity and get a Future object hello_future = self.say_hello_activity(name) # In a real SWF execution, futures.wait() would block until the activity completes. # For the local executor, the result is often resolved synchronously. # Accessing .result will retrieve the value when ready. final_greeting = hello_future.result print(f"Workflow received result from activity: {final_greeting}") return final_greeting # --- Quickstart Execution (using local executor for demonstration) --- if __name__ == "__main__": # The local executor allows running workflows without connecting to AWS SWF. # It executes activities and workflows synchronously in the same process. executor = Executor() print("\n--- Executing GreetingWorkflow locally ---") # Run the workflow. Arguments to the workflow's `run` method are passed as a tuple. # The executor's `run` method returns the final result of the workflow. workflow_input_name = "Simpleflow User" final_output = executor.run(GreetingWorkflow, (workflow_input_name,)) print(f"\n--- Workflow Execution Complete ---") print(f"Input name: '{workflow_input_name}'") print(f"Final output: '{final_output}'") assert final_output == f"Hello, {workflow_input_name}!" print("Local execution successful!")