dynamic_workflow.py

@inputs(in_str=Types.String)
@dynamic_task
def dynamic_wf_task(wf_params, in_str):
    wf_params.logging.info("Running dynamic task... yielding a code generated sub workflow")

    res = []
    nodes = {}
    previous_node = None
    i = 0
    for s in in_str:
        current_node = other_task(command=s)
        nodes['node-{}'.format(i++)] = current_node
        
        # This piece of code adds execution depedency between nodes to force them to run sequentially.
        if previous_node is not None:
          previous_node >> current_node
        previous_node = current_node
    
    MyUnregisteredWorkflow = workflow(
        inputs={},
        outputs={},
        nodes=nodes,
    )

    # This is an unfortunate setting that will hopefully not be necessary in the future.
    setattr(MyUnregisteredWorkflow, 'auto_assign_name', manual_assign_name)
    MyUnregisteredWorkflow._platform_valid_name = 'unregistered'

    yield MyUnregisteredWorkflow()