#243 Improve error message for type engine Pull request opened by pingsutw Before:

Error converting: Var:additional_properties, type:<class '_asyncio.Task'>, into:dict[str, float | None | str | bool | datetime.datetime | datetime.date | list[str]] | None, received_value <Task finished name='Task-36' coro=<TypeEngine.to_literal() done, defined at /opt/venv/lib/python3.12/site-packages/flyte/types/_type_engine.py:1082> exception=TypeTransformerFailedError("Value [{'c': ['hello'], 'algorithm': 'xgboost'}] is not of type dict[str, float | None | str | bool | datetime.datetime | datetime.date | list[str]] | None")>

[Screenshot 2025-10-03 at 4 31 46 PM](https://private-user-images.githubusercontent.com/37936015/497364115-272f3f67-09bf-4fb0-9338-01e8fe75b04e.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NTk3NzY2NjUsIm5iZiI6MTc1OTc3NjM2NSwicGF0aCI6Ii8zNzkzNjAxNS80OTczNjQxMTUtMjcyZjNmNjctMDliZi00ZmIwLTkzMzgtMDFlOGZlNzViMDRlLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNTEwMDYlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjUxMDA2VDE4NDYwNVomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTFiZmQ2MTgyMzk3MzE5NTM0ODQxNzBkYjJkMmRiNjg0YmQ0MTU2YzQ4OGQ5NmRiOGJmZGJmYTZhZWJiNGQyOWQmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0In0.Be2ETtChy_rt8Z8FKDzqHYtBM5JqTj7jiHenWWBdy-c) After:

Error converting: Var:additional_properties, type:<class 'list'>, into:dict[str, float | None | str | bool | datetime.datetime | datetime.date | list[str]] | None, received_value [{'algorithm': 'xgboost', 'c': ['hello']}]

[Screenshot 2025-10-03 at 4 33 08 PM](https://private-user-images.githubusercontent.com/37936015/497364255-193bcd96-225e-4f8d-b8d4-196d1cd1f226.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NTk3NzY2NjUsIm5iZiI6MTc1OTc3NjM2NSwicGF0aCI6Ii8zNzkzNjAxNS80OTczNjQyNTUtMTkzYmNkOTYtMjI1ZS00ZjhkLWI4ZDQtMTk2ZDFjZDFmMjI2LnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNTEwMDYlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjUxMDA2VDE4NDYwNVomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPWM4MmIxODhiMjZjNzJjOWMzYmNkN2RlYWNhOTQ3NDBlYWRjNTM4YjBhNWQ1YjU5ZGJhMmQ1MWZkY2U0ZTcyZDImWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0In0.hdmcGwlS5tJ9KjFhuWZdVeh9dPUStvAV2xFFQ4CDpqs) from datetime import datetime, date from enum import StrEnum from pathlib import Path import flyte image = ( flyte.Image.from_debian_base() .with_apt_packages("git") ) task_env = flyte.TaskEnvironment( name="hello_type", resources=flyte.Resources(cpu=(1, 2), memory=("400Mi", "1000Mi")), image=image ) class ArtifactRelType(StrEnum): DOWNSTREAM_OF = "downstream_of" RELATED_TO = "related_to" FlytePropertyValue = float | None | str | bool | datetime | date | list[str] @task_env.task 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_env.task def gen_outputs( ) -> dict[str, FlytePropertyValue] | None: return [{"c": ["hello"], "algorithm": "xgboost"}] @task_env.task def gen_outputs1( ) -> list[dict[str, FlytePropertyValue]] | dict[str, FlytePropertyValue] | None: return [{"c": ["hello"], "algorithm": "xgboost"}] @task_env.task def wf(): # Wrong outputs say_hello( my_val={"a": 1, "b": 2}, artifact_rel_type={"a": ArtifactRelType.DOWNSTREAM_OF}, additional_properties=gen_outputs(), ) @task_env.task def wf1(): # Wrong binding say_hello( my_val={"a": 1, "b": 2}, artifact_rel_type={"a": ArtifactRelType.DOWNSTREAM_OF}, additional_properties=[{"c": ["hello"], "algorithm": "xgboost"}], ) @task_env.task def wf2(additional_properties: dict[str, FlytePropertyValue] | None = [{"c": ["hello"], "algorithm": "xgboost"}]): # Wrong default inputs say_hello( my_val={"a": 1, "b": 2}, artifact_rel_type={"a": ArtifactRelType.DOWNSTREAM_OF}, additional_properties=additional_properties, ) @task_env.task def wf3(): # Wrong inputs res = gen_outputs1() say_hello( my_val={"a": 1, "b": 2}, artifact_rel_type={"a": ArtifactRelType.DOWNSTREAM_OF}, additional_properties=res, ) if name == "__main__": flyte.init_from_config() run = flyte.with_runcontext(env_vars={"LOG_LEVEL": "10"}).run(wf3) print("run name:", run.name) print("run url:", run.url) flyteorg/flyte-sdk