Add Pre-commit-ci (#196)

* Configure Ruff for linting in pyproject.toml

Add Ruff configuration for linting with specified rules.

* Add ruff pre-commit hooks for linting and formatting

* do not apply on notebooks for now

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fixes

* black fixes

* fixes

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* remove commended out code

* more fixes

* type fix

* more fixes

* more fixes

* black fixes

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* black fixes

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix

* black formatting

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: jan-janssen

.pre-commit-config.yaml

@@ -0,0 +1,10 @@
+repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.15.22
+    hooks:
+      - id: ruff
+        name: ruff lint
+        args: ["--fix"]
+        files: ^src/python_workflow_definition/
+      - id: ruff-format
+        name: ruff format

pyproject.toml

@@ -42,6 +42,39 @@ plot = [
     "ipython>=7.33.0,<=9.8.0",
 ]
 
+[tool.ruff]
+exclude = ["documentation", "example_workflows", "tests", "binder", "_version.py"]
+
+[tool.ruff.lint]
+select = [
+    # pycodestyle
+    "E",
+    # Pyflakes
+    "F",
+    # pyupgrade
+    "UP",
+    # flake8-bugbear
+    "B",
+    # flake8-simplify
+    "SIM",
+    # isort
+    "I",
+    # flake8-comprehensions
+    "C4",
+    # eradicate
+    "ERA",
+    # pylint
+    "PL",
+]
+ignore = [
+    # ignore line-length violations
+    "E501",
+    # Too many statements
+    "PLR0915",
+    # Too many branches
+    "PLR0912",
+]
+
 [tool.hatch.build]
 include = [
     "src/python_workflow_definition"

src/python_workflow_definition/aiida.py

@@ -99,7 +99,6 @@ def write_workflow_json(wg: WorkGraph, file_name: str) -> None:
     GRAPH_LEVEL_NAMES = ["graph_inputs", "graph_outputs", "graph_ctx"]
 
     for node in wg.tasks:
-
         if node.name in GRAPH_LEVEL_NAMES:
             continue
 

src/python_workflow_definition/cwl/__init__.py

@@ -20,17 +20,15 @@
 
 
 def _get_function_argument(argument: str, position: int = 3) -> dict:
-    return {
-        argument
-        + "_file": {
-            "type": "File",
-            "inputBinding": {
-                "prefix": "--arg_" + argument + "=",
-                "separate": False,
-                "position": position,
-            },
+    argument_dict = {
+        "type": "File",
+        "inputBinding": {
+            "prefix": "--arg_" + argument + "=",
+            "separate": False,
+            "position": position,
         },
     }
+    return {argument + "_file": argument_dict}
 
 
 def _get_function_template(function_name: str) -> dict:
@@ -44,35 +42,32 @@ def _get_function_template(function_name: str) -> dict:
 
 
 def _get_output_name(output_name: str) -> dict:
-    return {
-        output_name
-        + "_file": {"type": "File", "outputBinding": {"glob": output_name + ".pickle"}}
+    output_dict = {
+        "type": "File",
+        "outputBinding": {"glob": output_name + ".pickle"},
     }
+    return {output_name + "_file": output_dict}
 
 
 def _get_function(workflow):
     function_nodes_dict = {
         n["id"]: n["value"] for n in workflow[NODES_LABEL] if n["type"] == "function"
     }
     funct_dict = {}
-    for funct_id in function_nodes_dict.keys():
+    for funct_id in function_nodes_dict:
         target_ports = list(
-            set(
-                [
-                    e[TARGET_PORT_LABEL]
-                    for e in workflow[EDGES_LABEL]
-                    if e["target"] == funct_id
-                ]
-            )
+            {
+                e[TARGET_PORT_LABEL]
+                for e in workflow[EDGES_LABEL]
+                if e["target"] == funct_id
+            }
         )
         source_ports = list(
-            set(
-                [
-                    e[SOURCE_PORT_LABEL]
-                    for e in workflow[EDGES_LABEL]
-                    if e["source"] == funct_id
-                ]
-            )
+            {
+                e[SOURCE_PORT_LABEL]
+                for e in workflow[EDGES_LABEL]
+                if e["source"] == funct_id
+            }
         )
         funct_dict[funct_id] = {
             "targetPorts": target_ports,
@@ -86,7 +81,7 @@ def _write_function_cwl(workflow, directory_path: str = "."):
     export_path = Path(directory_path)
     export_path.mkdir(parents=True, exist_ok=True)
 
-    for i in function_nodes_dict.keys():
+    for i in function_nodes_dict:
         template: dict[str, Any] = {
             "cwlVersion": "v1.2",
             "class": "CommandLineTool",
@@ -142,10 +137,7 @@ def _write_workflow_config(workflow, directory_path: str = "."):
     export_path.mkdir(parents=True, exist_ok=True)
     with open(export_path / "workflow.yml", "w") as f:
         dump(
-            {
-                k + "_file": {"class": "File", "path": k + ".pickle"}
-                for k in input_dict.keys()
-            },
+            {k + "_file": {"class": "File", "path": k + ".pickle"} for k in input_dict},
             f,
             Dumper=Dumper,
         )
@@ -170,7 +162,7 @@ def _write_workflow(workflow, directory_path: str = "."):
     last_compute_id = [
         e[SOURCE_LABEL] for e in workflow[EDGES_LABEL] if e[TARGET_LABEL] == result_id
     ][0]
-    workflow_template["inputs"].update({k + "_file": "File" for k in input_dict.keys()})
+    workflow_template["inputs"].update({k + "_file": "File" for k in input_dict})
     if funct_dict[last_compute_id]["sourcePorts"] == [None]:
         workflow_template["outputs"] = {
             "result_file": {
@@ -209,29 +201,29 @@ def _write_workflow(workflow, directory_path: str = "."):
         for k, v in t[1].items():
             if v[SOURCE_LABEL] in input_id_dict:
                 in_dict[k + "_file"] = input_id_dict[v[SOURCE_LABEL]] + "_file"
+            elif v["sourcePort"] is None:
+                in_dict[k + "_file"] = (
+                    step_name_lst[v[SOURCE_LABEL]]
+                    + "_"
+                    + str(v[SOURCE_LABEL])
+                    + "/result_file"
+                )
             else:
-                if v["sourcePort"] is None:
-                    in_dict[k + "_file"] = (
-                        step_name_lst[v[SOURCE_LABEL]]
-                        + "_"
-                        + str(v[SOURCE_LABEL])
-                        + "/result_file"
-                    )
-                else:
-                    in_dict[k + "_file"] = (
-                        step_name_lst[v[SOURCE_LABEL]]
-                        + "_"
-                        + str(v[SOURCE_LABEL])
-                        + "/"
-                        + v[SOURCE_PORT_LABEL]
-                        + "_file"
-                    )
+                in_dict[k + "_file"] = (
+                    step_name_lst[v[SOURCE_LABEL]]
+                    + "_"
+                    + str(v[SOURCE_LABEL])
+                    + "/"
+                    + v[SOURCE_PORT_LABEL]
+                    + "_file"
+                )
+        step_dict = {
+            "run": node_script,
+            "in": in_dict,
+            "out": output,
+        }
         workflow_template["steps"].update(
-            {
-                step_name_lst[ind]
-                + "_"
-                + str(ind): {"run": node_script, "in": in_dict, "out": output}
-            }
+            {step_name_lst[ind] + "_" + str(ind): step_dict}
         )
     export_path = Path(directory_path)
     export_path.mkdir(parents=True, exist_ok=True)
@@ -240,7 +232,7 @@ def _write_workflow(workflow, directory_path: str = "."):
 
 
 def write_workflow(file_name: str, directory_path: str = "."):
-    with open(file_name, "r") as f:
+    with open(file_name) as f:
         workflow = json.load(f)
 
     _write_function_cwl(workflow=workflow, directory_path=directory_path)

src/python_workflow_definition/executorlib.py

@@ -11,10 +11,6 @@
     SOURCE_LABEL,
     SOURCE_PORT_LABEL,
     convert_nodes_list_to_dict,
-    get_dict,
-    get_kwargs,
-    get_list,
-    get_source_handles,
     remove_result,
 )
 
@@ -25,9 +21,9 @@ def get_item(obj, key):
 
 def _get_value(result_dict: dict, nodes_new_dict: dict, link_dict: dict, exe: Executor):
     source, source_handle = link_dict[SOURCE_LABEL], link_dict[SOURCE_PORT_LABEL]
-    if source in result_dict.keys():
+    if source in result_dict:
         result = result_dict[source]
-    elif source in nodes_new_dict.keys():
+    elif source in nodes_new_dict:
         result = nodes_new_dict[source]
     else:
         raise KeyError()

src/python_workflow_definition/jobflow.py

@@ -27,7 +27,7 @@
 
 
 def _get_function_dict(flow: Flow):
-    return {job.uuid: job.function for job in flow.jobs}
+    return {j.uuid: j.function for j in flow.jobs}
 
 
 def _get_nodes_dict(function_dict: dict):
@@ -62,8 +62,8 @@ def _get_edges_and_extend_nodes(
     flow_dict: dict, nodes_mapping_dict: dict, nodes_dict: dict
 ):
     edges_lst = []
-    for job in flow_dict["jobs"]:
-        for k, v in job["function_kwargs"].items():
+    for j in flow_dict["jobs"]:
+        for k, v in j["function_kwargs"].items():
             if (
                 isinstance(v, dict)
                 and "@module" in v
@@ -72,20 +72,18 @@ def _get_edges_and_extend_nodes(
             ):
                 edges_lst.append(
                     _get_edge_from_dict(
-                        target=nodes_mapping_dict[job["uuid"]],
+                        target=nodes_mapping_dict[j["uuid"]],
                         key=k,
                         value_dict=v,
                         nodes_mapping_dict=nodes_mapping_dict,
                     )
                 )
             elif isinstance(v, dict) and any(
-                [
-                    isinstance(el, dict)
-                    and "@module" in el
-                    and "@class" in el
-                    and "@version" in el
-                    for el in v.values()
-                ]
+                isinstance(el, dict)
+                and "@module" in el
+                and "@class" in el
+                and "@version" in el
+                for el in v.values()
             ):
                 node_dict_index = len(nodes_dict)
                 nodes_dict[node_dict_index] = get_dict
@@ -122,20 +120,18 @@ def _get_edges_and_extend_nodes(
                         )
                 edges_lst.append(
                     {
-                        TARGET_LABEL: nodes_mapping_dict[job["uuid"]],
+                        TARGET_LABEL: nodes_mapping_dict[j["uuid"]],
                         TARGET_PORT_LABEL: k,
                         SOURCE_LABEL: node_dict_index,
                         SOURCE_PORT_LABEL: None,
                     }
                 )
             elif isinstance(v, list) and any(
-                [
-                    isinstance(el, dict)
-                    and "@module" in el
-                    and "@class" in el
-                    and "@version" in el
-                    for el in v
-                ]
+                isinstance(el, dict)
+                and "@module" in el
+                and "@class" in el
+                and "@version" in el
+                for el in v
             ):
                 node_list_index = len(nodes_dict)
                 nodes_dict[node_list_index] = get_list
@@ -172,7 +168,7 @@ def _get_edges_and_extend_nodes(
                         )
                 edges_lst.append(
                     {
-                        TARGET_LABEL: nodes_mapping_dict[job["uuid"]],
+                        TARGET_LABEL: nodes_mapping_dict[j["uuid"]],
                         TARGET_PORT_LABEL: k,
                         SOURCE_LABEL: node_list_index,
                         SOURCE_PORT_LABEL: None,
@@ -186,7 +182,7 @@ def _get_edges_and_extend_nodes(
                     node_index = {tv: tk for tk, tv in nodes_dict.items()}[v]
                 edges_lst.append(
                     {
-                        TARGET_LABEL: nodes_mapping_dict[job["uuid"]],
+                        TARGET_LABEL: nodes_mapping_dict[j["uuid"]],
                         TARGET_PORT_LABEL: k,
                         SOURCE_LABEL: node_index,
                         SOURCE_PORT_LABEL: None,
@@ -196,10 +192,8 @@ def _get_edges_and_extend_nodes(
 
 
 def _resort_total_lst(total_dict: dict, nodes_dict: dict) -> dict:
-    nodes_with_dep_lst = list(sorted(total_dict.keys()))
-    nodes_without_dep_lst = [
-        k for k in nodes_dict.keys() if k not in nodes_with_dep_lst
-    ]
+    nodes_with_dep_lst = sorted(total_dict.keys())
+    nodes_without_dep_lst = [k for k in nodes_dict if k not in nodes_with_dep_lst]
     ordered_lst: list = []
     total_new_dict: dict[Any, dict] = {}
     while len(total_new_dict) < len(total_dict):
@@ -208,7 +202,7 @@ def _resort_total_lst(total_dict: dict, nodes_dict: dict) -> dict:
             if ind not in ordered_lst:
                 source_lst = [sd[SOURCE_LABEL] for sd in connect.values()]
                 if all(
-                    [s in ordered_lst or s in nodes_without_dep_lst for s in source_lst]
+                    s in ordered_lst or s in nodes_without_dep_lst for s in source_lst
                 ):
                     ordered_lst.append(ind)
                     total_new_dict[ind] = connect
@@ -220,7 +214,7 @@ def _group_edges(edges_lst: list) -> dict:
     for ed_major in edges_lst:
         target_id = ed_major[TARGET_LABEL]
         tmp_lst = []
-        if target_id not in total_dict.keys():
+        if target_id not in total_dict:
             for ed in edges_lst:
                 if target_id == ed[TARGET_LABEL]:
                     tmp_lst.append(ed)
@@ -237,15 +231,15 @@ def _get_workflow(
 ) -> list:
     def get_attr_helper(obj, source_handle):
         if source_handle is None:
-            return getattr(obj, "output")
+            return obj.output
         else:
-            return getattr(getattr(obj, "output"), source_handle)
+            return getattr(obj.output, source_handle)
 
     memory_dict: dict[Any, Any] = {}
-    for k in total_dict.keys():
+    for k, subdict in total_dict.items():
         v = nodes_dict[k]
         if isfunction(v):
-            if k in source_handles_dict.keys():
+            if k in source_handles_dict:
                 fn = job(
                     method=v,
                     data=[el for el in source_handles_dict[k] if el is not None],
@@ -261,7 +255,7 @@ def get_attr_helper(obj, source_handle):
                         source_handle=vw[SOURCE_PORT_LABEL],
                     )
                 )
-                for kw, vw in total_dict[k].items()
+                for kw, vw in subdict.items()
             }
             memory_dict[k] = fn(**kwargs)
     return list(memory_dict.values())