How to Create a Custom Type Transformer

Flyte uses a sophisticated type system to bridge the gap between Python's dynamic typing and the strongly-typed Flyte IDL. When you need to support a custom Python type that isn't natively handled by flyte-sdk, you must implement a TypeTransformer and register it with the TypeEngine.

Implement a Simple Transformer

For basic types that map directly to Flyte primitives (like int, str, or bool), use the SimpleTransformer class. This avoids the boilerplate of a full class implementation by using lambdas for conversion.

The following example shows how flyte-sdk implements the IntTransformer in types/_type_engine.py:

from flyte.models import types as types_pb2
from flyte.models.literals import Literal, Scalar, Primitive
from flyte.types._type_engine import SimpleTransformer, TypeEngine

# 1. Define the transformer
IntTransformer = SimpleTransformer(
    "int",
    int,
    types_pb2.LiteralType(simple=types_pb2.SimpleType.INTEGER),
    to_literal_transformer=lambda x: Literal(scalar=Scalar(primitive=Primitive(integer=x))),
    from_literal_transformer=lambda x: x.scalar.primitive.integer,
)

# 2. Register it with the TypeEngine
TypeEngine.register(IntTransformer)

Implement a Custom Type Transformer

For complex objects, inherit from TypeTransformer[T]. You must implement three core methods:

1. get_literal_type: Defines how the Python type maps to a Flyte LiteralType.
2. to_literal: Converts a Python object instance into a Flyte Literal.
3. to_python_value: Converts a Flyte Literal back into a Python object.

Consider a custom MyUser class that you want to pass between tasks:

Unverified example

This example could not be verified against this version of the codebase and may not work as shown. Validator finding: 'MyUser' has no member 'name' 'MyUser' has no member 'age'

import typing
import json
from flyte.models import types as types_pb2
from flyte.models.literals import Literal, Scalar, Binary
from flyte.types._type_engine import TypeTransformer, TypeEngine

class MyUser:
    name: str
    age: int
    def __init__(self, name: str, age: int):
        self.name = name
        self.age = age

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

    def get_literal_type(self, t: typing.Type[MyUser]) -> types_pb2.LiteralType:
        # Map this to a binary type for custom serialization
        return types_pb2.LiteralType(simple=types_pb2.SimpleType.BINARY, tag="my-user-json")

    async def to_literal(self, python_val: MyUser, python_type: typing.Type[MyUser], expected: types_pb2.LiteralType) -> Literal:
        # Serialize the object to bytes
        data = json.dumps({"name": python_val.name, "age": python_val.age}).encode("utf-8")
        return Literal(scalar=Scalar(binary=Binary(value=data, tag="my-user-json")))

    async def to_python_value(self, lv: Literal, expected_python_type: typing.Type[MyUser]) -> MyUser:
        # Deserialize bytes back to the Python object
        data = json.loads(lv.scalar.binary.value.decode("utf-8"))
        return MyUser(name=data["name"], age=data["age"])

    def guess_python_type(self, literal_type: types_pb2.LiteralType) -> typing.Type[MyUser]:
        if literal_type.simple == types_pb2.SimpleType.BINARY and literal_type.tag == "my-user-json":
            return MyUser
        raise ValueError("Cannot reverse literal type")

# Register the transformer
TypeEngine.register(MyUserTransformer())

Registering Additional Types

If a single transformer can handle multiple related Python types (e.g., a base class and its subclasses), use the additional_types parameter during registration. The TypeEngine uses these to populate its internal _REGISTRY.

# Registering a transformer for a type and its alias
TypeEngine.register(MyUserTransformer(), additional_types=[typing.NewType("SpecialUser", MyUser)])

Handling Dataframe Types

If you are implementing support for a new dataframe library, do not register directly with TypeEngine. Instead, flyte-sdk uses a specialized DataFrameTransformerEngine found in io/_dataframe/dataframe.py. You should register your handler with that engine to ensure it integrates with Flyte's structured dataset features.

Troubleshooting Type Mismatches

The TypeEngine performs validation during conversion. If type_assertions_enabled is True (the default in TypeTransformer), the engine calls assert_type before to_literal.

If you encounter a TypeTransformerFailedError, verify:

• The python_type passed to to_literal matches the type your transformer is registered for.
• Your guess_python_type implementation correctly identifies the LiteralType tags you defined in get_literal_type.
• For SimpleTransformer, ensure the input value exactly matches the type specified in the constructor, as it performs a strict type(python_val) is not self._type check.