flytekit.core.type_engine

Directory

Classes

Errors

Methods

Variables

Methods

convert_marshmallow_json_schema_to_python_class()

def convert_marshmallow_json_schema_to_python_class(
    schema: dict,
    schema_name: typing.Any,
) -> type

Generate a model class based on the provided JSON Schema

convert_mashumaro_json_schema_to_python_class()

def convert_mashumaro_json_schema_to_python_class(
    schema: dict,
    schema_name: typing.Any,
) -> type

Generate a model class based on the provided JSON Schema

dataclass_from_dict()

def dataclass_from_dict(
    cls: type,
    src: typing.Dict[str, typing.Any],
) -> typing.Any

Utility function to construct a dataclass object from dict

generate_attribute_list_from_dataclass_json()

def generate_attribute_list_from_dataclass_json(
    schema: dict,
    schema_name: typing.Any,
)

generate_attribute_list_from_dataclass_json_mixin()

def generate_attribute_list_from_dataclass_json_mixin(
    schema: dict,
    schema_name: typing.Any,
)

get_batch_size()

def get_batch_size(
    t: Type,
) -> Optional[int]

get_underlying_type()

def get_underlying_type(
    t: Type,
) -> Type

Return the underlying type for annotated types or the type itself

is_annotated()

def is_annotated(
    t: Type,
) -> bool

modify_literal_uris()

def modify_literal_uris(
    lit: Literal,
)

Modifies the literal object recursively to replace the URIs with the native paths in case they are of type “flyte://”

strict_type_hint_matching()

def strict_type_hint_matching(
    input_val: typing.Any,
    target_literal_type: LiteralType,
) -> typing.Type

Try to be smarter about guessing the type of the input (and hence the transformer). If the literal type from the transformer for type(v), matches the literal type of the interface, then we can use type(). Otherwise, fall back to guess python type from the literal type. Raises ValueError, like in case of [1,2,3] type() will just give list, which won’t work. Raises ValueError also if the transformer found for the raw type doesn’t have a literal type match.

flytekit.core.type_engine.AsyncTypeTransformer

Base transformer type that should be implemented for every python native type that can be handled by flytekit

class AsyncTypeTransformer(
    name: str,
    t: Type[T],
    enable_type_assertions: bool,
)

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

async_to_literal()

def async_to_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

async_to_python_value()

def async_to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> Optional[T]

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Type[T],
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[T]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: typing.Any,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> Optional[T]

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.BatchSize

This is used to annotate a FlyteDirectory when we want to download/upload the contents of the directory in batches. For example,

@task
def t1(directory: Annotated[FlyteDirectory, BatchSize(10)]) -> Annotated[FlyteDirectory, BatchSize(100)]:
    ...
    return FlyteDirectory(...)

In the above example flytekit will download all files from the input directory in chunks of 10, i.e. first it downloads 10 files, loads them to memory, then writes those 10 to local disk, then it loads the next 10, so on and so forth. Similarly, for outputs, in this case flytekit is going to upload the resulting directory in chunks of 100.

class BatchSize(
    val: int,
)

Properties

flytekit.core.type_engine.BinaryIOTransformer

Handler for BinaryIO

def BinaryIOTransformer()

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Type[typing.BinaryIO],
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[T]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: typing.BinaryIO,
    python_type: Type[typing.BinaryIO],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[typing.BinaryIO],
) -> typing.BinaryIO

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.DataclassTransformer

The Dataclass Transformer provides a type transformer for dataclasses.

The dataclass is converted to and from MessagePack Bytes by the mashumaro library and is transported between tasks using the Binary IDL representation. Also, the type declaration will try to extract the JSON Schema for the object, if possible, and pass it with the definition.

The lifecycle of the dataclass in the Flyte type system is as follows:

1. Serialization: The dataclass transformer converts the dataclass to MessagePack Bytes. (1) Handle dataclass attributes to make them serializable with mashumaro. (2) Use the mashumaro API to serialize the dataclass to MessagePack Bytes. (3) Use MessagePack Bytes to create a Flyte Literal. (4) Serialize the Flyte Literal to a Binary IDL Object.

2. Deserialization: The dataclass transformer converts the MessagePack Bytes back to a dataclass. (1) Convert MessagePack Bytes to a dataclass using mashumaro. (2) Handle dataclass attributes to ensure they are of the correct types.

For Json Schema, we use https://github.com/fuhrysteve/marshmallow-jsonschema library.

Example

@dataclass
class Test(DataClassJsonMixin):
    a: int
    b: str

from marshmallow_jsonschema import JSONSchema
t = Test(a=10,b="e")
JSONSchema().dump(t.schema())

Output will look like

{'$schema': 'http://json-schema.org/draft-07/schema#',
    'definitions': {'TestSchema': {'properties': {'a': {'title': 'a',
        'type': 'number',
        'format': 'integer'},
    'b': {'title': 'b', 'type': 'string'}},
    'type': 'object',
    'additionalProperties': False}},
    '$ref': '#/definitions/TestSchema'}


> [!NOTE]
> The schema support is experimental and is useful for auto-completing in the UI/CLI




```python
def DataclassTransformer()

Methods

assert_type()

def assert_type(
    expected_type: Type[DataClassJsonMixin],
    v: T,
)

from_binary_idl()

def from_binary_idl(
    binary_idl_object: Binary,
    expected_python_type: Type[T],
) -> 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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Type[T],
) -> LiteralType

Extracts the Literal type definition for a Dataclass and returns a type Struct. If possible also extracts the JSONSchema for the dataclass.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_generic_literal()

def to_generic_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> Literal

Serializes a dataclass or dictionary to a Flyte literal, handling both JSON and MessagePack formats. Set FLYTE_USE_OLD_DC_FORMAT=true to use the old JSON-based format. Note: This is deprecated and will be removed in the future.

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> T

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.DictTransformer

Transformer that transforms an univariate dictionary Dict[str, T] to a Literal Map or transforms an untyped dictionary to a Binary Scalar Literal with a Struct Literal Type.

def DictTransformer()

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

async_to_literal()

def async_to_literal(
    ctx: FlyteContext,
    python_val: typing.Any,
    python_type: Type[dict],
    expected: LiteralType,
) -> 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

async_to_python_value()

def async_to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[dict],
) -> dict

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

dict_to_binary_literal()

def dict_to_binary_literal(
    ctx: FlyteContext,
    v: dict,
    python_type: Type[dict],
    allow_pickle: bool,
) -> Literal

Converts a Python dictionary to a Flyte-specific Literal using MessagePack encoding. Falls back to Pickle if encoding fails and allow_pickle is True.

dict_to_generic_literal()

def dict_to_generic_literal(
    ctx: FlyteContext,
    v: dict,
    python_type: Type[dict],
    allow_pickle: bool,
) -> Literal

This is deprecated from flytekit 1.14.0. Creates a flyte-specific Literal value from a native python dictionary. Note: This is deprecated and will be removed in the future.

extract_types()

def extract_types(
    t: Optional[Type[dict]],
) -> typing.Tuple

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Type[dict],
) -> LiteralType

Transforms a native python dictionary to a flyte-specific LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Union[Type[dict], typing.Dict[Type, Type]]

Converts the Flyte LiteralType to a python object type.

is_pickle()

def is_pickle(
    python_type: Type[dict],
) -> bool

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: typing.Any,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> Optional[T]

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.EnumTransformer

Enables converting a python type enum.Enum to LiteralType.EnumType

def EnumTransformer()

Methods

assert_type()

def assert_type(
    t: Type[enum.Enum],
    v: T,
)

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Type[T],
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[enum.Enum]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: enum.Enum,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> T

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.ListTransformer

Transformer that handles a univariate typing.List[T]

def ListTransformer()

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

async_to_literal()

def async_to_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

async_to_python_value()

def async_to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> typing.Optional[typing.List[T]]

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Type[T],
) -> Optional[LiteralType]

Only univariate Lists are supported in Flyte

get_sub_type()

def get_sub_type(
    t: Type[T],
) -> Type[T]

Return the generic Type T of the List

get_sub_type_or_none()

def get_sub_type_or_none(
    t: Type[T],
) -> Optional[Type[T]]

Return the generic Type T of the List, or None if the generic type cannot be inferred

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> list

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: typing.Any,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> Optional[T]

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.LiteralsResolver

LiteralsResolver is a helper class meant primarily for use with the FlyteRemote experience or any other situation where you might be working with LiteralMaps. This object allows the caller to specify the Python type that should correspond to an element of the map.

class LiteralsResolver(
    literals: typing.Dict[str, Literal],
    variable_map: Optional[Dict[str, _interface_models.Variable]],
    ctx: Optional[FlyteContext],
)

Methods

as_python_native()

def as_python_native(
    python_interface: Interface,
) -> typing.Any

This should return the native Python representation, compatible with unpacking. This function relies on Python interface outputs being ordered correctly.

clear()

def clear()

D.clear() -> None. Remove all items from D.

copy()

def copy()

fromkeys()

def fromkeys(
    iterable,
    value,
)

get()

def get(
    attr: str,
    as_type: Optional[typing.Type],
) -> n: Python native value from the LiteralMap

This will get the attr value from the Literal map, and invoke the TypeEngine to convert it into a Python native value. A Python type can optionally be supplied. If successful, the native value will be cached and future calls will return the cached value instead.

get_literal()

def get_literal(
    key: str,
) -> Literal

items()

def items()

D.items() -> a set-like object providing a view on D’s items

keys()

def keys()

D.keys() -> a set-like object providing a view on D’s keys

pop()

def pop(
    key,
    default,
)

D.pop(k[,d]) -> v, remove specified key and return the corresponding value. If key is not found, d is returned if given, otherwise KeyError is raised.

popitem()

def popitem()

D.popitem() -> (k, v), remove and return some (key, value) pair as a 2-tuple; but raise KeyError if D is empty.

setdefault()

def setdefault(
    key,
    default,
)

D.setdefault(k[,d]) -> D.get(k,d), also set D[k]=d if k not in D

update()

def update(
    other,
    kwds,
)

D.update([E, ]**F) -> None. Update D from mapping/iterable E and F. If E present and has a .keys() method, does: for k in E.keys(): D[k] = E[k] If E present and lacks .keys() method, does: for (k, v) in E: D[k] = v In either case, this is followed by: for k, v in F.items(): D[k] = v

update_type_hints()

def update_type_hints(
    type_hints: typing.Dict[str, typing.Type],
)

values()

def values()

D.values() -> an object providing a view on D’s values

Properties

flytekit.core.type_engine.ProtobufTransformer

Base transformer type that should be implemented for every python native type that can be handled by flytekit

def ProtobufTransformer()

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Type[T],
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[T]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

tag()

def tag(
    expected_python_type: Type[T],
) -> str

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> Literal

Convert the protobuf struct to literal.

This conversion supports two types of python_val:

1. google.protobuf.struct_pb2.Struct: A dictionary-like message
2. google.protobuf.struct_pb2.ListValue: An ordered collection of values

For details, please refer to the following issue: https://github.com/flyteorg/flyte/issues/5959

Because the remote handling works without errors, we implement conversion with the logic as below: https://github.com/flyteorg/flyte/blob/a87585ab7cbb6a047c76d994b3f127c4210070fd/flytepropeller/pkg/controller/nodes/attr_path_resolver.go#L72-L106

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> T

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.RestrictedTypeError

Common base class for all non-exit exceptions.

flytekit.core.type_engine.RestrictedTypeTransformer

Types registered with the RestrictedTypeTransformer are not allowed to be converted to and from literals. In other words, Restricted types are not allowed to be used as inputs or outputs of tasks and workflows.

class RestrictedTypeTransformer(
    name: str,
    t: Type[T],
)

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Optional[Type[T]],
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[T]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> T

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.SimpleTransformer

A Simple implementation of a type transformer that uses simple lambdas to transform and reduces boilerplate

class SimpleTransformer(
    name: str,
    t: Type[T],
    lt: LiteralType,
    to_literal_transformer: typing.Callable[[T], Literal],
    from_literal_transformer: typing.Callable[[Literal], T],
)

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Optional[Type[T]],
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[T]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> T

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.TextIOTransformer

Handler for TextIO

def TextIOTransformer()

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: typing.TextIO,
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[T]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: typing.TextIO,
    python_type: Type[typing.TextIO],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[typing.TextIO],
) -> typing.TextIO

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.TypeEngine

Core Extensible TypeEngine of Flytekit. This should be used to extend the capabilities of FlyteKits type system. Users can implement their own TypeTransformers and register them with the TypeEngine. This will allow special handling of user objects

Methods

async_to_literal()

def async_to_literal(
    ctx: FlyteContext,
    python_val: typing.Any,
    python_type: Type[T],
    expected: LiteralType,
) -> Literal

Converts a python value of a given type and expected LiteralType into a resolved Literal value.

async_to_python_value()

def async_to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type,
) -> typing.Any

calculate_hash()

def calculate_hash(
    python_val: typing.Any,
    python_type: Type[T],
) -> Optional[str]

dict_to_literal_map()

def dict_to_literal_map(
    ctx: FlyteContext,
    d: typing.Dict[str, typing.Any],
    type_hints: Optional[typing.Dict[str, type]],
) -> LiteralMap

dict_to_literal_map_pb()

def dict_to_literal_map_pb(
    ctx: FlyteContext,
    d: typing.Dict[str, typing.Any],
    type_hints: Optional[typing.Dict[str, type]],
) -> Optional[literals_pb2.LiteralMap]

get_available_transformers()

def get_available_transformers()

Returns all python types for which transformers are available

get_transformer()

def get_transformer(
    python_type: Type,
) -> TypeTransformer[T]

Implements a recursive search for the transformer.

guess_python_type()

def guess_python_type(
    flyte_type: LiteralType,
) -> Type[T]

Transforms a flyte-specific LiteralType to a regular python value.

guess_python_types()

def guess_python_types(
    flyte_variable_dict: typing.Dict[str, _interface_models.Variable],
) -> typing.Dict[str, type]

Transforms a dictionary of flyte-specific Variable objects to a dictionary of regular python values.

lazy_import_transformers()

def lazy_import_transformers()

Only load the transformers if needed.

literal_map_to_kwargs()

def literal_map_to_kwargs(
    ctx: FlyteContext,
    lm: LiteralMap,
    python_types: typing.Optional[typing.Dict[str, type]],
    literal_types: typing.Optional[typing.Dict[str, _interface_models.Variable]],
) -> typing.Dict[str, typing.Any]

named_tuple_to_variable_map()

def named_tuple_to_variable_map(
    t: typing.NamedTuple,
) -> _interface_models.VariableMap

Converts a python-native NamedTuple to a flyte-specific VariableMap of named literals.

register()

def register(
    transformer: TypeTransformer,
    additional_types: Optional[typing.List[Type]],
)

This should be used for all types that respond with the right type annotation when you use type(…) function

register_additional_type()

def register_additional_type(
    transformer: TypeTransformer[T],
    additional_type: Type[T],
    override,
)

register_restricted_type()

def register_restricted_type(
    name: str,
    type: Type[T],
)

to_html()

def to_html(
    ctx: FlyteContext,
    python_val: typing.Any,
    expected_python_type: Type[typing.Any],
) -> str

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: typing.Any,
    python_type: Type[T],
    expected: LiteralType,
) -> Literal

The current dance is because we are allowing users to call from an async function, this synchronous to_literal function, and allowing this to_literal function, to then invoke yet another async function, namely an async transformer.

to_literal_checks()

def to_literal_checks(
    python_val: typing.Any,
    python_type: Type[T],
    expected: LiteralType,
)

to_literal_type()

def to_literal_type(
    python_type: Type[T],
) -> LiteralType

Converts a python type into a flyte specific LiteralType

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type,
) -> typing.Any

Converts a Literal value with an expected python type into a python value.

unwrap_offloaded_literal()

def unwrap_offloaded_literal(
    ctx: FlyteContext,
    lv: Literal,
) -> Literal

flytekit.core.type_engine.TypeTransformer

Base transformer type that should be implemented for every python native type that can be handled by flytekit

class TypeTransformer(
    name: str,
    t: Type[T],
    enable_type_assertions: bool,
)

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

from_binary_idl()

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)

from_generic_idl()

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.

get_literal_type()

def get_literal_type(
    t: Type[T],
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[T]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

to_html()

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

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> 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

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> Optional[T]

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.core.type_engine.TypeTransformerFailedError

Inappropriate argument type.

flytekit.core.type_engine.UnionTransformer

Transformer that handles a typing.Union[T1, T2, …]

def UnionTransformer()

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
)

async_to_literal()

def async_to_literal(
    ctx: FlyteContext,
    python_val: T,
    python_type: Type[T],
    expected: LiteralType,
) -> typing.Union[Literal, asyncio.Future]

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

async_to_python_value()

def async_to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
) -> Optional[typing.Any]

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

from_binary_idl()

def from_binary_idl(
    binary_idl_object: Binary,
    expected_python_type: Type[T],
) -> Optional[T]