Skip to main content

Guide to Built-in Transformers

When passing structured objects, collections, and domain models between tasks in Flyte workflows, raw Python objects must cross language-agnostic boundaries. The flyte-sdk type system uses dedicated TypeTransformer classes registered in TypeEngine (types._type_engine.py) to convert native Python types into Flyte IDL LiteralType metadata and serialize instance values into Flyte Literal representations.

This guide explains how flyte-sdk handles standard dataclasses, Pydantic models, string enums, lists, dictionaries, union variants, and Protobuf messages.

Dataclasses​

When passing user-defined data structures across tasks, decorating a Python class with @dataclass allows flyte-sdk to validate types, generate JSON schema definitions, and serialize data efficiently.

from dataclasses import dataclass
from flyte import task, workflow

@dataclass
class ModelConfig:
    batch_size: int
    learning_rate: float
    model_name: str = "resnet50"

@task
def train(config: ModelConfig) -> float:
    return config.learning_rate * config.batch_size

@workflow
def training_workflow(config: ModelConfig) -> float:
    return train(config=config)

Internal Mechanism​

Dataclass serialization and deserialization is implemented by DataclassTransformer in types/_type_engine.py:

  1. Schema Generation: DataclassTransformer.get_literal_type() extracts JSON Schema (Draft 2020-12) metadata using mashumaro.jsonschema.build_json_schema(..., plugins=[PydanticSchemaPlugin()]) and wraps it in a Flyte IDL LiteralType(simple=SimpleType.STRUCT, metadata=schema).
  2. Serialization: DataclassTransformer.to_literal() converts the dataclass instance into MessagePack binary bytes using a cached mashumaro MessagePackEncoder. The serialized payload is stored in a binary scalar Literal(scalar=Scalar(binary=Binary(value=msgpack_bytes, tag="msgpack"))).
  3. Deserialization: DataclassTransformer.to_python_value() reconstructs the Python dataclass from the MessagePack binary literal using mashumaro's MessagePackDecoder or DataClassJSONMixin.from_json if subclassed.
  4. Dynamic Reconstruction: When flyte-sdk inspects a workflow output or literal from remote execution via guess_python_type(), convert_mashumaro_json_schema_to_python_class() dynamically synthesizes a dataclass type matching the embedded JSON Schema.
Python @dataclass
       │
       ▼ (mashumaro MessagePackEncoder)
MessagePack Bytes (tag="msgpack")
       │
       ▼
Flyte Literal(scalar=Scalar(binary=Binary(...)))

Note on Custom Types in Dataclasses: If a dataclass contains complex custom nested types (such as FlyteFile or StructuredDataset), those types must inherit from mashumaro.types.SerializableType and implement _serialize and _deserialize methods.

Pydantic Models​

For models requiring field validation, constraint checking, or integration with external validation logic, flyte-sdk provides native support for Pydantic V2 BaseModel types via PydanticTransformer.

from pydantic import BaseModel, Field
from flyte import task, workflow

class PipelineParams(BaseModel):
    dataset_name: str
    epochs: int = Field(gt=0, le=100)
    validation_split: float = Field(default=0.2, ge=0.0, le=1.0)

@task
def run_pipeline(params: PipelineParams) -> str:
    return f"Trained on {params.dataset_name} for {params.epochs} epochs"

@workflow
def params_workflow(params: PipelineParams) -> str:
    return run_pipeline(params=params)

Internal Mechanism​

PydanticTransformer in types/_type_engine.py manages BaseModel serialization without requiring external mixins:

  1. Literal Type: get_literal_type() invokes t.model_json_schema() to populate the metadata dictionary of a SimpleType.STRUCT literal type.
  2. MessagePack Conversion: to_literal() dumps the model to JSON via python_val.model_dump_json(), converts the parsed dictionary into MessagePack bytes (msgpack.dumps(...)), and produces a binary scalar tagged as "msgpack".
  3. Validation on Deserialization: In from_binary_idl() and to_python_value(), flyte-sdk loads the MessagePack payload and calls expected_python_type.model_validate_json(json_str, strict=False, context={"deserialize": True}).

String Enums​

When restricting task inputs or outputs to fixed categorical options, inherit from enum.Enum (or str, Enum).

from enum import Enum
from flyte import task, workflow

class Stage(str, Enum):
    DEV = "dev"
    STAGING = "staging"
    PROD = "prod"

@task
def deploy(stage: Stage) -> str:
    return f"Deploying to {stage.value}"

@workflow
def deploy_workflow(stage: Stage) -> str:
    return deploy(stage=stage)

Constraints and MRO Handling​

EnumTransformer in types/_type_engine.py processes enumeration types:

  • String Values Required: Every member of the enum must have a string value (isinstance(values[0], str)). Numeric or composite enum values will trigger a TypeTransformerFailedError("Only EnumTypes with value of string are supported").
  • Literal Representation: get_literal_type() maps the enum to LiteralType(enum_type=types_pb2.EnumType(values=[v.value for v in t])). Values serialize into primitive string literals: Literal(scalar=Scalar(primitive=Primitive(string_value=python_val.value))).
  • TypeEngine Precedence: In TypeEngine._get_transformer(), an explicit check intercepts enum classes:
if inspect.isclass(python_type) and issubclass(python_type, enum.Enum):
    # Special case: prevent that for a type `FooEnum(str, Enum)`, the str transformer is used.
    return cls._ENUM_TRANSFORMER

This prevents FooEnum(str, Enum) from falling back to StrTransformer during Python MRO evaluation.

Collections: Lists and Dictionaries​

flyte-sdk provides distinct serialization strategies for univariate generic lists and key-value dictionaries.

Univariate Lists​

ListTransformer in types/_type_engine.py handles generic lists (typing.List[T] or list[T]).

from typing import List
from flyte import task, workflow

@task
def compute_averages(values: List[float]) -> float:
    return sum(values) / len(values) if values else 0.0

@workflow
def list_workflow(values: List[float]) -> float:
    return compute_averages(values=values)
  • Univariate Constraint: Untyped raw list declarations or multi-type collections (such as tuples) are rejected. ListTransformer.get_sub_type() inspects __args__[0] or Annotated wrappers to determine element type T.
  • Literal Mapping: Generates a LiteralType(collection_type=sub_type) and stores outputs as Literal(collection=LiteralCollection(literals=lit_list)).
  • Concurrent Chunking: Elements in the list are serialized concurrently using chunked coroutines (_run_coros_in_chunks).

Dictionaries: Typed Maps vs. Binary Structs​

DictTransformer in types/_type_engine.py uses dual serialization pathways depending on dictionary key types:

from typing import Dict
from flyte import task, workflow

@task
def process_metrics(named_scores: Dict[str, float], indexed_counts: Dict[int, int]) -> int:
    return len(named_scores) + len(indexed_counts)

@workflow
def dict_workflow(named_scores: Dict[str, float], indexed_counts: Dict[int, int]) -> int:
    return process_metrics(named_scores=named_scores, indexed_counts=indexed_counts)
Dictionary TypeIDL Literal TypeFlyte Storage Representation
Dict[str, V]LiteralType(map_value_type=sub_type)Literal(map=LiteralMap(literals=lit_map))
Dict[int, V], untyped dict, or complex keysLiteralType(simple=SimpleType.STRUCT)Literal(scalar=Scalar(binary=Binary(value=msgpack_bytes, tag="msgpack")))
  1. String-Keyed Dictionaries (Dict[str, V]): DictTransformer.get_literal_type() creates a map_value_type. Keys must be strings; values are recursively transformed into literals and stored inside a LiteralMap.
  2. Non-String or Untyped Dictionaries: Flyte IDL LiteralMap only supports string keys. If keys are non-strings (such as Dict[int, str]), DictTransformer serializes the entire dictionary into MessagePack binary bytes.
  3. Pickle Fallback: If an annotated dictionary contains {"allow_pickle": True} in its metadata and MessagePack encoding fails with a TypeError, DictTransformer.dict_to_binary_literal() offloads the object via FlytePickle.to_pickle(v) and wraps the remote path in a generic Struct scalar.

Unions and Optionals​

flyte-sdk supports multi-type inputs and outputs through typing.Union and PEP 604 union syntax (T1 | T2), including Optional[T].

from typing import Union, Optional
from flyte import task, workflow

@task
def parse_identifier(val: Union[int, str]) -> str:
    return f"ID: {val}"

@task
def handle_optional(val: Optional[str]) -> str:
    return val if val is not None else "default"

@workflow
def union_workflow(val: int | str, opt: str | None = None) -> tuple[str, str]:
    return parse_identifier(val=val), handle_optional(val=opt)

Internal Mechanism & Ambiguity Detection​

UnionTransformer in types/_type_engine.py processes union types:

  1. Type Tagging: UnionTransformer.get_literal_type() maps each variant to a tagged LiteralType(union_type=UnionType(variants=[...])).
  2. Direct Match Fast Path: When to_literal() runs, it checks whether type(python_val) matches one of the declared union subtypes directly:
if inferred_type in subtypes:
    transformer = TypeEngine.get_transformer(inferred_type)
    res = await transformer.to_literal(
        python_val, inferred_type, expected.union_type.variants[subtypes.index(inferred_type)]
    )
    res_type = _add_tag_to_type(transformer.get_literal_type(inferred_type), transformer.name)
    return Literal(scalar=Scalar(union=Union(value=res, type=res_type)))
  1. Polymorphic Search & Structural Ambiguity: If an exact type match is not found, UnionTransformer iterates through the union variants and attempts conversion. If more than one transformer succeeds, flyte-sdk raises a TypeError to prevent ambiguous deserialization:
if is_ambiguous:
    raise TypeError(
        f"Ambiguous choice of variant for union type.\n"
        f"Potential types: {potential_types}\n"
        "These types are structurally the same, because it's attributes have the same names and associated types."
    )

Protobuf Messages​

Workflows that interface with gRPC services or schema definitions defined in .proto files can use google.protobuf.message.Message instances or google.protobuf.struct_pb2.ListValue directly.

from google.protobuf import struct_pb2
from flyte import task, workflow

@task
def transform_struct(data: struct_pb2.Struct) -> struct_pb2.Struct:
    data.fields["status"].string_value = "processed"
    return data

@workflow
def proto_workflow(data: struct_pb2.Struct) -> struct_pb2.Struct:
    return transform_struct(data=data)

Internal Mechanism​

ProtobufTransformer in types/_type_engine.py converts protobuf messages into Flyte generic structs:

  1. Literal Type: ProtobufTransformer.get_literal_type() returns LiteralType(simple=SimpleType.STRUCT, metadata={"pb_type": f"{cls.__module__}.{cls.__name__}"}).
  2. Serialization:
    • struct_pb2.ListValue: Converted into a LiteralCollection of individual literals.
    • Message instances: Converted via protobuf dictionary utilities _MessageToDict(python_val) and saved as Literal(scalar=Scalar(generic=struct)).
  3. Deserialization: ProtobufTransformer.to_python_value() instantiates the target protobuf class (expected_python_type()), translates the Flyte generic struct to a dict via _MessageToDict, and populates the message using _ParseDict.

Async Coroutine Batching Configuration​

When transforming large collections of elements in ListTransformer and DictTransformer, flyte-sdk converts element literals asynchronously using _run_coros_in_chunks.

The chunk size for concurrent coroutines is controlled by the _F_TE_MAX_COROS environment variable:

# Set maximum concurrent coroutines during type engine transformation passes
export _F_TE_MAX_COROS=20

In types/_type_engine.py:

_TYPE_ENGINE_COROS_BATCH_SIZE = int(os.environ.get("_F_TE_MAX_COROS", "10"))

Increasing _F_TE_MAX_COROS allows higher concurrency when converting collections containing types that perform asynchronous operations (such as remote file uploads in FlyteFile or dataset metadata extraction).