flytekit.types.error.error
0.1.dev2192+g7c539c3.d20250403
flytekit.types.error.error
Class
Description
ErrorTransformer
Enables converting a python type FlyteError to LiteralType.
FlyteError
Special Task type that will be used in the failure node.
Property
Type
Description
T
TypeVar
Enables converting a python type FlyteError to LiteralType.Error
def assert_type (
t: Type[T],
v: T,
)
Parameter
Type
t
Type[T]
v
T
def from_binary_idl (
binary_idl_object: Binary,
expected_python_type: Type[T],
) -> Optional[T]
This function primarily handles deserialization for untyped dicts, dataclasses, Pydantic BaseModels, and attribute access.`
For untyped dict, dataclass, and pydantic basemodel:
Life Cycle (Untyped Dict as example):
python val -> msgpack bytes -> binary literal scalar -> msgpack bytes -> python val
(to_literal) (from_binary_idl)
For attribute access:
Life Cycle:
python val -> msgpack bytes -> binary literal scalar -> resolved golang value -> binary literal scalar -> msgpack bytes -> python val
(to_literal) (propeller attribute access) (from_binary_idl)
Parameter
Type
binary_idl_object
Binary
expected_python_type
Type[T]
def from_generic_idl (
generic: Struct,
expected_python_type: Type[T],
) -> Optional[T]
TODO: Support all Flyte Types.
This is for dataclass attribute access from input created from the Flyte Console.
Note:
This can be removed in the future when the Flyte Console support generate Binary IDL Scalar as input.
Parameter
Type
generic
Struct
expected_python_type
Type[T]
def get_literal_type (
t: typing. Type[~ T],
) -> flytekit. models. types. LiteralType
Converts the python type to a Flyte LiteralType
Parameter
Type
t
typing.Type[~T]
def guess_python_type (
literal_type: flytekit. models. types. LiteralType,
) -> typing. Type[flytekit. types. error. error. FlyteError]
Converts the Flyte LiteralType to a python object type.
Parameter
Type
literal_type
flytekit.models.types.LiteralType
def isinstance_generic (
obj,
generic_alias,
)
Parameter
Type
obj
generic_alias
def to_html (
ctx: FlyteContext,
python_val: T,
expected_python_type: Type[T],
) -> str
Converts any python val (dataframe, int, float) to a html string, and it will be wrapped in the HTML div
Parameter
Type
ctx
FlyteContext
python_val
T
expected_python_type
Type[T]
def to_literal (
ctx: flytekit. core. context_manager. FlyteContext,
python_val: flytekit. types. error. error. FlyteError,
python_type: typing. Type[~ T],
expected: flytekit. models. types. LiteralType,
) -> flytekit. models. literals. Literal
Converts a given python_val to a Flyte Literal, assuming the given python_val matches the declared python_type.
Implementers should refrain from using type(python_val) instead rely on the passed in python_type. If these
do not match (or are not allowed) the Transformer implementer should raise an AssertionError, clearly stating
what was the mismatch
Parameter
Type
ctx
flytekit.core.context_manager.FlyteContext
python_val
flytekit.types.error.error.FlyteError
python_type
typing.Type[~T]
expected
flytekit.models.types.LiteralType
def to_python_value (
ctx: flytekit. core. context_manager. FlyteContext,
lv: flytekit. models. literals. Literal,
expected_python_type: typing. Type[~ T],
) -> ~ T
Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised
Parameter
Type
ctx
flytekit.core.context_manager.FlyteContext
lv
flytekit.models.literals.Literal
expected_python_type
typing.Type[~T]
Property
Type
Description
is_async
name
python_type
This returns the python type
type_assertions_enabled
Indicates if the transformer wants type assertions to be enabled at the core type engine layer
Special Task type that will be used in the failure node. Propeller will pass this error to failure task, so users
have to add an input with this type to the failure task.
class FlyteError (
message: str,
failed_node_id: str,
)
Parameter
Type
message
str
failed_node_id
str
def from_dict (
d,
dialect,
)
def from_json (
data: typing. Union[str, bytes, bytearray],
decoder: collections. abc. Callable[[typing. Union[str, bytes, bytearray]], dict[typing. Any, typing. Any]],
from_dict_kwargs: typing. Any,
) -> ~ T
Parameter
Type
data
typing.Union[str, bytes, bytearray]
decoder
collections.abc.Callable[[typing.Union[str, bytes, bytearray]], dict[typing.Any, typing.Any]]
from_dict_kwargs
typing.Any
def to_json (
encoder: collections. abc. Callable[[typing. Any], typing. Union[str, bytes, bytearray]],
to_dict_kwargs: typing. Any,
) -> typing. Union[str, bytes, bytearray]
Parameter
Type
encoder
collections.abc.Callable[[typing.Any], typing.Union[str, bytes, bytearray]]
to_dict_kwargs
typing.Any