Core Extensible TypeEngine of Flytekit. This should be used to extend the capabilities of FlyteKits type system. Users can implement their own TypeTransformers and register them with the TypeEngine. This will allow special handling of user objects
Attributes
| Attribute | Type | Description |
|---|
| lazy_import_lock | threading.Lock = threading.Lock() | Thread lock used to prevent race conditions when lazily importing and registering type transformers. |
Constructor
Signature
Methods
register()
@classmethod
def register(
transformer: [TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer),
additional_types: Optional[typing.List[Type]] = None
) - > null
This should be used for all types that respond with the right type annotation when you use type(...) function
Parameters
| Name | Type | Description |
|---|
| transformer | [TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer) | The transformer instance to register for the associated Python type |
| additional_types | Optional[typing.List[Type]] = None | A list of extra Python types that should be handled by this specific transformer |
Returns
register_restricted_type()
@classmethod
def register_restricted_type(
name: str,
type: Type[T]
) - > null
Registers a Python type as restricted and associates it with a RestrictedTypeTransformer to prevent unauthorized usage.
Parameters
| Name | Type | Description |
|---|
| name | str | The logical name for the restricted type |
| type | Type[T] | The Python class or type to be restricted |
Returns
register_additional_type()
@classmethod
def register_additional_type(
transformer: [TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer)[T],
additional_type: Type[T],
override: boolean = False
) - > null
Maps an additional Python type to an existing transformer, optionally overriding any existing registration for that type.
Parameters
| Name | Type | Description |
|---|
| transformer | [TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer)[T] | The transformer to associate with the additional type |
| additional_type | Type[T] | The Python type to add to the registry |
| override | boolean = False | Whether to replace an existing transformer if the type is already registered |
Returns
@classmethod
def get_transformer(
python_type: Type
) - > [TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer)
Implements a recursive search for the transformer.
Parameters
| Name | Type | Description |
|---|
| python_type | Type | The Python type for which a transformer is required |
Returns
| Type | Description |
|---|
[TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer) | The transformer capable of handling the type, falling back to a pickle transformer if no specific match is found |
@classmethod
def lazy_import_transformers() - > null
Only load the transformers if needed.
Returns
to_literal_type()
@classmethod
def to_literal_type(
python_type: Type[T]
) - > LiteralType
Converts a python type into a flyte specific LiteralType
Parameters
| Name | Type | Description |
|---|
| python_type | Type[T] | The Python type to be converted |
Returns
| Type | Description |
|---|
LiteralType | The Flyte IDL representation of the provided Python type |
to_literal_checks()
@classmethod
def to_literal_checks(
python_val: typing.Any,
python_type: Type[T],
expected: LiteralType
) - > null
Validates that a Python value is compatible with the expected Flyte type, checking for unsupported untyped tuples and nullability constraints.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The runtime Python value to validate |
| python_type | Type[T] | The declared Python type hint |
| expected | LiteralType | The target Flyte LiteralType |
Returns
to_literal()
@classmethod
def to_literal(
python_val: typing.Any,
python_type: Type[T],
expected: types_pb2.LiteralType
) - > literals_pb2.Literal
Converts a Python value into a Flyte Literal proto, performing type assertions if enabled by the transformer.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The Python object to convert |
| python_type | Type[T] | The Python type hint used for conversion logic |
| expected | types_pb2.LiteralType | The expected Flyte type schema |
Returns
| Type | Description |
|---|
literals_pb2.Literal | The serialized Flyte Literal object |
unwrap_offloaded_literal()
@classmethod
def unwrap_offloaded_literal(
lv: literals_pb2.Literal
) - > literals_pb2.Literal
Downloads and deserializes a literal from remote storage if it has been offloaded due to size constraints.
Parameters
| Name | Type | Description |
|---|
| lv | literals_pb2.Literal | The literal potentially containing offloaded metadata |
Returns
| Type | Description |
|---|
literals_pb2.Literal | The actual Literal content retrieved from the offloaded URI |
to_python_value()
@classmethod
def to_python_value(
lv: Literal,
expected_python_type: Type
) - > typing.Any
Converts a Literal value with an expected python type into a python value.
Parameters
| Name | Type | Description |
|---|
| lv | Literal | The Flyte Literal to convert |
| expected_python_type | Type | The target Python type for the conversion |
Returns
| Type | Description |
|---|
typing.Any | The native Python representation of the Flyte literal |
to_html()
@classmethod
def to_html(
python_val: typing.Any,
expected_python_type: Type[typing.Any]
) - > str
Generates an HTML representation of a Python value, using either a custom Renderable annotation or the transformer's default HTML logic.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The value to render |
| expected_python_type | Type[typing.Any] | The Python type hint which may contain rendering metadata |
Returns
| Type | Description |
|---|
str | A string containing the HTML representation of the value |
named_tuple_to_variable_map()
@classmethod
def named_tuple_to_variable_map(
t: typing.NamedTuple
) - > interface_pb2.VariableMap
Converts a python-native NamedTuple to a flyte-specific VariableMap of named literals.
Parameters
| Name | Type | Description |
|---|
| t | typing.NamedTuple | The NamedTuple class to inspect |
Returns
| Type | Description |
|---|
interface_pb2.VariableMap | A map where keys are NamedTuple fields and values are Flyte Variables |
literal_map_to_kwargs()
@classmethod
def literal_map_to_kwargs(
lm: LiteralMap,
python_types: typing.Optional[typing.Dict[str, type]] = None,
literal_types: typing.Optional[typing.Dict[str, interface_pb2.Variable]] = None
) - > typing.Dict[str, typing.Any]
Given a LiteralMap (usually an input into a task - intermediate), convert to kwargs for the task
Parameters
| Name | Type | Description |
|---|
| lm | LiteralMap | The map of Flyte literals to convert |
| python_types | typing.Optional[typing.Dict[str, type]] = None | A mapping of argument names to their expected Python types |
| literal_types | typing.Optional[typing.Dict[str, interface_pb2.Variable]] = None | A mapping of argument names to Flyte Variable definitions |
Returns
| Type | Description |
|---|
typing.Dict[str, typing.Any] | A dictionary of Python-native keyword arguments |
dict_to_literal_map()
@classmethod
def dict_to_literal_map(
d: typing.Dict[str, typing.Any],
type_hints: Optional[typing.Dict[str, type]] = None
) - > LiteralMap
Given a dictionary mapping string keys to python values and a dictionary containing guessed types for such string keys, convert to a LiteralMap.
Parameters
| Name | Type | Description |
|---|
| d | typing.Dict[str, typing.Any] | The dictionary of Python values to convert |
| type_hints | Optional[typing.Dict[str, type]] = None | Optional type hints to guide the conversion, especially for generic collections |
Returns
| Type | Description |
|---|
LiteralMap | A Flyte LiteralMap containing the converted values |
@classmethod
def get_available_transformers() - > typing.KeysView[Type]
Returns all python types for which transformers are available
Returns
| Type | Description |
|---|
typing.KeysView[Type] | A view of all registered Python types in the engine |
guess_python_types()
@classmethod
def guess_python_types(
flyte_variable_list: typing.List[interface_pb2.VariableEntry]
) - > typing.Dict[str, Type[Any]]
Transforms a list of flyte-specific VariableEntry objects to a dictionary of regular python values.
Parameters
| Name | Type | Description |
|---|
| flyte_variable_list | typing.List[interface_pb2.VariableEntry] | A list of Flyte variables to be converted to Python types |
Returns
| Type | Description |
|---|
typing.Dict[str, Type[Any]] | A dictionary mapping variable names to their inferred Python types |
guess_python_type()
@classmethod
def guess_python_type(
flyte_type: LiteralType
) - > Type[T]
Transforms a flyte-specific LiteralType to a regular python value.
Parameters
| Name | Type | Description |
|---|
| flyte_type | LiteralType | The Flyte IDL type schema to reverse-engineer |
Returns
| Type | Description |
|---|
Type[T] | The Python type that most closely matches the Flyte LiteralType |