Support for multiple outputs in a dict.

lynxkite-app/web/src/workspace/nodes/LynxKiteNode.tsx

@@ -42,10 +42,12 @@ function getHandles(inputs: any[], outputs: any[]) {
     e.index = counts[e.position];
     counts[e.position]++;
   }
+  const simpleHorizontal =
+    counts.top === 0 && counts.bottom === 0 && counts.left <= 1 && counts.right <= 1;
+  const simpleVertical =
+    counts.left === 0 && counts.right === 0 && counts.top <= 1 && counts.bottom <= 1;
   for (const e of handles) {
     e.offsetPercentage = (100 * (e.index + 1)) / (counts[e.position] + 1);
-    const simpleHorizontal = counts.top === 0 && counts.bottom === 0 && handles.length <= 2;
-    const simpleVertical = counts.left === 0 && counts.right === 0 && handles.length <= 2;
     e.showLabel = !simpleHorizontal && !simpleVertical;
   }
   return handles;

lynxkite-graph-analytics/src/lynxkite_graph_analytics/core.py

@@ -161,11 +161,15 @@ def disambiguate_edges(ws: workspace.Workspace):
         seen.add((edge.target, edge.targetHandle))
 
 
+# Outputs are tracked by node ID and output ID.
+Outputs = dict[(str, str), typing.Any]
+
+
 @ops.register_executor(ENV)
 async def execute(ws: workspace.Workspace):
-    catalog: dict[str, ops.Op] = ops.CATALOGS[ws.env]
+    catalog = ops.CATALOGS[ws.env]
     disambiguate_edges(ws)
-    outputs = {}
+    outputs: Outputs = {}
     nodes = {node.id: node for node in ws.nodes}
     todo = set(nodes.keys())
     progress = True
@@ -173,8 +177,12 @@ async def execute(ws: workspace.Workspace):
         progress = False
         for id in list(todo):
             node = nodes[id]
-            input_nodes = [edge.source for edge in ws.edges if edge.target == id]
-            if all(input in outputs for input in input_nodes):
+            inputs_done = [
+                (edge.source, edge.sourceHandle) in outputs
+                for edge in ws.edges
+                if edge.target == id
+            ]
+            if all(inputs_done):
                 # All inputs for this node are ready, we can compute the output.
                 todo.remove(id)
                 progress = True
@@ -187,7 +195,9 @@ async def await_if_needed(obj):
     return obj
 
 
-async def _execute_node(node, ws, catalog, outputs):
+async def _execute_node(
+    node: workspace.WorkspaceNode, ws: workspace.Workspace, catalog: ops.Catalog, outputs: Outputs
+):
     params = {**node.data.params}
     op = catalog.get(node.data.title)
     if not op:
@@ -196,7 +206,9 @@ async def _execute_node(node, ws, catalog, outputs):
     node.publish_started()
     # TODO: Handle multi-inputs.
     input_map = {
-        edge.targetHandle: outputs[edge.source] for edge in ws.edges if edge.target == node.id
+        edge.targetHandle: outputs[edge.source, edge.sourceHandle]
+        for edge in ws.edges
+        if edge.target == node.id
     }
     # Convert inputs types to match operation signature.
     try:
@@ -228,8 +240,12 @@ async def _execute_node(node, ws, catalog, outputs):
             traceback.print_exc()
         result = ops.Result(error=str(e))
     result.input_metadata = [_get_metadata(i) for i in inputs]
-    if result.output is not None:
-        outputs[node.id] = result.output
+    if isinstance(result.output, dict):
+        for k, v in result.output.items():
+            outputs[node.id, k] = v
+    elif result.output is not None:
+        [k] = op.outputs
+        outputs[node.id, k.name] = result.output
     node.publish_result(result)