Merge branch 'main' into add_namespace_list

.github/workflows/ci.yaml

@@ -99,7 +99,7 @@ jobs:
             env:
                 AIIDA_WARN_v3: 1
             run: |
-                pytest -v tests --cov
+                pytest -v tests --cov --durations=0
 
         -   name: Upload coverage reports to Codecov
             uses: codecov/[email protected]

aiida_workgraph/__init__.py

@@ -9,6 +9,6 @@
 from .task import Task
 from .decorator import task, build_task
 
-__version__ = "0.3.14"
+__version__ = "0.3.15"
 
 __all__ = ["WorkGraph", "Task", "task", "build_task"]

aiida_workgraph/calculations/python.py

@@ -50,6 +50,9 @@ def define(cls, spec: CalcJobProcessSpec) -> None:  # type: ignore[override]
         spec.input(
             "function_name", valid_type=Str, serializer=to_aiida_type, required=False
         )
+        spec.input(
+            "process_label", valid_type=Str, serializer=to_aiida_type, required=False
+        )
         spec.input_namespace(
             "function_kwargs", valid_type=Data, required=False
         )  # , serializer=serialize_to_aiida_nodes)
@@ -137,7 +140,16 @@ def _build_process_label(self) -> str:
 
         :returns: The process label to use for ``ProcessNode`` instances.
         """
-        return f"PythonJob<{self.inputs.function_name.value}>"
+        if self.inputs.process_label:
+            return self.inputs.process_label.value
+        else:
+            return f"PythonJob<{self.inputs.function_name.value}>"
+
+    def on_create(self) -> None:
+        """Called when a Process is created."""
+
+        super().on_create()
+        self.node.label = self.inputs.process_label.value
 
     def prepare_for_submission(self, folder: Folder) -> CalcInfo:
         """Prepare the calculation for submission.
@@ -265,9 +277,10 @@ def prepare_for_submission(self, folder: Folder) -> CalcInfo:
         dirpath = pathlib.Path(folder._abspath)
         with folder.open(filename, "wb") as handle:
             pickle.dump(input_values, handle)
-            # create a singlefiledata object for the pickled data
-            file_data = SinglefileData(file=f"{dirpath}/{filename}")
-            local_copy_list.append((file_data.uuid, file_data.filename, filename))
+        # create a singlefiledata object for the pickled data
+        file_data = SinglefileData(file=f"{dirpath}/{filename}")
+        file_data.store()
+        local_copy_list.append((file_data.uuid, file_data.filename, filename))
 
         codeinfo = CodeInfo()
         codeinfo.stdin_name = self.options.input_filename

aiida_workgraph/calculations/python_parser.py

@@ -11,7 +11,7 @@ def parse(self, **kwargs):
 
         The outputs could be a namespce, e.g.,
         outputs=[
-            {"identifier": "Namespace", "name": "add_multiply"},
+            {"identifier": "workgraph.namespace", "name": "add_multiply"},
             {"name": "add_multiply.add"},
             {"name": "add_multiply.multiply"},
             {"name": "minus"},
@@ -100,15 +100,16 @@ def serialize_output(self, result, output):
         """Serialize outputs."""
 
         name = output["name"]
-        if output["identifier"].upper() == "NAMESPACE":
+        if output["identifier"].upper() == "WORKGRAPH.NAMESPACE":
             if isinstance(result, dict):
                 serialized_result = {}
                 for key, value in result.items():
                     full_name = f"{name}.{key}"
                     full_name_output = self.find_output(full_name)
                     if (
                         full_name_output
-                        and full_name_output["identifier"].upper() == "NAMESPACE"
+                        and full_name_output["identifier"].upper()
+                        == "WORKGRAPH.NAMESPACE"
                     ):
                         serialized_result[key] = self.serialize_output(
                             value, full_name_output

aiida_workgraph/collection.py

@@ -38,6 +38,9 @@ def new(
             # make links between the tasks
             task.set(links)
             return task
+        if isinstance(identifier, str) and identifier.upper() == "WHILE":
+            task = super().new("workgraph.while", name, uuid, **kwargs)
+            return task
         if isinstance(identifier, WorkGraph):
             identifier = build_task_from_workgraph(identifier)
         return super().new(identifier, name, uuid, **kwargs)
@@ -86,6 +89,7 @@ def new(
 
         # build the socket on the fly if the identifier is a callable
         if callable(identifier):
+            print("identifier is callable", identifier)
             identifier = build_socket_from_AiiDA(identifier)
         # Call the original new method
         return super().new(identifier, name, **kwargs)