Skip to main content

The Flyte Type Engine

When you pass complex Python data structures between Flyte tasks, Flyte cannot simply pass in-memory Python objects because tasks may run in isolated containers across different machines or execution environments. The flyte-sdk type system solves this by translating Python types and values into Flyte IDL primitives (LiteralType, Literal, and LiteralMap), validating type contracts statically and dynamically, and deserializing them back into Python types on task invocation.

At the center of this mechanism is TypeEngine and its extensible transformer architecture based on TypeTransformer.

Core Architecture and Data Flow​

In flyte-sdk, type translation operates in two distinct phases:

  1. Interface Compilation (Static Analysis): TypeEngine.to_literal_type(python_type) inspects task signatures and returns a Protobuf LiteralType defining the input and output interface for FlytePropeller and the Flyte control plane.
  2. Runtime Execution (Serialization & Deserialization):
    • When a task completes, TypeEngine.to_literal(python_val, python_type, expected) serializes the Python output object into a Protobuf Literal.
    • When a downstream task executes, TypeEngine.to_python_value(lv, expected_python_type) deserializes the incoming Literal back into the required Python object.
+-----------------------------------------------------------------------------------------+
|                                    TypeEngine                                           |
|                                                                                         |
|   Python Type / Annotation                           Flyte IDL Representation           |
|  +------------------------+                        +--------------------------+         |
|  | int, str, float, bool  |  to_literal_type()     | LiteralType              |         |
|  | List[T], Dict[str, V]  | ---------------------> |  (simple, collection,    |         |
|  | Dataclass, BaseModel   |                        |   map, struct, union...) |         |
|  +------------------------+                        +--------------------------+         |
|             |                                                   |                       |
|             | to_literal()                                      | to_python_value()     |
|             v                                                   v                       |
|  +------------------------+                        +--------------------------+         |
|  | Python Instance / Val  | ---------------------> | Literal / LiteralMap     |         |
|  |  (e.g., File("..."))   | <--------------------- |  (scalar.primitive, blob,|         |
|  +------------------------+                        |   scalar.binary, map...) |         |
+-----------------------------------------------------------------------------------------+

The Transformer Resolution Algorithm​

When TypeEngine.get_transformer(python_type) searches for an appropriate transformer, it evaluates candidates in types/_type_engine.py in the following order:

  1. Annotated Types: If wrapped in typing.Annotated, TypeEngine inspects metadata arguments. If an explicit TypeTransformer instance is present in the annotation arguments, it returns that transformer; otherwise, it resolves against the underlying wrapped type (args[0]).
  2. Enums: Subclasses of enum.Enum are explicitly matched to _ENUM_TRANSFORMER (avoiding accidental matching to string transformers if the enum inherits from str).
  3. Generic Types: If python_type contains __origin__, TypeEngine first checks if the full generic type is registered in _REGISTRY, and falls back to checking python_type.__origin__.
  4. PEP 604 Union Types: types.UnionType maps to the registered union transformer.
  5. Direct Registry Match: Exact key lookup in _REGISTRY.
  6. Method Resolution Order (MRO): Walks inspect.getmro(python_type) to check if a base class has a registered transformer.
  7. Dataclass Fallback: If dataclasses.is_dataclass(python_type) is true, returns _DATACLASS_TRANSFORMER.
  8. Pickle Fallback: If no transformer matches, TypeEngine logs a warning (display_pickle_warning) and returns an instance of FlytePickleTransformer.

Built-In Type Transformers​

flyte-sdk registers transformers for standard Python types during module initialization in types/_type_engine.py:

Primitives and Simple Types​

SimpleTransformer handles standard Python scalar types using direct mappings:

  • int -> SimpleType.INTEGER (Literal.scalar.primitive.integer)
  • float -> SimpleType.FLOAT (Literal.scalar.primitive.float_value)
  • str -> SimpleType.STRING (Literal.scalar.primitive.string_value)
  • bool -> SimpleType.BOOLEAN (Literal.scalar.primitive.boolean)
  • datetime.datetime -> SimpleType.DATETIME
  • datetime.timedelta -> SimpleType.DURATION
  • None / NoneType -> SimpleType.NONE

Dataclasses and Pydantic Models​

  • Dataclasses: DataclassTransformer uses mashumaro (MessagePackEncoder and MessagePackDecoder) to serialize dataclass instances to MessagePack binary payloads wrapped in a Literal(scalar=Scalar(binary=Binary(value=msgpack_bytes, tag="msgpack"))). It also extracts JSON Schema definitions using mashumaro.jsonschema.build_json_schema for backend interface metadata.
  • Pydantic Models: PydanticTransformer converts subclasses of pydantic.BaseModel into MessagePack binary IDL literals and generates JSON Schema metadata via BaseModel.model_json_schema().

Collections and Containers​

  • Lists (ListTransformer): Supports univariate typing.List[T]. Transforms each element recursively using TypeEngine.to_literal and packages elements into a Protobuf LiteralCollection.
  • Dictionaries (DictTransformer):
    • Typed dictionaries with string keys (Dict[str, T]) serialize into LiteralMap with map_value_type.
    • Untyped dictionaries or dictionaries with non-string keys serialize into binary MessagePack structs via dict_to_binary_literal.
  • Unions (UnionTransformer): Supports typing.Union[T1, T2, ...]. Serializes values into a Scalar.union literal containing the inner literal and a tag identifying the selected variant. If a Python value matches multiple union variants structurally, UnionTransformer raises a TypeError indicating ambiguous choices.

Files, Directories, and Blobs​

Types like File (in io/_file.py) and Dir (in io/_dir.py) register specialized transformers that map instances to types_pb2.BlobType and literals_pb2.Blob containing URI references to remote or local storage.

Implementing and Registering Custom Type Transformers​

To support custom domain objects, subclass TypeTransformer[T] and register your class with TypeEngine.register().

Step 1: Subclass TypeTransformer​

Implement the required abstract methods: get_literal_type, to_literal, and to_python_value. You can optionally override guess_python_type and assert_type. The following example demonstrates how FileTransformer in io/_file.py implements this interface:

from pathlib import Path
from typing import Type
from flyteidl.core import literals_pb2, types_pb2
from flyte.io import File
from flyte.types import TypeEngine, TypeTransformer
from flyte.types._type_engine import TypeTransformerFailedError


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

    def get_literal_type(self, t: Type[File]) -> types_pb2.LiteralType:
        """Get the Flyte literal type for a File type."""
        return types_pb2.LiteralType(
            blob=types_pb2.BlobType(
                format="",
                dimensionality=types_pb2.BlobType.BlobDimensionality.SINGLE,
            )
        )

    async def to_literal(
        self,
        python_val: File,
        python_type: Type[File],
        expected: types_pb2.LiteralType,
    ) -> literals_pb2.Literal:
        """Convert a File object to a Flyte literal."""
        if not isinstance(python_val, File):
            raise TypeTransformerFailedError(f"Expected File object, received \{type(python_val)\}")

        return literals_pb2.Literal(
            scalar=literals_pb2.Scalar(
                blob=literals_pb2.Blob(
                    metadata=literals_pb2.BlobMetadata(
                        type=types_pb2.BlobType(
                            format=python_val.format,
                            dimensionality=types_pb2.BlobType.BlobDimensionality.SINGLE,
                        )
                    ),
                    uri=python_val.path,
                )
            ),
            hash=python_val.hash if python_val.hash else None,
        )

    async def to_python_value(
        self,
        lv: literals_pb2.Literal,
        expected_python_type: Type[File],
    ) -> File:
        """Convert a Flyte literal to a File object."""
        if not lv.scalar.HasField("blob"):
            raise TypeTransformerFailedError(f"Expected blob literal, received \{lv\}")
        if not lv.scalar.blob.metadata.type.dimensionality == types_pb2.BlobType.BlobDimensionality.SINGLE:
            raise TypeTransformerFailedError(
                f"Expected single part blob, received \{lv.scalar.blob.metadata.type.dimensionality\}"
            )

        uri = lv.scalar.blob.uri
        filename = Path(uri).name
        hash_value = lv.hash if lv.hash else None
        return File(path=uri, name=filename, format=lv.scalar.blob.metadata.type.format, hash=hash_value)

    def guess_python_type(self, literal_type: types_pb2.LiteralType) -> Type[File]:
        """Guess the Python type from a Flyte literal type."""
        if (
            literal_type.HasField("blob")
            and literal_type.blob.dimensionality == types_pb2.BlobType.BlobDimensionality.SINGLE
        ):
            return File
        raise ValueError(f"Transformer \{self\} cannot reverse \{literal_type\}")

Step 2: Register the Transformer​

Register your transformer with TypeEngine at module import time:

TypeEngine.register(CustomFileTransformer())

Attempting to register a transformer for a Python type that already has a registered transformer will raise a ValueError to prevent unintended overriding:

# Raises ValueError: Transformer CustomFileTransformer for type \<class 'flyte.io._file.File'> is already registered.
TypeEngine.register(CustomFileTransformer())

Runtime Execution Lifecycle​

During workflow and task execution, flyte-sdk runtime routines in _internal/runtime/convert.py invoke TypeEngine helper methods to translate batches of task inputs and outputs.

Converting Inputs to Kwargs​

When a task executes, incoming parameters arrive as a LiteralMap. The runtime invokes TypeEngine.literal_map_to_kwargs:

# _internal/runtime/convert.py
native_vals = await TypeEngine.literal_map_to_kwargs(
    literals_pb2.LiteralMap(literals=literals),
    python_interface.get_input_types(),
)

Internally, literal_map_to_kwargs spawns concurrent asyncio tasks to convert every literal in parallel:

kwargs = \{\}
for i, k in enumerate(lm.literals):
    kwargs[k] = asyncio.create_task(
        TypeEngine.to_python_value(lm.literals[k], python_interface_inputs[k])
    )
await asyncio.gather(*kwargs.values())

Offloaded Literals Unwrapping​

When handling large payloads offloaded to remote storage (e.g. S3, GCS), incoming literals contain offloaded_metadata. TypeEngine.to_python_value automatically detects this field and calls TypeEngine.unwrap_offloaded_literal(lv):

if lv.HasField("offloaded_metadata"):
    lv = await cls.unwrap_offloaded_literal(lv)

unwrap_offloaded_literal downloads the serialized Protobuf file asynchronously using flyte.storage and loads the unwrapped Literal before passing it to the resolved transformer.

Batched Async Transformations​

When transforming collections (ListTransformer) and maps (DictTransformer), flyte-sdk processes items in concurrent batches via _run_coros_in_chunks. The batch concurrency limit is configured by the environment variable _F_TE_MAX_COROS (defaulting to 10):

# Batching coroutines in types/_type_engine.py
lit_list = [TypeEngine.to_literal(x, t, expected.collection_type) for x in python_val]
lit_list = await _run_coros_in_chunks(lit_list, batch_size=_TYPE_ENGINE_COROS_BATCH_SIZE)

Type Constraints and Edge Cases​

Restricted Types​

tuple, typing.Tuple, and typing.NamedTuple are explicitly registered under TypeEngine._RESTRICTED_TYPES with RestrictedTypeTransformer. Using tuples directly as task input or output types will raise a RestrictedTypeError or AssertionError:

# Tuples cannot be converted directly to individual literal values
TypeEngine.to_literal_checks(python_val=(1, 2), python_type=tuple, expected=...)
# Raises: AssertionError: Tuples are not a supported type for individual values in Flyte...

For multiple task outputs, use typed output syntax, @dataclass, or named attributes instead of raw tuples.

Enum Constraints​

EnumTransformer requires all enum members to have string values. Enums with integer or arbitrary values will raise TypeTransformerFailedError("Only EnumTypes with value of string are supported").

String Keys in Dictionaries​

For standard Flyte MapType conversion, dictionary keys must be strings (Dict[str, T]). Dictionaries with non-string keys (e.g., Dict[int, str]) bypass MapType and are serialized into binary MessagePack structs.

Cloudpickle Fallback​

When a Python type without a registered transformer is passed through TypeEngine, it falls back to FlytePickleTransformer. While this ensures serialization works out of the box, pickled objects:

  • Cannot be inspected or visualized in the Flyte UI.
  • Require binary compatibility and matching Python/library versions across tasks.
  • Incur serialization overhead for large data structures.

Implement and register a custom TypeTransformer for production workflows requiring portable, inspectable IDL types.