Skip to main content

Type System and Type Engine

Every time you annotate a task parameter with int, List[str], or a @dataclass, flytekit translates that Python type into a Flyte LiteralType and marshals values to and from Flyte Literal objects at execution time. The machinery behind all of this is the type engine — a registry of TypeTransformer implementations that sits in flytekit/core/type_engine.py.

TypeEngine: The Central Registry​

TypeEngine is never instantiated. It operates entirely through class methods, and it owns a class-level dictionary _REGISTRY that maps Python types to TypeTransformer instances. When you call TypeEngine.to_literal_type(int), flytekit looks up the right transformer for int in that registry, calls get_literal_type(int) on it, and returns LiteralType(simple=SimpleType.INTEGER).

The full lifecycle for a single typed value looks like this:

Python type annotation
    → TypeEngine.to_literal_type()   → LiteralType  (used at compile time for interface building)
    → TypeEngine.to_literal()        → Literal       (serialization at task execution)
    → TypeEngine.to_python_value()   → Python value  (deserialization when task receives inputs)

TypeEngine.get_transformer(python_type) implements the lookup logic in layers:

  1. If the type is Annotated[T, ...], it checks whether any annotation is itself a TypeTransformer instance and returns it directly. Otherwise it strips annotations and recurses on T.
  2. If the type is a subclass of enum.Enum, the EnumTransformer is returned — this prevents a class Color(str, Enum) from accidentally resolving to the str transformer.
  3. For generic types like List[int] or Dict[str, int], flytekit checks the origin type (list, dict) against the registry.
  4. For Python 3.10+ str | int union syntax (PEP 604), types.UnionType is matched against the registry.
  5. Then a direct registry lookup is done.
  6. If nothing matches, flytekit walks the MRO of the type and repeats the above for each base class.
  7. If the type is a @dataclass, the shared DataclassTransformer singleton (TypeEngine._DATACLASS_TRANSFORMER) is returned as the final named fallback.
  8. For everything else, FlytePickleTransformer is returned and a warning is logged.

The _register_default_type_transformers() function at the bottom of the module calls TypeEngine.register() for all built-in transformers when the module is first imported. Optional transformers (for pandas, PyTorch, sklearn, pydantic, PIL, numpy) are loaded lazily via TypeEngine.lazy_import_transformers(), which is invoked on every call to get_transformer(). A threading.Lock prevents race conditions during concurrent transformer loading.

Built-in Type Mappings​

Python's primitive types map directly to Flyte's SimpleType variants through module-level SimpleTransformer instances:

Python typeFlyte SimpleTypeTransformer instance
intINTEGERIntTransformer
floatFLOATFloatTransformer
strSTRINGStrTransformer
boolBOOLEANBoolTransformer
datetime.datetimeDATETIMEDatetimeTransformer
datetime.dateDATETIMEDateTransformer
datetime.timedeltaDURATIONTimedeltaTransformer
None / type(None)NONENoneTransformer

SimpleTransformer takes lambda functions for both directions of conversion, so the implementation is compact:

# From flytekit/core/type_engine.py
IntTransformer = SimpleTransformer(
    "int",
    int,
    _type_models.LiteralType(simple=_type_models.SimpleType.INTEGER),
    lambda x: Literal(scalar=Scalar(primitive=Primitive(integer=x))),
    _handle_flyte_console_float_input_to_int,
)

BoolTransformer = SimpleTransformer(
    "bool",
    bool,
    _type_models.LiteralType(simple=_type_models.SimpleType.BOOLEAN),
    lambda x: Literal(scalar=Scalar(primitive=Primitive(boolean=x))),
    lambda x: x.scalar.primitive.boolean,
)

SimpleTransformer.to_literal() applies a strict type check — type(python_val) != self._type raises TypeTransformerFailedError before the lambda is ever called.

You can verify the mapping yourself:

from flytekit.core.type_engine import TypeEngine
from flytekit.models import types as model_types
import typing
from datetime import timedelta

lt = TypeEngine.to_literal_type(int)
assert lt.simple == model_types.SimpleType.INTEGER

# Nested container types compose correctly
lt = TypeEngine.to_literal_type(typing.Dict[str, typing.List[typing.Dict[str, timedelta]]])
assert lt.map_value_type.collection_type.map_value_type.simple == model_types.SimpleType.DURATION

Container Types: List, Dict, and Union​

Lists​

ListTransformer handles typing.List[T] (and the bare list type). It is an AsyncTypeTransformer — its async_to_literal and async_to_python_value methods process list elements concurrently using _run_coros_in_chunks from fsspec. The batch size is controlled by the _F_TE_MAX_COROS environment variable (default 10).

get_literal_type returns a LiteralType(collection_type=<element LiteralType>). The element type must be resolvable — List without a type parameter will fail when the transformer tries to call get_sub_type(t).

from flytekit.core.type_engine import TypeEngine

lt = TypeEngine.to_literal_type(typing.List[int])
assert lt.collection_type.simple == model_types.SimpleType.INTEGER

Dicts​

DictTransformer handles three distinct cases:

  • Dict[str, V] → LiteralType(map_value_type=<V LiteralType>). This uses Flyte's native LiteralMap representation, and keys must be strings.
  • Dict[non-str, V] or bare dict → LiteralType(simple=STRUCT) with a MessagePack annotation. The entire dict is serialized as MessagePack bytes.
  • Annotated[dict, kwtypes(allow_pickle=True)] → same STRUCT representation with a pickle fallback if MessagePack encoding fails.

The extract_types(t) static method is how DictTransformer pulls the key and value types out of a generic dict annotation.

# Dict[str, int] → proper map literal type
lt = TypeEngine.to_literal_type(typing.Dict[str, int])
assert lt.map_value_type.simple == model_types.SimpleType.INTEGER

# Dict[int, str] → STRUCT (MessagePack)
lt = TypeEngine.to_literal_type(typing.Dict[int, str])
assert lt.simple == model_types.SimpleType.STRUCT

Unions​

UnionTransformer handles typing.Union[T1, T2, ...] and — on Python 3.10+ — the pipe syntax T1 | T2. get_literal_type builds a LiteralType(union_type=UnionType(variants=[...])) where each variant carries a TypeStructure(tag=<transformer.name>). That tag is used during deserialization to find the right transformer without guessing.

During serialization, async_to_literal tries each variant in order. If more than one succeeds, it raises TypeError("Ambiguous choice of variant"). During deserialization, the stored tag is used to skip straight to the matching transformer.

import typing
from flytekit.core.type_engine import TypeEngine
from flytekit.core.context_manager import FlyteContextManager
from flytekit.models.types import LiteralType, SimpleType, TypeStructure

pt = typing.Union[str, int]
lt = TypeEngine.to_literal_type(pt)
assert lt.union_type.variants == [
    LiteralType(simple=SimpleType.STRING, structure=TypeStructure(tag="str")),
    LiteralType(simple=SimpleType.INTEGER, structure=TypeStructure(tag="int")),
]

ctx = FlyteContextManager.current_context()
lv = TypeEngine.to_literal(ctx, 3, pt, lt)
assert lv.scalar.union.stored_type.structure.tag == "int"
assert lv.scalar.union.value.scalar.primitive.integer == 3

The PEP 604 pipe syntax produces the same literal type as typing.Union:

# Python 3.10+
pt_604 = str | int
lt_604 = TypeEngine.to_literal_type(pt_604)
assert lt == lt_604

Optional[T] is a special case of Union[T, None]. UnionTransformer.is_optional_type(t) and get_sub_type_in_optional(t) are static helpers used throughout the codebase to detect and unwrap Optional annotations.

Dataclass Transformer​

When a @dataclass type is used as a task input or output, DataclassTransformer handles serialization. Since flytekit 1.14.0, the default format is MessagePack Binary IDL: the dataclass is encoded to MessagePack bytes and wrapped in a Binary scalar with the tag "msgpack". To revert to the old JSON/protobuf Struct format, set the environment variable FLYTE_USE_OLD_DC_FORMAT=true.

The transformer supports three dataclass styles: plain @dataclass, mashumaro's DataClassJSONMixin, and dataclasses_json's DataClassJsonMixin. The serialization path diverges based on which mixin is used:

  • DataClassJSONMixin (mashumaro): calls python_val.to_json(), parses to dict, encodes to MessagePack.
  • Plain @dataclass: uses a cached MessagePackEncoder(python_type) from mashumaro.
  • Legacy JSON format: uses a JSONEncoder(python_type) and wraps the JSON string in a protobuf Struct.
from dataclasses import dataclass
from dataclasses_json import DataClassJsonMixin
from flytekit.core.type_engine import DataclassTransformer
from flytekit.core.context_manager import FlyteContext
import typing

@dataclass
class InnerStruct(DataClassJsonMixin):
    a: int
    b: typing.Optional[str]
    c: typing.List[int]

ctx = FlyteContext.current_context()
tf = DataclassTransformer()
o = InnerStruct(a=5, b=None, c=[1, 2, 3])
lv = tf.to_literal(ctx, o, InnerStruct, tf.get_literal_type(InnerStruct))
ot = tf.to_python_value(ctx, lv=lv, expected_python_type=InnerStruct)
assert ot == o

JSON Schema for UI Autocomplete​

DataclassTransformer.get_literal_type() attempts to extract a JSON Schema from the dataclass using mashumaro.jsonschema.build_json_schema (producing JSON Schema draft 2020-12) or, as a fallback, marshmallow-jsonschema. This schema is stored in the LiteralType.metadata field and surfaces in the Flyte UI for input autocomplete.

The method also builds a TypeStructure.dataclass_type by iterating over the dataclass fields and calling TypeEngine.to_literal_type() on each one — this structure enables FlytePropeller to perform attribute-level access on the serialized value.

Nested Flyte Types in Dataclasses​

If a dataclass field is a FlyteFile, FlyteDirectory, or StructuredDataset, _make_dataclass_serializable() is called before encoding. This method walks the dataclass tree and ensures that string paths are upgraded to the correct Flyte type object so mashumaro can serialize them. Using a raw string like "s3://bucket/file" where a FlyteFile is expected is deprecated and will emit a warning.

Protobuf Integer Precision​

Protobuf's Struct type does not support explicit integers — all numbers are upcast to double. When deserializing from the legacy JSON format, _fix_dataclass_int() walks the dataclass tree and casts float values back to int where the declared type requires it. This is not needed for the MessagePack path, which preserves integer precision natively.

Enum Transformer​

EnumTransformer handles any enum.Enum subclass, but only string-valued enums are supported. An enum with integer values like RED = 1 raises TypeTransformerFailedError when get_literal_type() is called.

from enum import Enum
from flytekit.core.type_engine import TypeEngine
from flytekit.core.context_manager import FlyteContextManager

class Color(Enum):
    RED = "red"
    GREEN = "green"
    BLUE = "blue"

lt = TypeEngine.to_literal_type(Color)
assert lt.enum_type.values == ["red", "green", "blue"]

ctx = FlyteContextManager.current_context()
lv = TypeEngine.to_literal(ctx, Color.RED, Color, lt)
assert lv.scalar.primitive.string_value == "red"

v = TypeEngine.to_python_value(ctx, lv, Color)
assert v == Color.RED

Multi-inheritance enums like class MultiInheritanceColor(str, Enum) are explicitly detected as enums (not strings), which is why the enum check in _get_transformer() happens before the MRO walk.

When guess_python_type() is called with an enum literal type, EnumTransformer synthesizes a new enum.Enum class named "DynamicEnum" with the values from the literal type metadata.

Annotated Types and FlyteAnnotation​

Wrap any type in Annotated[T, FlyteAnnotation({...})] to attach arbitrary metadata to its LiteralType.annotation. This is the standard way to pass Flyte-specific hints (like cache key customizations) alongside a type without changing the type itself.

from typing_extensions import Annotated
from flytekit.core.annotation import FlyteAnnotation
from flytekit.core.type_engine import TypeEngine

lt = TypeEngine.to_literal_type(Annotated[int, FlyteAnnotation({"foo": "bar"})])
assert lt.annotation.annotations == {"foo": "bar"}

TypeEngine.to_literal_type() enforces two constraints:

  • At most one FlyteAnnotation per type hint — a second one raises ValueError.
  • FlyteAnnotation data must not contain the key "cache-key-metadata" (the constant CACHE_KEY_METADATA) — that key is reserved for internal use and raises AssertionError.

FlyteAnnotation is not supported on enums, dicts with FlyteAnnotation applied directly, or dataclasses annotated with FlyteAnnotation — all three raise ValueError in their respective get_literal_type() implementations.

Embedding a Transformer in Annotated​

A more powerful use of Annotated is to place a TypeTransformer instance directly in the annotation metadata. When _get_transformer() inspects the Annotated args and finds a TypeTransformer instance, it returns that transformer — bypassing the registry entirely. This is the mechanism used by the JSONSerialized pattern:

from typing import Dict
from typing_extensions import Annotated, get_args
from flytekit.core.type_engine import TypeEngine, TypeTransformer
from flytekit.models.types import LiteralType, SimpleType
from flytekit.models.literals import Literal, Scalar, Primitive

class JsonTypeTransformer(TypeTransformer[T]):
    LiteralType = LiteralType(simple=SimpleType.STRING)

    def get_literal_type(self, t):
        return self.LiteralType

    def to_literal(self, ctx, python_val, python_type, expected):
        return Literal(scalar=Scalar(primitive=Primitive(string_value=json.dumps(python_val))))

    def to_python_value(self, ctx, lv, expected_python_type):
        return json.loads(lv.scalar.primitive.string_value)

class JSONSerialized:
    def __class_getitem__(cls, item):
        return Annotated[item, JsonTypeTransformer(name=f"json[{item}]", t=item)]

MyJsonDict = JSONSerialized[Dict[str, int]]
assert TypeEngine.to_literal_type(MyJsonDict) == JsonTypeTransformer.LiteralType

Extending the Type System: Custom TypeTransformer​

To add flytekit support for a new Python type, subclass TypeTransformer[T] and implement three abstract methods:

  • get_literal_type(t: Type[T]) -> LiteralType — defines what Flyte type represents the Python type.
  • to_literal(ctx, python_val, python_type, expected) -> Literal — serializes a Python value.
  • to_python_value(ctx, lv, expected_python_type) -> T — deserializes a Literal.

Then call TypeEngine.register(transformer) to make the transformer visible to the registry. Registration happens once at import time, typically at module level.

The PydanticTransformer in flytekit/extras/pydantic_transformer/transformer.py is a complete real-world example:

from pydantic import BaseModel
from flytekit.core.type_engine import TypeEngine, TypeTransformer, TypeTransformerFailedError
from flytekit.models.literals import Binary, Literal, Scalar
from flytekit.models.types import LiteralType, TypeStructure
import msgpack

class PydanticTransformer(TypeTransformer[BaseModel]):
    def __init__(self):
        super().__init__("Pydantic Transformer", BaseModel, enable_type_assertions=False)

    def get_literal_type(self, t: Type[BaseModel]) -> LiteralType:
        schema = t.model_json_schema()
        literal_type = {}
        for name, python_type in t.__annotations__.items():
            try:
                literal_type[name] = TypeEngine.to_literal_type(python_type)
            except Exception as e:
                logger.warning(...)
        ts = TypeStructure(tag="", dataclass_type=literal_type)
        return types.LiteralType(
            simple=types.SimpleType.STRUCT,
            metadata=schema,
            structure=ts,
            annotation=TypeAnnotationModel({CACHE_KEY_METADATA: {SERIALIZATION_FORMAT: MESSAGEPACK}}),
        )

    def to_literal(self, ctx, python_val, python_type, expected):
        json_str = python_val.model_dump_json()
        dict_obj = json.loads(json_str)
        msgpack_bytes = msgpack.dumps(dict_obj)
        return Literal(scalar=Scalar(binary=Binary(value=msgpack_bytes, tag=MESSAGEPACK)))

    def to_python_value(self, ctx, lv, expected_python_type):
        if lv and lv.scalar and lv.scalar.binary is not None:
            return self.from_binary_idl(lv.scalar.binary, expected_python_type)
        json_str = _json_format.MessageToJson(lv.scalar.generic)
        return expected_python_type.model_validate_json(json_str, strict=False)

TypeEngine.register(PydanticTransformer())

SimpleTransformer for Lightweight Types​

For types that map to a single Flyte primitive without complex logic, SimpleTransformer eliminates boilerplate by accepting the conversion lambdas directly in its constructor:

from flytekit.core.type_engine import SimpleTransformer, TypeEngine
from flytekit.models.literals import Literal, Scalar, Primitive
from flytekit.models.types import LiteralType, SimpleType

class MyInt:
    def __init__(self, x: int):
        self.val = x

TypeEngine.register(
    SimpleTransformer(
        "MyInt",
        MyInt,
        LiteralType(simple=SimpleType.INTEGER),
        lambda x: Literal(scalar=Scalar(primitive=Primitive(integer=x.val))),
        lambda x: MyInt(x.scalar.primitive.integer),
    )
)

AsyncTypeTransformer for Async Serialization​

When serialization involves I/O (uploading files, reading from object storage), subclass AsyncTypeTransformer[T] instead. Implement async_to_literal and async_to_python_value. The sync to_literal / to_python_value methods are automatically provided via loop_manager.synced(), so the transformer works in both sync and async call sites. Both ListTransformer and DictTransformer use this pattern.

MRO-Based Inheritance Lookup​

If you register a transformer for a base class, it will be found for all subclasses via the MRO walk in get_transformer():

@dataclass
class ParentDC:
    ...

@dataclass
class ChildDC(ParentDC):
    ...

class ParentDCTransformer(TypeTransformer[ParentDC]):
    def __init__(self):
        super().__init__("ParentDC Transformer", ParentDC)

TypeEngine.register(ParentDCTransformer())

# ChildDC is resolved to the same transformer as ParentDC via MRO
assert TypeEngine.get_transformer(ChildDC) == TypeEngine.get_transformer(ParentDC)

Note that TypeEngine.register() raises ValueError if a transformer for the type is already registered. There is no public unregister method — tests that need to clean up use del TypeEngine._REGISTRY[MyType] directly. Use register_additional_type(transformer, type, override=True) if you need to silently override an existing registration for an additional type alias.

Restricted Types​

RestrictedTypeTransformer marks a Python type as not usable as a task input or output. All three of its methods (get_literal_type, to_literal, to_python_value) raise RestrictedTypeError. tuple, typing.Tuple, and NamedTuple are registered as restricted types at startup. The comment in the registration code notes that Flyte IDL could support tuples as structs but does not do so presently.

LiteralsResolver​

When working with execution outputs from FlyteRemote, the raw data arrives as a LiteralMap — a dictionary of Literal objects. LiteralsResolver is a UserDict subclass that wraps that map and lazily converts values to Python using TypeEngine.to_python_value().

from flytekit.core.type_engine import TypeEngine, LiteralsResolver
from flytekit.core.context_manager import FlyteContext
from dataclasses import dataclass
from dataclasses_json import DataClassJsonMixin

@dataclass
class Foo(DataClassJsonMixin):
    x: int
    y: str
    z: typing.Dict[str, int]

lt = TypeEngine.to_literal_type(Foo)
foo = Foo(1, "hello", {"world": 3})
lv = TypeEngine.to_literal(FlyteContext.current_context(), foo, Foo, lt)

lr = LiteralsResolver({"a": lv})
assert lr.get("a", Foo) == foo

get(attr, as_type=None) is the main access method. If as_type is omitted, it tries to guess the Python type from the variable_map passed to the constructor (which corresponds to one side of a Flyte TypedInterface). The converted value is cached in _native_values, so repeated calls are free.

as_python_native(python_interface) converts the full literal map to a single value or a tuple, matching the positional order of the interface's output variables. This is how flytekit unpacks multi-output task results when executing locally or fetching from the remote.

If you have a LiteralsResolver returned by FlyteRemote without type hints embedded in the variable_map, call update_type_hints({"output": MyType}) before accessing values, or always pass as_type explicitly to get().

Important Constraints and Edge Cases​

Enum values must be strings. EnumTransformer raises TypeTransformerFailedError for any enum whose first value is not a str. Integer-valued enums like RED = 1 are not supported.

Dict keys must be strings for native map representation. DictTransformer only produces a LiteralType(map_value_type=...) when the key type is str. Any other key type falls back to MessagePack STRUCT encoding, and the keys are no longer individually addressable via FlytePropeller attribute access.

Union ambiguity is a runtime error. When multiple union variants can serialize the same value — for instance, a Union[int, MyInt] where both transformers accept a plain integer — UnionTransformer.async_to_literal() raises TypeError("Ambiguous choice of variant"). This shows up at serialization time, not compile time. When deserializing a literal that carries a stored_type tag, the tag is used to skip ambiguous candidates, so round-tripping works as long as the tag was recorded correctly during serialization.

tuple and NamedTuple are restricted. These are registered via register_restricted_type() at startup and will raise RestrictedTypeError if used as task inputs or outputs. A task that needs to return multiple values should use a NamedTuple at the task signature level (flytekit unpacks it to a VariableMap), but cannot accept one as an input.

TextIOTransformer and BinaryIOTransformer are read-only. Both transformers' to_literal() methods raise NotImplementedError. They can be used only as task input types (the transformer downloads the blob file and opens it), not as output types.

Unknown types fall back to pickle. If get_transformer() finds no match after exhausting the registry, MRO walk, and dataclass check, it returns a FlytePickleTransformer and logs a warning via display_pickle_warning(). This means any Python object can technically be passed between tasks, but at the cost of portability and type safety.

DataclassTransformer is a shared singleton. TypeEngine._DATACLASS_TRANSFORMER is one instance shared for all @dataclass types. Its MessagePackEncoder/Decoder and JSONEncoder/Decoder caches are keyed by Python type, so they accumulate across the lifetime of the process.