Decorator allow users to define the inputs manually for dynamic input (#281)

Author: superstar54

aiida_workgraph/decorator.py

@@ -207,6 +207,7 @@ def build_task_from_AiiDA(
     spec = executor.spec()
     args = []
     kwargs = []
+    user_defined_input_names = [input["name"] for input in inputs]
     for _key, port in spec.inputs.ports.items():
         add_input_recursive(inputs, port, args, kwargs, required=port.required)
     for _key, port in spec.outputs.ports.items():
@@ -230,6 +231,14 @@ def build_task_from_AiiDA(
                     "property": {"identifier": "workgraph.any", "default": {}},
                 }
             )
+        # When the input is dyanmic, if user defines some input names does not included in the args and kwargs,
+        # which means the user define the input names manually, we must add them to the kwargs
+        for key in user_defined_input_names:
+            if key not in args and key not in kwargs:
+                if key == name:
+                    continue
+                kwargs.append(key)
+
     # TODO In order to reload the WorkGraph from process, "is_pickle" should be True
     # so I pickled the function here, but this is not necessary
     # we need to update the node_graph to support the path and name of the function

tests/test_calcfunction.py

@@ -1,5 +1,6 @@
 import pytest
-from aiida_workgraph import WorkGraph
+from aiida_workgraph import WorkGraph, task
+from aiida import orm
 
 
 def test_run(wg_calcfunction: WorkGraph) -> None:
@@ -14,10 +15,16 @@ def test_run(wg_calcfunction: WorkGraph) -> None:
 
 
 @pytest.mark.usefixtures("started_daemon_client")
-def test_submit(wg_calcfunction: WorkGraph) -> None:
-    """Submit simple calcfunction."""
-    wg = wg_calcfunction
-    wg.name = "test_submit_calcfunction"
-    wg.submit(wait=True)
-    # print("results: ", results[])
-    assert wg.tasks["sumdiff2"].outputs["sum"].value == 9
+def test_dynamic_inputs() -> None:
+    """Test dynamic inputs.
+    For dynamic inputs, we allow the user to define the inputs manually.
+    """
+
+    @task.calcfunction(inputs=[{"name": "x"}, {"name": "y"}])
+    def add(**kwargs):
+        return kwargs["x"] + kwargs["y"]
+
+    wg = WorkGraph("test_dynamic_inputs")
+    wg.add_task(add, name="add1", x=orm.Int(1), y=orm.Int(2))
+    wg.run()
+    assert wg.tasks["add1"].outputs["result"].value == 3