Progress indicator in one_by_one.py.

lynxkite-app/src/lynxkite_app/crdt.py

@@ -197,6 +197,7 @@ async def workspace_changed(name: str, changes: pycrdt.MapEvent, ws_crdt: pycrdt
         getattr(change, "keys", {}).get("__execution_delay", {}).get("newValue", 0)
         for change in changes
     )
+    print(f"Running {name} in {ws_pyd.env}...")
     if delay:
         task = asyncio.create_task(execute(name, ws_crdt, ws_pyd, delay))
         delayed_executions[name] = task

lynxkite-core/src/lynxkite/core/executors/one_by_one.py

@@ -104,11 +104,11 @@ async def execute(ws: workspace.Workspace, catalog, cache=None):
     tasks = {}
     NO_INPUT = object()  # Marker for initial tasks.
     for node in ws.nodes:
-        node.data.error = None
         op = catalog.get(node.data.title)
         if op is None:
-            node.data.error = f'Operation "{node.data.title}" not found.'
+            node.publish_error(f'Operation "{node.data.title}" not found.')
             continue
+        node.publish_error(None)
         # Start tasks for nodes that have no non-batch inputs.
         if all([i.position in "top or bottom" for i in op.inputs.values()]):
             tasks[node.id] = [NO_INPUT]
@@ -123,12 +123,12 @@ async def execute(ws: workspace.Workspace, catalog, cache=None):
                 next_stage.setdefault(n, []).extend(ts)
                 continue
             node = nodes[n]
-            data = node.data
-            op = catalog[data.title]
-            params = {**data.params}
+            op = catalog[node.data.title]
+            params = {**node.data.params}
             if has_ctx(op):
                 params["_ctx"] = contexts[node.id]
             results = []
+            node.publish_started()
             for task in ts:
                 try:
                     inputs = []
@@ -146,11 +146,12 @@ async def execute(ws: workspace.Workspace, catalog, cache=None):
                             cache[key] = output
                         output = cache[key]
                     else:
+                        op.publish_started()
                         result = op(*inputs, **params)
                         output = await await_if_needed(result.output)
                 except Exception as e:
                     traceback.print_exc()
-                    data.error = str(e)
+                    node.publish_error(e)
                     break
                 contexts[node.id].last_result = output
                 # Returned lists and DataFrames are considered multiple tasks.
@@ -161,7 +162,7 @@ async def execute(ws: workspace.Workspace, catalog, cache=None):
                 results.extend(output)
             else:  # Finished all tasks without errors.
                 if result.display:
-                    data.display = await await_if_needed(result.display)
+                    result.display = await await_if_needed(result.display)
                 for edge in edges[node.id]:
                     t = nodes[edge.target]
                     op = catalog[t.data.title]
@@ -172,5 +173,6 @@ async def execute(ws: workspace.Workspace, catalog, cache=None):
                         ).extend(results)
                     else:
                         tasks.setdefault(edge.target, []).extend(results)
+                op.publish_result(result)
         tasks = next_stage
     return contexts

lynxkite-core/src/lynxkite/core/ops.py

@@ -9,6 +9,9 @@ import typing
 from dataclasses import dataclass
 from typing_extensions import Annotated
 
+if typing.TYPE_CHECKING:
+    from . import workspace
+
 CATALOGS = {}
 EXECUTORS = {}
 
@@ -233,9 +236,15 @@ def register_passive_op(env: str, name: str, inputs=[], outputs=["output"], para
 
 
 def register_executor(env: str):
-    """Decorator for registering an executor."""
+    """Decorator for registering an executor.
 
-    def decorator(func):
+    The executor is a function that takes a workspace and executes the operations in it.
+    When it starts executing an operation, it should call `node.publish_started()` to indicate
+    the status on the UI. When the execution is finished, it should call `node.publish_result()`.
+    This will update the UI with the result of the operation.
+    """
+
+    def decorator(func: typing.Callable[[workspace.Workspace], typing.Any]):
         EXECUTORS[env] = func
         return func
 

lynxkite-core/src/lynxkite/core/workspace.py

@@ -49,9 +49,12 @@ class WorkspaceNode(BaseConfig):
 
     def publish_started(self):
         """Notifies the frontend that work has started on this node."""
+        self.data.error = None
         self.data.status = NodeStatus.active
         if hasattr(self, "_crdt"):
-            self._crdt["data"]["status"] = NodeStatus.active
+            with self._crdt.doc.transaction():
+                self._crdt["data"]["error"] = None
+                self._crdt["data"]["status"] = NodeStatus.active
 
     def publish_result(self, result: ops.Result):
         """Sends the result to the frontend. Call this in an executor when the result is available."""
@@ -64,8 +67,10 @@ class WorkspaceNode(BaseConfig):
                 self._crdt["data"]["error"] = result.error
                 self._crdt["data"]["status"] = NodeStatus.done
 
-    def publish_error(self, error: Exception | str):
-        self.publish_result(ops.Result(error=str(error)))
+    def publish_error(self, error: Exception | str | None):
+        """Can be called with None to clear the error state."""
+        result = ops.Result(error=str(error) if error else None)
+        self.publish_result(result)
 
 
 class WorkspaceEdge(BaseConfig):