Skip to content

[Feature] Export to Python Workflow Definition #882

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jan-janssen merged 8 commits into from
Jan 10, 2026
Merged

[Feature] Export to Python Workflow Definition #882

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
c2a7d4c
Export to Python Workflow Definition
jan-janssen Jan 4, 2026
338d9c9
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jan 4, 2026
8683249
Format black
pyiron-runner Jan 4, 2026
c8be904
Update dependency.py
jan-janssen Jan 4, 2026
df6c8ea
[pre-commit.ci] auto fixes from pre-commit.com hooks
pre-commit-ci[bot] Jan 4, 2026
b36d3a4
Add test
jan-janssen Jan 5, 2026
e146432
extend test
jan-janssen Jan 5, 2026
c0cf665
fix
jan-janssen Jan 5, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions src/executorlib/executor/flux.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@ class FluxJobExecutor(BaseExecutor):
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

Examples:
Expand Down Expand Up @@ -105,6 +106,7 @@ def __init__(
refresh_rate: float = 0.01,
plot_dependency_graph: bool = False,
plot_dependency_graph_filename: Optional[str] = None,
export_workflow_filename: Optional[str] = None,
log_obj_size: bool = False,
):
"""
Expand Down Expand Up @@ -152,6 +154,7 @@ def __init__(
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

"""
Expand Down Expand Up @@ -189,6 +192,7 @@ def __init__(
refresh_rate=refresh_rate,
plot_dependency_graph=plot_dependency_graph,
plot_dependency_graph_filename=plot_dependency_graph_filename,
export_workflow_filename=export_workflow_filename,
)
)
else:
Expand Down Expand Up @@ -255,6 +259,7 @@ class FluxClusterExecutor(BaseExecutor):
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

Examples:
Expand Down Expand Up @@ -293,6 +298,7 @@ def __init__(
refresh_rate: float = 0.01,
plot_dependency_graph: bool = False,
plot_dependency_graph_filename: Optional[str] = None,
export_workflow_filename: Optional[str] = None,
log_obj_size: bool = False,
):
"""
Expand Down Expand Up @@ -338,6 +344,7 @@ def __init__(
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

"""
Expand Down Expand Up @@ -420,6 +427,7 @@ def __init__(
refresh_rate=refresh_rate,
plot_dependency_graph=plot_dependency_graph,
plot_dependency_graph_filename=plot_dependency_graph_filename,
export_workflow_filename=export_workflow_filename,
)
)

Expand Down
8 changes: 8 additions & 0 deletions src/executorlib/executor/single.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@ class SingleNodeExecutor(BaseExecutor):
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

Examples:
Expand Down Expand Up @@ -94,6 +95,7 @@ def __init__(
refresh_rate: float = 0.01,
plot_dependency_graph: bool = False,
plot_dependency_graph_filename: Optional[str] = None,
export_workflow_filename: Optional[str] = None,
log_obj_size: bool = False,
):
"""
Expand Down Expand Up @@ -138,6 +140,7 @@ def __init__(
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

"""
Expand Down Expand Up @@ -171,6 +174,7 @@ def __init__(
refresh_rate=refresh_rate,
plot_dependency_graph=plot_dependency_graph,
plot_dependency_graph_filename=plot_dependency_graph_filename,
export_workflow_filename=export_workflow_filename,
)
)
else:
Expand Down Expand Up @@ -226,6 +230,7 @@ class TestClusterExecutor(BaseExecutor):
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

Examples:
Expand Down Expand Up @@ -262,6 +267,7 @@ def __init__(
refresh_rate: float = 0.01,
plot_dependency_graph: bool = False,
plot_dependency_graph_filename: Optional[str] = None,
export_workflow_filename: Optional[str] = None,
log_obj_size: bool = False,
):
"""
Expand Down Expand Up @@ -299,6 +305,7 @@ def __init__(
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

"""
Expand Down Expand Up @@ -358,6 +365,7 @@ def __init__(
refresh_rate=refresh_rate,
plot_dependency_graph=plot_dependency_graph,
plot_dependency_graph_filename=plot_dependency_graph_filename,
export_workflow_filename=export_workflow_filename,
)
)

Expand Down
8 changes: 8 additions & 0 deletions src/executorlib/executor/slurm.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ class SlurmClusterExecutor(BaseExecutor):
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

Examples:
Expand Down Expand Up @@ -101,6 +102,7 @@ def __init__(
refresh_rate: float = 0.01,
plot_dependency_graph: bool = False,
plot_dependency_graph_filename: Optional[str] = None,
export_workflow_filename: Optional[str] = None,
log_obj_size: bool = False,
):
"""
Expand Down Expand Up @@ -146,6 +148,7 @@ def __init__(
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

"""
Expand Down Expand Up @@ -225,6 +228,7 @@ def __init__(
refresh_rate=refresh_rate,
plot_dependency_graph=plot_dependency_graph,
plot_dependency_graph_filename=plot_dependency_graph_filename,
export_workflow_filename=export_workflow_filename,
)
)

Expand Down Expand Up @@ -275,6 +279,7 @@ class SlurmJobExecutor(BaseExecutor):
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

Examples:
Expand Down Expand Up @@ -312,6 +317,7 @@ def __init__(
refresh_rate: float = 0.01,
plot_dependency_graph: bool = False,
plot_dependency_graph_filename: Optional[str] = None,
export_workflow_filename: Optional[str] = None,
log_obj_size: bool = False,
):
"""
Expand Down Expand Up @@ -360,6 +366,7 @@ def __init__(
plot_dependency_graph (bool): Plot the dependencies of multiple future objects without executing them. For
debugging purposes and to get an overview of the specified dependencies.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.
log_obj_size (bool): Enable debug mode which reports the size of the communicated objects.

"""
Expand Down Expand Up @@ -394,6 +401,7 @@ def __init__(
refresh_rate=refresh_rate,
plot_dependency_graph=plot_dependency_graph,
plot_dependency_graph_filename=plot_dependency_graph_filename,
export_workflow_filename=export_workflow_filename,
)
)
else:
Expand Down
26 changes: 20 additions & 6 deletions src/executorlib/task_scheduler/interactive/dependency.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@
)
from executorlib.task_scheduler.base import TaskSchedulerBase
from executorlib.task_scheduler.interactive.dependency_plot import (
export_dependency_graph_function,
generate_nodes_and_edges_for_plotting,
generate_task_hash_for_plotting,
plot_dependency_graph_function,
Expand All @@ -28,6 +29,7 @@ class DependencyTaskScheduler(TaskSchedulerBase):
refresh_rate (float, optional): The refresh rate for updating the executor queue. Defaults to 0.01.
plot_dependency_graph (bool, optional): Whether to generate and plot the dependency graph. Defaults to False.
plot_dependency_graph_filename (str): Name of the file to store the plotted graph in.
export_workflow_filename (str): Name of the file to store the exported workflow graph in.

Attributes:
_future_hash_dict (Dict[str, Future]): A dictionary mapping task hash to future object.
Expand All @@ -44,6 +46,7 @@ def __init__(
refresh_rate: float = 0.01,
plot_dependency_graph: bool = False,
plot_dependency_graph_filename: Optional[str] = None,
export_workflow_filename: Optional[str] = None,
) -> None:
super().__init__(max_cores=max_cores)
self._process_kwargs = {
Expand All @@ -61,7 +64,11 @@ def __init__(
self._future_hash_dict: dict = {}
self._task_hash_dict: dict = {}
self._plot_dependency_graph_filename = plot_dependency_graph_filename
if plot_dependency_graph_filename is None:
self._export_workflow_filename = export_workflow_filename
if (
plot_dependency_graph_filename is None
and export_workflow_filename is not None
):
self._generate_dependency_graph = plot_dependency_graph
else:
self._generate_dependency_graph = True
Expand Down Expand Up @@ -209,11 +216,18 @@ def __exit__(
v: k for k, v in self._future_hash_dict.items()
},
)
return plot_dependency_graph_function(
node_lst=node_lst,
edge_lst=edge_lst,
filename=self._plot_dependency_graph_filename,
)
if self._export_workflow_filename is not None:
return export_dependency_graph_function(
node_lst=node_lst,
edge_lst=edge_lst,
file_name=self._export_workflow_filename,
)
else:
return plot_dependency_graph_function(
node_lst=node_lst,
edge_lst=edge_lst,
filename=self._plot_dependency_graph_filename,
)
else:
return None

Expand Down
75 changes: 75 additions & 0 deletions src/executorlib/task_scheduler/interactive/dependency_plot.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
import inspect
import json
import os.path
from concurrent.futures import Future
from typing import Optional

import cloudpickle
import numpy as np

from executorlib.standalone.select import FutureSelector

Expand Down Expand Up @@ -230,3 +232,76 @@ def plot_dependency_graph_function(
from IPython.display import SVG, display # noqa

display(SVG(nx.nx_agraph.to_agraph(graph).draw(prog="dot", format="svg")))


def export_dependency_graph_function(
node_lst: list, edge_lst: list, file_name: str = "workflow.json"
):
"""
Export the graph visualization of nodes and edges as a JSON dictionary.

Args:
node_lst (list): List of nodes.
edge_lst (list): List of edges.
file_name (str): Name of the file to store the exported graph in.
"""
pwd_nodes_lst = []
for n in node_lst:
if n["type"] == "function":
pwd_nodes_lst.append(
{"id": n["id"], "type": n["type"], "value": n["value"]}
)
elif n["type"] == "input" and isinstance(n["value"], np.ndarray):
pwd_nodes_lst.append(
{
"id": n["id"],
"type": n["type"],
"value": n["value"].tolist(),
"name": n["name"],
}
)
else:
pwd_nodes_lst.append(
{
"id": n["id"],
"type": n["type"],
"value": n["value"],
"name": n["name"],
}
)

final_node = {"id": len(pwd_nodes_lst), "type": "output", "name": "result"}
pwd_nodes_lst.append(final_node)
pwd_edges_lst = [
(
{
"target": e["end"],
"targetPort": e["label"],
"source": e["start"],
"sourcePort": None,
}
if "start_label" not in e
else {
"target": e["end"],
"targetPort": e["end_label"],
"source": e["start"],
"sourcePort": e["start_label"],
}
)
for e in edge_lst
]
pwd_edges_lst.append(
{
"target": final_node["id"],
"targetPort": None,
"source": max([e["target"] for e in pwd_edges_lst]),
"sourcePort": None,
}
)
Comment on lines +293 to +300

@coderabbitai coderabbitai Bot Jan 4, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Handle empty edge list to prevent ValueError on max().

If edge_lst is empty, pwd_edges_lst will be empty, and max([e["target"] for e in pwd_edges_lst]) will raise ValueError: max() arg is an empty sequence. This can occur with single-node workflows or graphs without edges.

🔎 Proposed fix 
-    pwd_edges_lst.append(
-        {
-            "target": final_node["id"],
-            "targetPort": None,
-            "source": max([e["target"] for e in pwd_edges_lst]),
-            "sourcePort": None,
-        }
-    )
+    if pwd_edges_lst:
+        pwd_edges_lst.append(
+            {
+                "target": final_node["id"],
+                "targetPort": None,
+                "source": max(e["target"] for e in pwd_edges_lst),
+                "sourcePort": None,
+            }
+        )
🤖 Prompt for AI Agents 
In src/executorlib/task_scheduler/interactive/dependency_plot.py around lines
293-300, the code calls max() on pwd_edges_lst which raises ValueError when
pwd_edges_lst is empty; update the logic to handle an empty list by computing
source only if pwd_edges_lst has elements (e.g., source = max(e["target"] for e
in pwd_edges_lst)), otherwise set source to a safe fallback (for example
final_node["id"] or None) or skip adding the edge entirely; implement a short
conditional that selects the fallback and then append the dict using that source
to avoid the max() on an empty sequence.

pwd_dict = {
"version": "0.1.0",
"nodes": pwd_nodes_lst,
"edges": pwd_edges_lst,
}
with open(file_name, "w") as f:
json.dump(pwd_dict, f, indent=4)
Loading