Add kwargs support for workflow invocations

src/ert/config/ert_script.py

@@ -115,6 +115,7 @@ def initializeAndRun(
         argument_types: list[type[Any]],
         argument_values: list[str],
         fixtures: dict[str, Any] | None = None,
+        kwargs: dict[str, Any] | None = None,
     ) -> Any:
         fixtures = {} if fixtures is None else fixtures
         arguments = []
@@ -128,11 +129,21 @@ def initializeAndRun(
         fixtures["workflow_args"] = arguments
         try:
             func_args = inspect.signature(self.run).parameters
+            kwargs_defaults = {
+                k: v.default
+                for k, v in func_args.items()
+                if k not in fixtures
+                and v.kind == v.VAR_KEYWORD
+                and not str(v).startswith("*")
+            }
+            use_kwargs = {
+                k: (kwargs or {}).get(k, v) for k, v in kwargs_defaults.items()
+            }
             # If the user has specified *args, we skip injecting fixtures, and just
             # pass the user configured arguments
             if not any(p.kind == p.VAR_POSITIONAL for p in func_args.values()):
                 try:
-                    arguments = self.insert_fixtures(func_args, fixtures)
+                    arguments = self.insert_fixtures(func_args, fixtures, use_kwargs)
                 except ValueError as e:
                     # This is here for backwards compatibility, the user does not have *argv
                     # but positional arguments. Can not be mixed with using fixtures.
@@ -143,7 +154,14 @@ def initializeAndRun(
             self._ert = fixtures.get("ert_config")
             self._ensemble = fixtures.get("ensemble")
             self._storage = fixtures.get("storage")
-            return self.run(*arguments)
+            if not arguments and not use_kwargs:
+                return self.run()
+            elif arguments and not use_kwargs:
+                return self.run(*arguments)
+            elif not arguments and use_kwargs:
+                return self.run(**use_kwargs)
+            else:
+                return self.run(*arguments, **use_kwargs)
         except AttributeError as e:
             error_msg = str(e)
             if not hasattr(self, "run"):
@@ -171,13 +189,14 @@ def insert_fixtures(
         self,
         func_args: MappingProxyType[str, inspect.Parameter],
         fixtures: dict[str, Fixtures],
+        kwargs: dict[str, Any],
     ) -> list[Any]:
         arguments = []
         errors = []
         for val in func_args:
             if val in fixtures:
                 arguments.append(fixtures[val])
-            else:
+            elif val not in kwargs:
                 errors.append(val)
         if errors:
             raise ValueError(

src/ert/gui/tools/plugins/plugin.py

@@ -42,7 +42,7 @@ def getArguments(self, fixtures: dict[str, Any]) -> list[Any]:
         script = self.__loadPlugin()
         fixtures["parent"] = self.__parent_window
         func_args = inspect.signature(script.getArguments).parameters
-        arguments = script.insert_fixtures(func_args, fixtures)
+        arguments = script.insert_fixtures(func_args, fixtures, {})
 
         # Part of deprecation
         script._ert = fixtures.get("ert_config")

src/ert/libres_facade.py

@@ -210,6 +210,7 @@ def run_ertscript(  # type: ignore
         storage: Storage,
         ensemble: Ensemble,
         *args: Any,
+        **kwargs: dict[str, Any],
     ) -> Any:
         warnings.warn(
             "run_ertscript is deprecated, use the workflow runner",
@@ -219,6 +220,7 @@ def run_ertscript(  # type: ignore
         return ertscript().initializeAndRun(
             [],
             argument_values=args,
+            kwargs=kwargs,
             fixtures={
                 "ert_config": self.config,
                 "ensemble": ensemble,

src/ert/workflow_runner.py

@@ -57,7 +57,7 @@ def run(
         else:
             raise UserWarning("Unknown script type!")
         result = self.__script.initializeAndRun(  # type: ignore
-            self.job.argument_types(), arguments, fixtures=fixtures
+            self.job.argument_types(), arguments, fixtures
         )
         self.__running = False