Fix: Improve error handling in workflow compilation when output binding fails

flytekit/core/workflow.py

@@ -746,14 +746,19 @@ def compile(self, **kwargs):
                     )
                 workflow_outputs = workflow_outputs[0]
             t = self.python_interface.outputs[output_names[0]]
-            b, _ = binding_from_python_std(
-                ctx,
-                output_names[0],
-                self.interface.outputs[output_names[0]].type,
-                workflow_outputs,
-                t,
-            )
-            bindings.append(b)
+            try:
+                b, _ = binding_from_python_std(
+                    ctx,
+                    output_names[0],
+                    self.interface.outputs[output_names[0]].type,
+                    workflow_outputs,
+                    t,
+                )
+                bindings.append(b)
+            except Exception as e:
+                raise FlyteValidationException(
+                    f"Failed to bind output {output_names[0]} for function {self.name}: {e}"
+                ) from e
         elif len(output_names) > 1:
             if not isinstance(workflow_outputs, tuple):
                 raise AssertionError("The Workflow specification indicates multiple return values, received only one")
@@ -763,14 +768,17 @@ def compile(self, **kwargs):
                 if isinstance(workflow_outputs[i], ConditionalSection):
                     raise AssertionError("A Conditional block (if-else) should always end with an `else_()` clause")
                 t = self.python_interface.outputs[out]
-                b, _ = binding_from_python_std(
-                    ctx,
-                    out,
-                    self.interface.outputs[out].type,
-                    workflow_outputs[i],
-                    t,
-                )
-                bindings.append(b)
+                try:
+                    b, _ = binding_from_python_std(
+                        ctx,
+                        out,
+                        self.interface.outputs[out].type,
+                        workflow_outputs[i],
+                        t,
+                    )
+                    bindings.append(b)
+                except Exception as e:
+                    raise FlyteValidationException(f"Failed to bind output {out} for function {self.name}: {e}") from e
 
         # Save all the things necessary to create an WorkflowTemplate, except for the missing project and domain
         self._nodes = all_nodes

tests/flytekit/unit/core/test_type_hints.py

@@ -34,6 +34,7 @@
 from flytekit.core.testing import patch, task_mock
 from flytekit.core.type_engine import RestrictedTypeError, SimpleTransformer, TypeEngine
 from flytekit.core.workflow import workflow
+from flytekit.exceptions.user import FlyteValidationException
 from flytekit.models import literals as _literal_models
 from flytekit.models.core import types as _core_types
 from flytekit.models.interface import Parameter
@@ -133,6 +134,15 @@ def my_task() -> str:
     assert context_manager.FlyteContextManager.size() == 1
 
 
+def test_missing_output():
+    @workflow
+    def wf() -> str:
+        return None  # type: ignore
+
+    with pytest.raises(FlyteValidationException, match="Failed to bind output"):
+        wf.compile()
+
+
 def test_engine_file_output():
     basic_blob_type = _core_types.BlobType(
         format="",
@@ -794,7 +804,7 @@ def t1(a: int) -> typing.NamedTuple("OutputsBC", t1_int_output=int, c=str):
     def t2(a: str) -> str:
         return a
 
-    with pytest.raises(TypeError):
+    with pytest.raises(FlyteValidationException):
 
         @workflow
         def my_wf(a: int, b: str) -> (int, str):

tests/flytekit/unit/experimental/test_eager_workflows.py

@@ -9,7 +9,7 @@
 from hypothesis import given, settings
 
 from flytekit import dynamic, task, workflow
-from flytekit.core.type_engine import TypeTransformerFailedError
+from flytekit.exceptions.user import FlyteValidationException
 from flytekit.experimental import EagerException, eager
 from flytekit.types.directory import FlyteDirectory
 from flytekit.types.file import FlyteFile
@@ -213,7 +213,7 @@ async def eager_wf(x: int) -> int:
         out = await local_wf(x=x)
         return await double(x=out)
 
-    with pytest.raises(TypeTransformerFailedError):
+    with pytest.raises(FlyteValidationException):
         asyncio.run(eager_wf(x=x_input))