Local dynamic execution was missing (#569)

Signed-off-by: wild-endeavor <[email protected]>

flytekit/core/python_function_task.py

@@ -291,3 +291,8 @@ def dynamic_execute(self, task_function: Callable, **kwargs) -> Any:
                 )
 
             return self.compile_into_workflow(ctx, task_function, **kwargs)
+
+        if ctx.execution_state and ctx.execution_state.mode == ExecutionState.Mode.LOCAL_TASK_EXECUTION:
+            return exception_scopes.user_entry_point(task_function)(**kwargs)
+
+        raise ValueError(f"Invalid execution provided, execution state: {ctx.execution_state}")

tests/flytekit/unit/core/test_dynamic.py

@@ -60,3 +60,49 @@ def my_wf(a: int) -> typing.List[str]:
             )
 
     assert context_manager.FlyteContextManager.size() == 1
+
+
+def test_dynamic_local():
+    @task
+    def t1(a: int) -> str:
+        a = a + 2
+        return "fast-" + str(a)
+
+    @dynamic
+    def ranged_int_to_str(a: int) -> typing.List[str]:
+        s = []
+        for i in range(a):
+            s.append(t1(a=i))
+        return s
+
+    res = ranged_int_to_str(a=5)
+    assert res == ["fast-2", "fast-3", "fast-4", "fast-5", "fast-6"]
+
+
+def test_nested_dynamic_local():
+    @task
+    def t1(a: int) -> str:
+        a = a + 2
+        return "fast-" + str(a)
+
+    @dynamic
+    def ranged_int_to_str(a: int) -> typing.List[str]:
+        s = []
+        for i in range(a):
+            s.append(t1(a=i))
+        return s
+
+    @dynamic
+    def add_and_range(a: int, b: int) -> typing.List[str]:
+        x = a + b
+        return ranged_int_to_str(a=x)
+
+    res = add_and_range(a=2, b=3)
+    assert res == ["fast-2", "fast-3", "fast-4", "fast-5", "fast-6"]
+
+    @workflow
+    def wf(a: int, b: int) -> typing.List[str]:
+        return add_and_range(a=a, b=b)
+
+    res = wf(a=2, b=3)
+    assert res == ["fast-2", "fast-3", "fast-4", "fast-5", "fast-6"]