support FlyteRemote.execute with no inputs (#3266)

Signed-off-by: Niels Bantilan <niels.bantilan@gmail.com>

Author: cosmicBboy

flytekit/remote/remote.py

@@ -1506,7 +1506,7 @@ def register_launch_plan(
     def _execute(
         self,
         entity: typing.Union[FlyteTask, FlyteWorkflow, FlyteLaunchPlan],
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         project: str = None,
         domain: str = None,
         execution_name: typing.Optional[str] = None,
@@ -1541,6 +1541,9 @@ def _execute(
         :param execution_cluster_label: Specify label of cluster(s) on which newly created execution should be placed.
         :returns: :class:`~flytekit.remote.workflow_execution.FlyteWorkflowExecution`
         """
+        if inputs is None:
+            inputs = {}
+
         if execution_name is not None and execution_name_prefix is not None:
             raise ValueError("Only one of execution_name and execution_name_prefix can be set, but got both set")
         # todo: The prefix should be passed to the backend
@@ -1675,7 +1678,7 @@ def execute(
         entity: typing.Union[
             FlyteTask, FlyteLaunchPlan, FlyteWorkflow, PythonTask, WorkflowBase, LaunchPlan, ReferenceEntity
         ],
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         project: str = None,
         domain: str = None,
         name: str = None,
@@ -1898,7 +1901,7 @@ def execute(
     def execute_remote_task_lp(
         self,
         entity: typing.Union[FlyteTask, FlyteLaunchPlan],
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         project: str = None,
         domain: str = None,
         execution_name: typing.Optional[str] = None,
@@ -1938,7 +1941,7 @@ def execute_remote_task_lp(
     def execute_remote_wf(
         self,
         entity: FlyteWorkflow,
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         project: str = None,
         domain: str = None,
         execution_name: typing.Optional[str] = None,
@@ -1981,7 +1984,7 @@ def execute_remote_wf(
     def execute_reference_task(
         self,
         entity: ReferenceTask,
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         execution_name: typing.Optional[str] = None,
         execution_name_prefix: typing.Optional[str] = None,
         options: typing.Optional[Options] = None,
@@ -2030,7 +2033,7 @@ def execute_reference_task(
     def execute_reference_workflow(
         self,
         entity: ReferenceWorkflow,
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         execution_name: typing.Optional[str] = None,
         execution_name_prefix: typing.Optional[str] = None,
         options: typing.Optional[Options] = None,
@@ -2093,7 +2096,7 @@ def execute_reference_workflow(
     def execute_reference_launch_plan(
         self,
         entity: ReferenceLaunchPlan,
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         execution_name: typing.Optional[str] = None,
         execution_name_prefix: typing.Optional[str] = None,
         options: typing.Optional[Options] = None,
@@ -2145,7 +2148,7 @@ def execute_reference_launch_plan(
     def execute_local_task(
         self,
         entity: PythonTask,
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         project: str = None,
         domain: str = None,
         name: str = None,
@@ -2227,7 +2230,7 @@ def execute_local_task(
     def execute_local_workflow(
         self,
         entity: WorkflowBase,
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
         project: str = None,
         domain: str = None,
         name: str = None,
@@ -2337,7 +2340,8 @@ def execute_local_workflow(
     def execute_local_launch_plan(
         self,
         entity: LaunchPlan,
-        inputs: typing.Dict[str, typing.Any],
+        inputs: typing.Optional[typing.Dict[str, typing.Any]] = None,
+        *,
         version: str,
         project: typing.Optional[str] = None,
         domain: typing.Optional[str] = None,

tests/flytekit/unit/remote/test_remote.py

@@ -33,7 +33,7 @@
 from flytekit.models.core.identifier import Identifier, ResourceType, WorkflowExecutionIdentifier
 from flytekit.models.execution import Execution
 from flytekit.models.task import Task
-from flytekit.remote import FlyteTask, FlyteWorkflow
+from flytekit.remote import FlyteTask, FlyteWorkflow, FlyteWorkflowExecution
 from flytekit.remote.lazy_entity import LazyEntity
 from flytekit.remote.remote import FlyteRemote, _get_git_repo_url, _get_pickled_target_dict
 from flytekit.tools.translator import Options, get_serializable, get_serializable_launch_plan
@@ -209,7 +209,7 @@ def local_assertions(*args, **kwargs):
     )
 
 
-def test_execute_with_wrong_input_key(remote, mock_wf_exec):
+def test_execute_with_wrong_input_key(remote: FlyteRemote, mock_wf_exec):
     # mock_url.get.return_value = "localhost"
     # mock_insecure.get.return_value = True
     mock_wf_exec.return_value = True
@@ -228,6 +228,16 @@ def test_execute_with_wrong_input_key(remote, mock_wf_exec):
         )
 
 
+def test_execute_with_no_inputs(remote: FlyteRemote, mock_wf_exec):
+    mock_wf_exec.return_value = True
+    mock_client = MagicMock()
+    remote._client = mock_client
+
+    mock_entity = MagicMock()
+    out = remote._execute(mock_entity, project="proj", domain="dev")
+    assert isinstance(out, FlyteWorkflowExecution)
+
+
 def test_form_config():
     remote = FlyteRemote(config=Config.auto(), default_project="p1", default_domain="d1")
     assert remote.default_project == "p1"