WIP

ui/sdk/src/hamilton_sdk/adapters.py

@@ -297,6 +297,8 @@ def post_node_execute(
                 },
                 attribute_role="error",
             )
+        task_attr_2 = task_attr.copy()
+        task_attr_2["name"] = "task_attr_2"
         task_run.end_time = datetime.datetime.now(timezone.utc)
         tracking_state.update_task(node_.name, task_run)
         task_update = dict(
@@ -309,7 +311,7 @@ def post_node_execute(
         )
         self.client.update_tasks(
             self.dw_run_ids[run_id],
-            attributes=[task_attr],
+            attributes=[task_attr, task_attr_2],
             task_updates=[task_update],
             in_samples=[task_run.is_in_sample],
         )

ui/sdk/src/hamilton_sdk/api/clients.py

@@ -5,7 +5,6 @@
 import threading
 import time
 from collections import defaultdict
-from functools import reduce
 from typing import Any, Callable, Dict, List
 from urllib.parse import urlencode
 
@@ -234,13 +233,19 @@ def flush(self, batch):
                         task_updates[task_update["node_name"]].append(task_update)
 
             # this assumes correct ordering of the attributes and task_updates
-            attributes_list = [
-                reduce(lambda x, y: {**x, **y}, attributes[node_name]) for node_name in attributes
-            ]
-            task_updates_list = [
-                reduce(lambda x, y: {**x, **y}, task_updates[node_name])
-                for node_name in task_updates
-            ]
+            # attributes_list = [
+            #     reduce(lambda x, y: {**x, **y}, attributes[node_name]) for node_name in attributes
+            # ]
+            # task_updates_list = [
+            #     reduce(lambda x, y: {**x, **y}, task_updates[node_name])
+            #     for node_name in task_updates
+            # ]
+            attributes_list = []
+            for node_name in attributes:
+                attributes_list.extend(attributes[node_name])
+            task_updates_list = []
+            for node_name in task_updates:
+                task_updates_list.extend(task_updates[node_name])
 
             response = requests.put(
                 f"{self.base_url}/dag_runs_bulk?dag_run_id={dag_run_id}",