Skip to main content

The Flyte Type Engine

When running workflows in Flyte, data must move between isolated tasks, distinct containers, and Flyte's control plane (FlytePropeller and FlyteAdmin). Because task containers may execute on different machines or even be written in different languages, Python objects cannot simply be shared in memory.

The Flyte type system solves this by translating Python native types into Flyte IDL (Interface Definition Language) Protocol Buffer messages (LiteralType for types, and Literal for values) and reconstructing Python instances when tasks execute.

Two core classes in flyte-sdk govern this mechanism:

  • TypeEngine (flyte.types.TypeEngine in types/_type_engine.py): The global registry and dispatch engine. It resolves types, coordinates serialization and deserialization, manages offloaded blobs, and handles batch coroutines.
  • TypeTransformer (flyte.types.TypeTransformer in types/_type_engine.py): The base interface that defines how an individual Python type converts to and from Flyte IDL representations.

The TypeEngine Architecture​

At runtime, TypeEngine manages the conversion lifecycle between Python types and Flyte literals. It maintains an internal registry (_REGISTRY) mapping Python types to their corresponding TypeTransformer instances.

Type Resolution Order​

When TypeEngine.get_transformer(python_type) resolves a transformer for a given type annotation, it executes the following lookup pipeline:

  1. Lazy Loading: Calls lazy_import_transformers() under a thread lock (lazy_import_lock) to import lazy handlers (such as dataframe handlers via lazy_import_dataframe_handler()).
  2. Annotated Types: Inspects typing.Annotated metadata. If a TypeTransformer instance is embedded directly in the annotation arguments, TypeEngine returns that transformer. Otherwise, it unwraps the underlying type (args[0]).
  3. Enum Check: If python_type is an enum.Enum subclass, it routes directly to _ENUM_TRANSFORMER (EnumTransformer) to prevent subclasses like class Foo(str, Enum) from resolving to StrTransformer.
  4. Generic Types (__origin__): For generic types (such as list[int] or dict[str, Any]), it checks for the exact parameterized type first, and falls back to checking python_type.__origin__.
  5. Union Types (types.UnionType / typing.Union): Matches PEP 604 unions (int | str) and typing.Union against UnionTransformer.
  6. Direct Registry Match: Checks if python_type exists as an exact key in TypeEngine._REGISTRY.
  7. MRO Traversal: Inspects the method resolution order (inspect.getmro(python_type)) to find a transformer registered for a superclass.
  8. Dataclass Evaluation: Checks dataclasses.is_dataclass(python_type) against _DATACLASS_TRANSFORMER (DataclassTransformer). This check happens after MRO traversal so you can register custom transformers for dataclass-like classes.
  9. Fallback to FlytePickle: If no transformer matches, TypeEngine logs a warning (display_pickle_warning) and returns FlytePickleTransformer.
Python Type Annotation
         │
         ▼
  typing.Annotated? ───(Yes)──► Transformer in args or unwrap args[0]
         │ (No)
         ▼
    enum.Enum?     ───(Yes)──► EnumTransformer
         │ (No)
         ▼
Generic __origin__?───(Yes)──► Check exact type or origin in _REGISTRY
         │ (No)
         ▼
  In _REGISTRY?    ───(Yes)──► Matched TypeTransformer
         │ (No)
         ▼
   MRO Traversal   ───(Yes)──► Superclass TypeTransformer
         │ (No)
         ▼
   is_dataclass?   ───(Yes)──► DataclassTransformer
         │ (No)
         ▼
  Fallback: FlytePickleTransformer (cloudpickle remote blob)

Registry Operations​

TypeEngine provides class methods to register transformers:

  • TypeEngine.register(transformer, additional_types=None): Registers a transformer for transformer.python_type and any optional additional types. If a type is already registered, it raises a ValueError to prevent accidental overrides.
  • TypeEngine.register_additional_type(transformer, additional_type, override=False): Adds an alias type mapping to an existing transformer, allowing explicit overrides when override=True.
  • TypeEngine.register_restricted_type(name, type): Registers types that are forbidden as task inputs or outputs, wrapping them in a RestrictedTypeTransformer.

Converting Data: Serialization and Deserialization​

TypeEngine provides async methods to convert data across task boundaries:

# Convert a Python value to a Flyte IDL Literal protobuf
literal = await TypeEngine.to_literal(
    python_val=42,
    python_type=int,
    expected=TypeEngine.to_literal_type(int),
)

# Convert a Flyte IDL Literal protobuf back to a Python value
python_val = await TypeEngine.to_python_value(
    lv=literal,
    expected_python_type=int,
)

Offloaded Literal Handling​

When Flyte transfers large payloads, it offloads them to object storage and annotates the Literal message with offloaded_metadata. When TypeEngine.to_python_value(lv, expected_python_type) receives a literal with lv.HasField("offloaded_metadata"), it calls unwrap_offloaded_literal(lv):

  1. Checks lv.offloaded_metadata.uri.
  2. Downloads the raw protobuf payload via storage.get(uri, local_file).
  3. Deserializes the underlying Literal using load_proto_from_file.
  4. Dispatches the unwrapped literal to the resolved TypeTransformer.

Batch Processing of Collections​

When transforming items inside composite collections (such as list or dict), ListTransformer and DictTransformer run coroutines concurrently in chunks using _run_coros_in_chunks. The concurrency chunk size defaults to 10 and is configurable via the _F_TE_MAX_COROS environment variable.

Mapping Task Interfaces​

At task runtime, flyte-sdk converts complete parameter sets using helper methods on TypeEngine:

  • dict_to_literal_map(d, type_hints=None): Converts a Python dictionary of kwargs into a Flyte LiteralMap protobuf message.
  • literal_map_to_kwargs(lm, python_types=None, literal_types=None): Converts a LiteralMap back into a dictionary of Python kwargs matching the task's signature.

For FlyteRemote execution, LiteralsResolver wraps a LiteralMap to provide lazy on-demand deserialization of individual outputs with caching:

from flyte.types import LiteralsResolver

# Resolves values on demand and caches the result in native_values
resolver = LiteralsResolver(literals=literal_map.literals, variable_map=interface.outputs)
val = await resolver.get("my_output_key", as_type=int)

Implementing Custom Type Transformers​

To support custom domain objects without falling back to pickle, subclass TypeTransformer[T] and implement its abstract methods.

The TypeTransformer Interface​

from flyte.types import TypeTransformer, TypeEngine, TypeTransformerFailedError
from flyteidl.core import types_pb2, literals_pb2

class CoordinatesTransformer(TypeTransformer[tuple]):
    def __init__(self):
        super().__init__(name="CoordinatesTransformer", t=tuple)

    def get_literal_type(self, t: type[tuple]) -> types_pb2.LiteralType:
        """Declare the IDL type representation."""
        return types_pb2.LiteralType(
            simple=types_pb2.SimpleType.STRUCT,
            metadata={"type": "coordinates"},
        )

    async def to_literal(
        self,
        python_val: tuple,
        python_type: type[tuple],
        expected: types_pb2.LiteralType,
    ) -> literals_pb2.Literal:
        """Convert a Python object into a Flyte Literal message."""
        if not isinstance(python_val, tuple):
            raise TypeTransformerFailedError(
                f"Expected tuple, got {type(python_val)}"
            )
        struct = literals_pb2.struct_pb2.Struct()
        struct.update({"lat": python_val[0], "lon": python_val[1]})
        return literals_pb2.Literal(
            scalar=literals_pb2.Scalar(generic=struct)
        )

    async def to_python_value(
        self,
        lv: literals_pb2.Literal,
        expected_python_type: type[tuple],
    ) -> tuple:
        """Reconstruct the Python object from a Flyte Literal message."""
        if not (lv and lv.HasField("scalar") and lv.scalar.HasField("generic")):
            raise TypeTransformerFailedError("Expected generic struct literal")
        data = lv.scalar.generic
        return (data["lat"], data["lon"])

    def guess_python_type(self, literal_type: types_pb2.LiteralType) -> type[tuple]:
        """Optionally map a LiteralType back to this Python type."""
        if literal_type.simple == types_pb2.SimpleType.STRUCT:
            if literal_type.metadata.get("type") == "coordinates":
                return tuple
        raise ValueError(f"Cannot reverse literal type {literal_type}")

    def to_html(self, python_val: tuple, expected_python_type: type[tuple]) -> str:
        """Render a custom HTML representation for the Flyte UI."""
        return f"<div><strong>Lat:</strong> {python_val[0]}, <strong>Lon:</strong> {python_val[1]}</div>"

Simple Primitive Registration​

For simple 1:1 scalar conversions, use SimpleTransformer to eliminate boilerplate:

from flyte.types._type_engine import SimpleTransformer
from flyteidl.core import types_pb2, literals_pb2

# SimpleTransformer accepts lambda functions for to_literal and from_literal
my_str_transformer = SimpleTransformer(
    name="CustomStr",
    t=str,
    lt=types_pb2.LiteralType(simple=types_pb2.SimpleType.STRING),
    to_literal_transformer=lambda v: literals_pb2.Literal(
        scalar=literals_pb2.Scalar(primitive=literals_pb2.Primitive(string_value=v))
    ),
    from_literal_transformer=lambda lv: lv.scalar.primitive.string_value,
)

Built-In Transformers​

flyte-sdk registers transformers for standard Python types during initialization (_register_default_type_transformers()):

Primitives and Datetime Types​

  • int, float, str, bool: Mapped via SimpleTransformer to SimpleType.INTEGER, SimpleType.FLOAT, SimpleType.STRING, and SimpleType.BOOLEAN.
  • datetime.datetime, datetime.date, datetime.timedelta: Mapped via SimpleTransformer to SimpleType.DATETIME, SimpleType.DATETIME (date as datetime), and SimpleType.DURATION.
  • NoneType / None: Mapped to SimpleType.NONE.

Containers and Collections​

  • typing.List[T] (ListTransformer): Expects univariate lists. Recursively converts elements via TypeEngine.to_literal_type and produces a LiteralCollection.
  • typing.Dict[str, T] (DictTransformer): Maps string-keyed dictionaries to types_pb2.LiteralType(map_value_type=...) and produces a LiteralMap. Non-string-keyed or untyped dictionaries are serialized as MessagePack binary scalars.
  • typing.Union[T1, T2, ...] (UnionTransformer): Creates LiteralType(union_type=UnionType(variants=...)). When converting to a literal, it tags the literal value with the matched transformer name to ensure unambiguous deserialization.

Structured Models (Pydantic and Dataclasses)​

  • pydantic.BaseModel (PydanticTransformer): Converts models to JSON via model_dump_json(), encodes the result into MessagePack bytes, and wraps them in a binary scalar literal (Binary(value=msgpack_bytes, tag="msgpack")).
  • @dataclasses.dataclass (DataclassTransformer): Uses mashumaro (MessagePackEncoder and MessagePackDecoder) to serialize dataclasses into MessagePack binary IDL literals and extracts JSON Schema drafts for UI inspection using mashumaro.jsonschema.

Enums and Protobufs​

  • enum.Enum (EnumTransformer): Maps string-valued enums to LiteralType.EnumType(values=...) and stores values as string primitives.
  • google.protobuf.message.Message (ProtobufTransformer): Serializes generic Protobuf structures and struct_pb2.Struct dictionaries into generic literals.

Fallbacks and Restricted Types​

Pickling Fallback (FlytePickle)​

When passing an arbitrary Python object that has no registered TypeTransformer, TypeEngine uses FlytePickleTransformer:

  1. Serializes the Python object into bytes using cloudpickle.dumps().
  2. Calculates the MD5 digest of the bytes and writes them to local storage.
  3. Uploads the file to remote object storage using storage.put().
  4. Returns a Literal with a Blob scalar of format PythonPickle and SINGLE dimensionality.

Pickled objects cannot be inspected or modified by backend tasks in other languages and emit a console warning on conversion.

Restricted Types (tuple and NamedTuple)​

Flyte explicitly restricts tuple, typing.Tuple, and typing.NamedTuple from being used directly as input or output value types:

TypeEngine.register_restricted_type("non typed tuple", tuple)
TypeEngine.register_restricted_type("non typed tuple", typing.Tuple)
TypeEngine.register_restricted_type("named tuple", NamedTuple)

Attempting to pass or return a tuple directly raises an AssertionError or RestrictedTypeError:

AssertionError: Tuples are not a supported type for individual values in Flyte - got a tuple - (1, 2).
If using named tuple in an inner task, please, de-reference the actual attribute that you want to use.

To pass multiple outputs from a task, use a typing.NamedTuple as the return type annotation of the task interface itself (where TypeEngine.named_tuple_to_variable_map maps each field to an independent output variable), rather than returning a raw tuple as a single value.