Fix writing workflow results to database

server/server/main.py

@@ -143,7 +143,7 @@ async def run_workflow(
 ):
     try:
         workflow = await db.get_workflow(request.id, config.app_id)
-        runner = WorkflowRunner()
+        runner = WorkflowRunner(db=db)
         await runner.run_workflow(workflow)
         return workflow
     except Exception as e:

server/workflow_runner/workflow_runner.py

@@ -23,24 +23,37 @@
 from collections import deque
 from node_runner import NodeRunner
 from database import Database
+import json
 
 
 class WorkflowRunner:
-    def __init__(self):
-        pass
+    def __init__(self, db: Database):
+        self.db = db
 
     async def save_workflow_run(
-        self, db: Database, workflow_id: str, status: WorkflowRunStatus, results: Dict
+        self, workflow_id: str, status: WorkflowRunStatus, results: Dict
     ):
+        # Save the first 100 rows of each result table
+        validated_results = {}
+        for node_id, result in results.items():
+            if result is not None:
+                new_result = result[:100]
+                # remove all NaN values and replace them with None
+                new_result = [
+                    [None if pd.isna(value) else value for value in row]
+                    for row in new_result
+                ]
+                validated_results[node_id] = new_result
+
         workflow_run = WorkflowRun(
             workflow_id=workflow_id,
-            results=results,
+            results=validated_results,
             status=status,
         )
 
-        await db.save_workflow_run(workflow_run)
+        await self.db.save_workflow_run(workflow_run)
 
-    async def run_workflow(self, workflow: Workflow, db) -> Dict:
+    async def run_workflow(self, workflow: Workflow) -> Dict:
         nodes = workflow.nodes
         edges = workflow.edges
 
@@ -84,12 +97,10 @@ async def run_workflow(self, workflow: Workflow, db) -> Dict:
         if len(topological_order) == len(nodes):
             # All nodes executed successfully in topological order.
             await self.save_workflow_run(
-                db, workflow.id, WorkflowRunStatus.successful, results
+                workflow.id, WorkflowRunStatus.successful, results
             )
             return results
         else:
             # There exists a cycle in the graph.
-            await self.save_workflow_run(
-                db, workflow.id, WorkflowRunStatus.failed, results
-            )
+            await self.save_workflow_run(workflow.id, WorkflowRunStatus.failed, results)
             raise ValueError("Workflow contains a cycle")