add more complicated example

Author: JaGeo

example_workflows/nested/get_jobflow.py

@@ -1,6 +1,6 @@
 from python_workflow_definition.jobflow import load_workflow_json
 
-flow=load_workflow_json("main.pwd.json")
+flow=load_workflow_json("main2.pwd.json")
 
 from jobflow import run_locally
 flow.draw_graph(figsize=(3, 3)).show()

example_workflows/nested/jobflow.ipynb

@@ -0,0 +1,21 @@
+{
+  "version": "0.1.1",
+  "nodes": [
+    { "id": 0, "value": 3, "type": "input", "name": "a" },
+    { "id": 1, "value": 2, "type": "input", "name": "b" },
+    { "id": 2, "value": 4, "type": "input", "name": "c" },
+    { "id": 3, "type": "function", "value": "workflow.get_prod_and_div" },
+    { "id": 4, "type": "workflow", "value": "prod_div.json" },
+    { "id": 5, "type": "function", "value": "workflow.get_sum" },
+    { "id": 6, "type": "output", "name": "final_result" }
+  ],
+  "edges": [
+    { "target": 3, "targetPort": "x", "source": 0, "sourcePort": null },
+    { "target": 3, "targetPort": "y", "source": 2, "sourcePort": null },
+    { "target": 4, "targetPort": "x", "source": 3, "sourcePort": "prod" },
+    { "target": 4, "targetPort": "y", "source": 3, "sourcePort": "div" },
+    { "target": 5, "targetPort": "x", "source": 4, "sourcePort": "result" },
+    { "target": 5, "targetPort": "y", "source": 1, "sourcePort": null },
+    { "target": 6, "targetPort": null, "source": 5, "sourcePort": null }
+  ]
+}

example_workflows/nested/main2.pwd.json

@@ -0,0 +1,16 @@
+{
+  "version": "0.1.1",
+  "nodes": [
+    { "id": 0, "type": "workflow", "value": "prod_div.json" },
+    { "id": 1, "value": 4, "type": "input", "name": "a" },
+    { "id": 2, "value": 3, "type": "input", "name": "b" },
+    { "id": 3, "type": "workflow", "value": "prod.json" },
+    { "id": 4, "type": "output", "name": "final_result" }
+  ],
+  "edges": [
+    { "target": 0, "targetPort": "x", "source": 1, "sourcePort": null },
+    { "target": 0, "targetPort": "y", "source": 2, "sourcePort": null },
+    { "target": 3, "targetPort": "x", "source": 0, "sourcePort": "result" },
+    { "target": 4, "targetPort": "null", "source": 3, "sourcePort": "result" }
+  ]
+}

example_workflows/nested/main_two_wf.pwd.json

@@ -0,0 +1,12 @@
+{
+  "version": "0.1.1",
+  "nodes": [
+    { "id": 0, "type": "function", "value": "workflow.get_square" },
+    { "id": 1, "type": "input", "value": 1, "name": "x" },
+    { "id": 2, "type": "output", "name": "result" }
+  ],
+  "edges": [
+    { "target": 0, "targetPort": "x", "source": 1, "sourcePort": null },
+    { "target": 2, "targetPort": null, "source": 1, "sourcePort": null }
+  ]
+}

example_workflows/nested/prod.json

@@ -235,21 +235,33 @@ def _get_workflow(
     nodes_dict: dict, input_dict: dict, total_dict: dict, source_handles_dict: dict
 ) -> list:
     def get_attr_helper(obj, source_handle):
-        if source_handle is None:
-            return getattr(obj, "output")
+        print("here2")
+        print(obj)
+        if type(obj) is not list:
+            if source_handle is None:
+                return getattr(obj, "output")
+            else:
+                return getattr(getattr(obj, "output"), source_handle)
         else:
-            return getattr(getattr(obj, "output"), source_handle)
+            if source_handle is None:
+                return getattr(obj[-1], "output")
+            else:
+                return getattr(getattr(obj[-1], "output"), source_handle)
 
     memory_dict = {}
-    print(total_dict)
+
+    output_reference=None
     for k in total_dict.keys():
         v = nodes_dict[k]
         if type(v) is list:
+            # i need to export the uuid and output references
+            # from workflows and jobs
             fn=v
             job1_dict=v[0].as_dict()
             uuid=job1_dict["uuid"]
             mod = import_module(job1_dict["function"]["@module"])
             method=getattr(mod, job1_dict["function"]["@callable"])
+
             if k in source_handles_dict.keys():
                 new_job1 = job(
                     method=method,
@@ -258,39 +270,64 @@ def get_attr_helper(obj, source_handle):
                 )
             else:
                 new_job1 = job(method=method, uuid=uuid)
-            kwargs = {
-                kw: (
-                    input_dict[vw[SOURCE_LABEL]]
-                    if vw[SOURCE_LABEL] in input_dict
-                    else get_attr_helper(
-                        obj=memory_dict[vw[SOURCE_LABEL]],
-                        source_handle=vw[SOURCE_PORT_LABEL],
-                    )
-                )
-                for kw, vw in total_dict[k].items()
-            }
+            kwargs = {}
+
+            for kw, vw in total_dict[k].items():
+                if output_reference is not None:
+                    # not sure this works?
+                    kwargs[kw] = output_reference
+                    output_reference = None
+                else:
+                    if vw[SOURCE_LABEL] in input_dict:
+                        kwargs[kw] = input_dict[vw[SOURCE_LABEL]]
+                    else:
+                        kwargs[kw] = get_attr_helper(
+                            obj=memory_dict[vw[SOURCE_LABEL]],
+                            source_handle=vw[SOURCE_PORT_LABEL],
+                        )
+            output_reference=v[-1].output
+
+
             memory_stuff = new_job1(**kwargs)
+
             fn = [memory_stuff]+v[1:]
+
             memory_dict[k] =Flow(fn)
+
         if isfunction(v):
+            #if output_reference is None:
+
+            #    output_reference=None
+
             if k in source_handles_dict.keys():
                 fn = job(
                     method=v,
                     data=[el for el in source_handles_dict[k] if el is not None],
                 )
             else:
                 fn = job(method=v)
-            kwargs = {
-                kw: (
-                    input_dict[vw[SOURCE_LABEL]]
-                    if vw[SOURCE_LABEL] in input_dict
-                    else get_attr_helper(
+
+            kwargs={}
+
+            for kw, vw in total_dict[k].items():
+                if output_reference is not None:
+                    # not sure this works?
+                    kwargs[kw] = output_reference
+                    output_reference=None
+                else:    
+                    if  vw[SOURCE_LABEL] in input_dict:
+                        kwargs[kw]=input_dict[vw[SOURCE_LABEL]]
+                    else:
+                        kwargs[kw]= get_attr_helper(
                         obj=memory_dict[vw[SOURCE_LABEL]],
                         source_handle=vw[SOURCE_PORT_LABEL],
-                    )
-                )
-                for kw, vw in total_dict[k].items()
-            }
+                        )
+
+            print("here")
+            print(kwargs)
+
+
+
             memory_dict[k] = fn(**kwargs)
 
     return list(memory_dict.values())
@@ -312,7 +349,7 @@ def load_workflow_json(file_name: str) -> Flow:
         source_handles_dict=source_handles_dict,
     )
     print(task_lst)
-    return Flow(task_lst)
+    return Flow(task_lst, output=task_lst[-1].output)
 
 
 def recursive_load_workflow_json(file_name: str) -> list: