The TypeTransformer Interface

The conversion lifecycle

A Flyte task has a Python annotation and value on one side, but its interface and run data use Flyte IDL protobuf messages on the other. TypeTransformer[T] is the per-native-type adapter between those representations: get_literal_type describes the value with a LiteralType, to_literal produces a Literal, and to_python_value reconstructs a Python value from a Literal.

The public flyte.types package re-exports both TypeTransformer and TypeTransformerFailedError:

from flyte.types import TypeTransformer, TypeTransformerFailedError

A transformer is constructed with a display name, the Python type it handles, and an assertion setting. FileTransformer in io/_file.py shows the normal constructor pattern:

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

The base constructor stores these values as name, python_type, and type_assertions_enabled. It also creates per-instance MessagePack encoder and decoder dictionaries. The name appears in the transformer's representation—for example, __repr__ formats it as "{name} Transforms ({python type}) to Flyte native".

The data flow is:

Python annotation ──get_literal_type──> LiteralType
Python value      ──to_literal────────> Literal
Literal           ──to_python_value───> Python value

TypeEngine selects the transformer registered for the declared or expected Python type and coordinates each leg of this flow. It also handles recursive collection, map, and union conversion around individual transformers.

The three required methods

Describe the representation with get_literal_type

Implement get_literal_type(self, t) to map a Python annotation to a Flyte LiteralType. This is schema construction, not value serialization. TypeEngine.to_literal_type(input_type) uses this information when it builds typed interfaces, and the default-input path uses the resulting type as the expected type for value conversion.

FileTransformer declares a single-part blob and leaves its format empty because the format is determined by the generic type in the surrounding implementation:

def get_literal_type(self, t: Type[File]) -> types_pb2.LiteralType:
    return types_pb2.LiteralType(
        blob=types_pb2.BlobType(
            format="",
            dimensionality=types_pb2.BlobType.BlobDimensionality.SINGLE,
        )
    )

DirTransformer has the same shape but declares MULTIPART dimensionality. Thus, the literal type is part of the contract: a structurally valid blob with the wrong dimensionality is not accepted by the corresponding reverse conversion.

Serialize with async to_literal

Implement to_literal(self, python_val, python_type, expected) as an asynchronous method. python_val is the actual value; python_type is the declared/interface type; and expected is the LiteralType expected by the caller. The method's docstring explicitly directs implementers to use the passed python_type rather than infer a type from type(python_val). That distinction matters when the declaration is generic, a union, or otherwise differs from the runtime class.

The FileTransformer implementation validates the wrapper and writes its path, format, dimensionality, and optional hash into a protobuf literal:

async def to_literal(
    self,
    python_val: File,
    python_type: Type[File],
    expected: types_pb2.LiteralType,
) -> literals_pb2.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,
    )

This transformer does not perform file I/O. Its source docstring states that I/O is the user's responsibility; serialization records the path/URI and metadata.

Deserialize with async to_python_value

Implement to_python_value(self, lv, expected_python_type) to turn a received Literal into the requested native type. The expected Python type is supplied by the interface, rather than discovered from the literal alone.

For files, the reverse method checks both the protobuf shape and blob dimensionality before reconstructing the wrapper:

async def to_python_value(
    self,
    lv: literals_pb2.Literal,
    expected_python_type: Type[File],
) -> File:
    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
    f: File = File(path=uri, name=filename, format=lv.scalar.blob.metadata.type.format, hash=hash_value)
    return f

DirTransformer applies the same validation pattern with MULTIPART instead of SINGLE and reconstructs a Dir. These checks mean that File and Dir transformers do not merely decode any blob: they require the dimensionality declared by their native wrapper.

How TypeEngine invokes a transformer

For forward conversion, TypeEngine.to_literal first looks up a transformer using python_type. It invokes the shared assertion hook only when type_assertions_enabled is true, awaits the transformer's to_literal, and then calls modify_literal_uris before returning the literal:

@classmethod
async def to_literal(
    cls, python_val: typing.Any, python_type: Type[T], expected: types_pb2.LiteralType
) -> literals_pb2.Literal:
    transformer = cls.get_transformer(python_type)

    if transformer.type_assertions_enabled:
        transformer.assert_type(python_type, python_val)

    lv = await transformer.to_literal(python_val, python_type, expected)

    modify_literal_uris(lv)
    return lv

For reverse conversion, TypeEngine.to_python_value first unwraps an offloaded literal when offloaded_metadata is present. It downloads the referenced protobuf with storage.get, loads it using load_proto_from_file, looks up the transformer for expected_python_type, and awaits to_python_value:

@classmethod
async def to_python_value(cls, lv: Literal, expected_python_type: Type) -> typing.Any:
    if lv.HasField("offloaded_metadata"):
        lv = await cls.unwrap_offloaded_literal(lv)

    transformer = cls.get_transformer(expected_python_type)
    res = await transformer.to_python_value(lv, expected_python_type)
    return res

The expected argument therefore has two related but distinct roles. TypeEngine passes a LiteralType to to_literal so the implementation can honor the expected Flyte representation; reverse conversion passes the declared Python type so the implementation knows what native result to produce.

Shared validation and fallback behavior

The base class provides useful behavior, but the three conversion methods remain the extension contract.

Type assertions

assert_type(t, v) raises TypeTransformerFailedError for an ordinary type mismatch. On Python 3.10 and later, if t is a types.GenericAlias, it delegates to isinstance_generic. That helper checks the generic origin—for example, it checks that a list[int] value is a list:

def isinstance_generic(self, obj, generic_alias):
    origin = get_origin(generic_alias)

    if not isinstance(obj, origin):
        raise TypeTransformerFailedError(f"Value '{obj}' is not of container type {origin}")

This is a shallow container check; the base method does not recursively validate the element types. A transformer can override assert_type. DataFrameTransformerEngine does so with an empty implementation because it handles multiple dataframe implementations, and its to_literal performs its own wrapper check instead.

The engine may also reject a None value before transformer conversion when the declared type is not optional. TypeEngine.to_literal_checks raises TypeTransformerFailedError with the expected Python and literal types in that case. A transformer should therefore not assume that every invalid value reaches to_literal.

Reverse type inference

guess_python_type(literal_type) is not required by the abstract interface. The base implementation raises:

raise ValueError("By default, transformers do not translate from Flyte types back to Python types")

Override it only when a transformer can infer a native type from a Flyte literal type. FileTransformer recognizes a single-part blob (except the PythonPickle format) and returns File; DirTransformer recognizes a multipart blob and returns Dir; FlytePickleTransformer recognizes its single-part PythonPickle blob and returns FlytePickle.

Binary literals and HTML

from_binary_idl is the base deserialization path used for untyped dictionaries, dataclasses, Pydantic models, and attribute access. It accepts only a Binary whose tag is MESSAGEPACK. It caches a MessagePackDecoder by expected_python_type, creating it with _default_msgpack_decoder on the first request for that type. A different binary tag raises TypeTransformerFailedError:

if binary_idl_object.tag == MESSAGEPACK:
    try:
        decoder = self._msgpack_decoder[expected_python_type]
    except KeyError:
        decoder = MessagePackDecoder(expected_python_type, pre_decoder_func=_default_msgpack_decoder)
        self._msgpack_decoder[expected_python_type] = decoder
    python_val = decoder.decode(binary_idl_object.value)
    return python_val
else:
    raise TypeTransformerFailedError(f"Unsupported binary format `{binary_idl_object.tag}`")

A custom binary representation must override this method; the base implementation is MessagePack-only. to_html has a simpler default and returns str(python_val).

One failure vocabulary: TypeTransformerFailedError

TypeTransformerFailedError is deliberately lightweight:

class TypeTransformerFailedError(TypeError, AssertionError, ValueError): ...

Because it inherits from all three base exception types, existing type-, assertion-, or value-error handling can also recognize it. Transformer implementations use it for mismatched values and invalid literal shapes, while the base binary path uses it for unsupported tags. UnionTransformer uses transformer failures while probing alternatives: a rejected variant is skipped, and a final TypeTransformerFailedError is raised if no variant converts the value. If more than one variant succeeds, UnionTransformer raises a separate TypeError for an ambiguous choice rather than choosing arbitrarily.

At the runtime boundary, the exception is given task-output context. _internal/runtime/convert.py catches it while converting each output and raises RuntimeDataValidationError with the output name and task name:

for (output_name, python_type), v in zip(interface.outputs.items(), o):
    try:
        lit = await TypeEngine.to_literal(v, python_type, TypeEngine.to_literal_type(python_type))
        named.append(run_definition_pb2.NamedLiteral(name=output_name, value=lit))
    except TypeTransformerFailedError as e:
        raise flyte.errors.RuntimeDataValidationError(output_name, e, task_name)

Concrete integration choices

File and directory wrappers

Use the existing FileTransformer and DirTransformer behavior when your value is represented by a path/URI wrapper. FileTransformer emits a single-part blob and DirTransformer emits a multipart blob; neither uploads or downloads data during conversion. Their reverse methods derive name with Path(uri).name and preserve the blob format and optional literal hash.

Dataframes

DataFrameTransformerEngine is a higher-level transformer for dataframe types rather than a direct registration for every concrete dataframe class. The source explicitly directs custom dataframe types to register with this engine. During registration it associates the engine with an additional type using override=True:

engine = DataFrameTransformerEngine()
TypeEngine.register_additional_type(engine, h.python_type, override=True)

Its assert_type returns without checking the value. Its to_literal later checks a declared DataFrame type and raises a TypeTransformerFailedError when the value is not a DataFrame wrapper, including the message suggesting that the dataframe may need wrapping.

Pickle fallback

TypeEngine.get_transformer searches the registry and the type's MRO, gives registered handling a chance before dataclass handling, and finally returns a new FlytePickleTransformer for an otherwise-unregistered class. FlytePickleTransformer overrides assert_type because its source states that every type can serialize to pickle. It writes a single-part blob with format PythonPickle, stores the Python class name in literal metadata, and reads the blob URI back with FlytePickle.from_pickle.

This fallback means an unregistered class may serialize successfully as a pickle instead of using a custom native representation. Register a transformer when you need a different LiteralType or wire format.

Runtime call sequences

Default interface values demonstrate the schema-before-value order. convert_upload_default_inputs obtains lt with TypeEngine.to_literal_type(input_type), schedules TypeEngine.to_literal(default_value, input_type, lt), and places both the type and converted literal in the upload parameter:

for input_name, (input_type, default_value) in interface.inputs.items():
    if default_value and default_value is not inspect.Parameter.empty:
        lt = TypeEngine.to_literal_type(input_type)
        literal_coros.append(TypeEngine.to_literal(default_value, input_type, lt))
        vars.append((input_name, lt))

literals: List[literals_pb2.Literal] = await asyncio.gather(*literal_coros)

For task inputs, _internal/runtime/convert.py passes the incoming LiteralMap and declared input types to TypeEngine.literal_map_to_kwargs; that path invokes each selected transformer's to_python_value. For outputs, it derives the declared literal type and calls TypeEngine.to_literal for each returned value, as shown above.

Implementing and registering a transformer

When adding a native type, use this contract as the implementation checklist:

• Subclass TypeTransformer[T] and call super().__init__ with the transformer's name and supported Python type.
• Implement get_literal_type to describe the exact Flyte representation.
• Implement both conversion methods as async def; callers await them.
• Treat python_type as the declared type and expected as the expected literal schema; do not replace either with an inferred runtime type.
• Decide whether the base assertion behavior is appropriate. The engine consults type_assertions_enabled; integrations such as DataFrameTransformerEngine can override assert_type.
• Raise TypeTransformerFailedError with useful expected/received details for conversion failures and invalid literal shapes.
• Override guess_python_type if your integration needs literal-type-driven reverse inference.
• Register the transformer with TypeEngine.register, or associate it with another type using TypeEngine.register_additional_type. The registry rejects duplicate registrations; use override=True only when replacement is intentional.
• Override from_binary_idl for a binary format other than the base MESSAGEPACK path.
• Document representation constraints such as FileTransformer's single-part and DirTransformer's multipart requirement.

Keep the surrounding engine rules in view: tuples are rejected as individual Flyte values by TypeEngine.to_literal_checks, non-optional None values are rejected before conversion, and union conversion can fail with an ambiguity TypeError when multiple variants accept the same value. Also remember that this checkout's source contains the implementation and illustrative comments but no matching test, README, or examples files; the contracts above are therefore taken from the actual transformer and runtime code rather than from repository examples.