Skip to main content

Restricting Types from I/O

The interface boundary​

A Python object used inside a task is not automatically a valid task or workflow input or output. Flyte's type system represents interface values as LiteralType metadata and runtime values as Literal protobufs; the flyte.types package documents these as the universal, serializable representations used when data crosses process, container, and language boundaries. TypeEngine is the dispatch point that converts between Python values and those representations.

That boundary matters for execution-only structures. If a type cannot be represented by Flyte's literal model, registering an ordinary transformer—or allowing the eventual pickle fallback—would make it appear usable in an interface. RestrictedTypeTransformer provides the opposite behavior: it is a registry entry whose conversion operations always fail.

Registering a restricted type​

TypeEngine.register_restricted_type accepts a display name and a Python type:

TypeEngine.register_restricted_type(
    "execution-only value",
    ExecutionOnlyValue,
)

The method in types/_type_engine.py performs two operations:

@classmethod
def register_restricted_type(
    cls,
    name: str,
    type: Type[T],
):
    cls._RESTRICTED_TYPES.append(type)
    cls.register(RestrictedTypeTransformer(name, type))  # type: ignore

The type is recorded in TypeEngine._RESTRICTED_TYPES, and a RestrictedTypeTransformer is installed in the normal TypeEngine._REGISTRY. This is important because restriction is enforced through the same transformer lookup used by ordinary types, rather than through a separate special case in each task-runtime conversion path.

For example, an application-defined execution-only class can be registered as follows:

from flyte.types import TypeEngine


class ExecutionOnlyValue:
    pass


TypeEngine.register_restricted_type("execution-only value", ExecutionOnlyValue)

TypeEngine.register rejects an already occupied registry key with ValueError. Consequently, calling register_restricted_type twice for the same type in one process first appends the type to _RESTRICTED_TYPES, then fails when register detects the duplicate; the restricted-type API has no override argument. Registration is process-global because both registry structures are class variables.

What the restricted transformer rejects​

RestrictedTypeTransformer is defined in types/_type_engine.py and inherits from TypeTransformer. Its constructor passes the supplied name and Python type to the base class, but its three conversion methods are deliberately non-functional:

class RestrictedTypeTransformer(TypeTransformer[T], ABC):
    """
    Types registered with the RestrictedTypeTransformer are not allowed to be converted to and from literals.
     In other words,
    Restricted types are not allowed to be used as inputs or outputs of tasks and workflows.
    """

    def __init__(self, name: str, t: Type[T]):
        super().__init__(name, t)

    def get_literal_type(self, t: Optional[Type[T]] = None) -> LiteralType:
        raise RestrictedTypeError

    async def to_literal(self, python_val: T, python_type: Type[T], expected: LiteralType) -> Literal:
        raise RestrictedTypeError

    async def to_python_value(self, lv: Literal, expected_python_type: Type[T]) -> T:
        raise RestrictedTypeError

The three failure points cover both metadata and data conversion:

  • get_literal_type fails when Flyte needs to construct the type description for an interface.
  • to_literal fails when a Python input, default, or returned output would be serialized.
  • to_python_value fails when a Literal would be reconstructed as the restricted Python type.

All three raise RestrictedTypeError with the same message shape: Transformer for type <python_type> is restricted currently. RestrictedTypeError derives directly from Exception; it is not a TypeTransformerFailedError.

Built-in restrictions: tuples and nested named tuples​

When the default transformers are initialized during import of types._type_engine, flyte-sdk registers these three restrictions:

TypeEngine.register_restricted_type("non typed tuple", tuple)
TypeEngine.register_restricted_type("non typed tuple", typing.Tuple)
TypeEngine.register_restricted_type("named tuple", NamedTuple)

The surrounding source comment gives the intended boundary. Tuple values are not currently supported as individual Flyte values, and typing.Tuple is also unsupported even though its element types are inspectable. The typing.NamedTuple registration is nuanced: a task's return signature may be a top-level named tuple describing multiple outputs, but a named tuple nested inside another task value is not treated as a serializable value. The comment also notes that tuples could be represented as structs, but flyte-sdk does not do so here.

That distinction is visible in the interface examples in _interface.py:

import typing


nt1 = typing.NamedTuple("NT1", x_str=str, y_int=int)


def t(a: int, b: str) -> nt1:
    ...


def u(a: int, b: str) -> typing.Tuple[int, str]:
    ...

These annotations describe multiple task outputs. They should not be confused with passing one tuple or nested named-tuple object as an individual input or output value. TypeEngine.to_literal_checks also has a separate runtime guard that raises AssertionError for any tuple value and advises returning a named field such as v.x instead of returning an inner named tuple.

TypeEngine.get_transformer checks exact registrations and generic origins before falling back to dataclass handling or FlytePickleTransformer. Thus a restricted registration is found before the pickle fallback. In particular, a parameterized tuple type can resolve through its registered tuple origin, so the fallback does not turn a restricted tuple into a pickle value.

Where the restriction is enforced​

Interface construction​

_internal/runtime/types_serde.py converts a native interface into a Flyte IDL TypedInterface. Each input and output annotation eventually reaches transform_type:

def transform_type(x: type, description: Optional[str] = None) -> interface_pb2.Variable:
    # add artifact handling eventually
    return interface_pb2.Variable(
        type=TypeEngine.to_literal_type(x),
        description=description,
    )

Because TypeEngine.to_literal_type selects the registered transformer and calls get_literal_type, a restricted annotation raises RestrictedTypeError while the typed interface is being built. The interface therefore cannot be emitted with that restricted type as an input or output type.

Task input deserialization​

At runtime, _internal/runtime/convert.py collects named input literals and passes them, together with the native Python input types, to TypeEngine.literal_map_to_kwargs:

async def convert_inputs_to_native(inputs: Inputs, python_interface: NativeInterface) -> Dict[str, Any]:
    literals = {named_literal.name: named_literal.value for named_literal in inputs.proto_inputs.literals}
    native_vals = await TypeEngine.literal_map_to_kwargs(
        literals_pb2.LiteralMap(literals=literals), python_interface.get_input_types()
    )
    return native_vals

literal_map_to_kwargs calls TypeEngine.to_python_value for each declared type. If that type resolves to a restricted transformer, the literal cannot be reconstructed as a native task argument. The same mechanism is used by convert_outputs_to_native for deserializing task outputs:

kwargs = await TypeEngine.literal_map_to_kwargs(lm, interface.outputs)

Task output serialization​

convert_from_native_to_outputs obtains the declared output type and converts the returned value to a literal:

async def convert_from_native_to_outputs(o: Any, interface: NativeInterface, task_name: str = "") -> Outputs:
    # Always make it a tuple even if it's just one item to simplify logic below
    if not isinstance(o, tuple):
        o = (o,)

    # ... output-count validation above this loop ...
    named = []
    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)

    return Outputs(proto_outputs=run_definition_pb2.Outputs(literals=named))

A restricted output can fail in either to_literal_type or to_literal, so no output literal is materialized. There is also an exception-handling edge case: this function catches only TypeTransformerFailedError. Since RestrictedTypeError inherits directly from Exception, it propagates as RestrictedTypeError rather than being wrapped in RuntimeDataValidationError by this except block.

The same TypeEngine.to_literal_type and TypeEngine.to_literal operations are used when _internal/runtime/convert.py serializes default inputs. Restriction therefore applies to that interface-adjacent path as well, not only to values returned from task bodies.

Practical guidance​

Use TypeEngine.register_restricted_type when a Python type is valid only inside task execution and must not be represented in the task or workflow interface. Register it once during process setup, before code constructs interfaces or converts values. Do not register a second transformer for the same type: TypeEngine.register intentionally rejects duplicate keys, while register_additional_type is a separate API for associating an existing transformer with another type and optionally overriding that association.

The public flyte.types exports include TypeEngine, but not RestrictedTypeError or RestrictedTypeTransformer. Those symbols are defined in the private module flyte.types._type_engine; normal application code generally needs only the public registration method:

from flyte.types import TypeEngine


class TaskLocalResource:
    pass


TypeEngine.register_restricted_type("task-local resource", TaskLocalResource)

The built-in tuple registrations are activated automatically as part of default transformer initialization when types._type_engine is imported. No environment variable enables them. _F_TE_MAX_COROS affects batching for concurrent collection conversions in TypeEngine, but does not enable, disable, or alter restricted-type enforcement.