<#3334 Improve error message for type engine >
Pul...
# flyte-githuba
acoustic-carpenter-78188
10/03/2025, 12:24 AM#3334 Improve error message for type engine
Pull request opened by pingsutw
## Tracking issue
NA
## Why are the changes needed?
Improve the error message so people can debug much more easily.
## What changes were proposed in this pull request?
• Update repr for LiteralType
• Update the error message for wrong inputs/outputs/bindings
## How was this patch tested?
from datetime import datetime, date
from enum import StrEnum
from typing import Any
from click.testing import CliRunner
from flytekit import task, workflow, ImageSpec, dynamic
from flytekit.clis.sdk_in_container import pyflyte
new_flytekit = "git+https://github.com/flyteorg/flytekit.git@d341be82f4291e48fc0ccd2e53a59bcd22273d4f"
image = ImageSpec(
# registry="ghcr.io/flyteorg",
packages=[new_flytekit, "gcsfs==2025.7.0", "fsspec==2025.7.0"],
apt_packages=["git"],
builder="union"
)
class ArtifactRelType(StrEnum):
DOWNSTREAM_OF = "downstream_of"
RELATED_TO = "related_to"
FlytePropertyValue = float | None | str | bool | datetime | date | list[str]
@task(container_image=image)
def say_hello(
my_val: dict[str, float | int],
artifact_rel_type: dict[str, ArtifactRelType] | None = None,
additional_properties: dict[str, FlytePropertyValue] | None = None,
) -> dict[str, float | int]:
print(my_val)
return my_val
@task(container_image=image)
def gen_outputs(
) -> dict[str, FlytePropertyValue] | None:
return [{"c": ["hello"], "algorithm": "xgboost"}]
@task(container_image=image)
def gen_outputs1(
) -> list[dict[str, FlytePropertyValue]] | dict[str, FlytePropertyValue] | None:
return [{"c": ["hello"], "algorithm": "xgboost"}]
@workflow
def wf():
# Wrong outputs
# union run --remote flyte-example/fix_type.py wf
say_hello(
my_val={"a": 1, "b": 2},
artifact_rel_type={"a": ArtifactRelType.DOWNSTREAM_OF},
additional_properties=gen_outputs(),
)
@workflow
def wf1():
# Wrong binding
# union run --remote flyte-example/fix_type.py wf1
say_hello(
my_val={"a": 1, "b": 2},
artifact_rel_type={"a": ArtifactRelType.DOWNSTREAM_OF},
additional_properties=[{"c": ["hello"], "algorithm": "xgboost"}],
)
@workflow()
def wf2(additional_properties: dict[str, FlytePropertyValue] | None = [{"c": ["hello"], "algorithm": "xgboost"}]):
# Wrong default inputs
# union run --remote flyte-example/fix_type.py wf2
say_hello(
my_val={"a": 1, "b": 2},
artifact_rel_type={"a": ArtifactRelType.DOWNSTREAM_OF},
additional_properties=additional_properties,
)
@workflow()
def wf3():
# Wrong inputs
# union run flyte-example/fix_type.py wf3
res = gen_outputs1()
say_hello(
my_val={"a": 1, "b": 2},
artifact_rel_type={"a": ArtifactRelType.DOWNSTREAM_OF},
additional_properties=res,
)
### Setup process
### Screenshots
Before:
Copy code
TypeTransformerFailedError: Error encountered while converting inputs of 'fix_type.say_hello':
Error converting input 'additional_properties' at position 2:
Literal value: Flyte Serialized object (Literal):
scalar:
union:
value:
collection:
literals: [map { literals { key: "c" value { scalar { union { value { collection { [...]
type:
collection_type:
map_value_type:
union_type:
variants: [simple: FLOAT structure { tag: "float" } , simple: NONE structure { tag: [...]
structure:
tag: Typed List
Expected Python type: dict[str, float | None | str | bool | datetime.datetime | datetime.date | list[str]] | None
Exception: Cannot convert from Flyte Serialized object (Literal):
scalar:
union:
value:
collection:
literals: [map { literals { key: "c" value { scalar { union { value { collection { [...]
type:
collection_type:
map_value_type:
union_type:
variants: [simple: FLOAT structure { tag: "float" } , simple: NONE structure { tag: [...]
structure:
tag: Typed List to dict[str, float | None | str | bool | datetime.datetime | datetime.date | list[str]] | None (using tag Typed List)acoustic-carpenter-78188
10/03/2025, 7:24 PM