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 | A threading.Lock instance used to prevent race conditions during the lazy loading of transformers. |
Methods
register()
@classmethod
def register(
transformer: [TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer),
additional_types: Optional[typing.List[Type]] = None
)
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 TypeTransformer instance to register for handling specific Python types. |
| additional_types | Optional[typing.List[Type]] = None | An optional list of additional Python types that this transformer should also handle. |
register_restricted_type()
@classmethod
def register_restricted_type(
name: str,
type: Type[T]
)
Registers a type as restricted, meaning it cannot be directly used in certain contexts, and associates it with a RestrictedTypeTransformer.
Parameters
| Name | Type | Description |
|---|
| name | str | The name to associate with the restricted type. |
| type | Type[T] | The Python type to register as restricted. |
register_additional_type()
@classmethod
def register_additional_type(
transformer: [TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer)[T],
additional_type: Type[T],
override: bool = False
)
Registers an additional Python type to be handled by an existing TypeTransformer, optionally overriding a previous registration.
Parameters
| Name | Type | Description |
|---|
| transformer | [TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer)[T] | The TypeTransformer instance that will handle the additional type. |
| additional_type | Type[T] | The Python type to associate with the provided transformer. |
| override | bool = False | A boolean indicating whether to override an existing transformer registration for the additional_type. |
@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 to retrieve a transformer. |
Returns
| Type | Description |
|---|
[TypeTransformer](typetransformer.md?sid=flyte_types__type_engine_typetransformer) | The TypeTransformer instance capable of handling the python_type. |
@classmethod
def lazy_import_transformers()
Loads transformers from plugins and other sources only when they are needed, preventing unnecessary imports and potential circular dependencies.
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 convert. |
Returns
| Type | Description |
|---|
LiteralType | The Flyte LiteralType representation of the given Python type. |
to_literal_checks()
@classmethod
def to_literal_checks(
python_val: typing.Any,
python_type: Type[T],
expected: LiteralType
)
Performs type checks and validations before converting a Python value to a Flyte literal, ensuring type compatibility and handling unsupported types like untyped tuples.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The Python value to check. |
| python_type | Type[T] | The expected Python type of the value. |
| expected | LiteralType | The expected Flyte LiteralType. |
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 of a specified Python type into a Flyte Literal representation, applying type assertions and handling offloaded literals.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The Python value to convert. |
| python_type | Type[T] | The Python type of the value. |
| expected | types_pb2.LiteralType | The expected Flyte LiteralType for the conversion. |
Returns
| Type | Description |
|---|
literals_pb2.Literal | The Flyte Literal representation of the Python value. |
unwrap_offloaded_literal()
@classmethod
def unwrap_offloaded_literal(
lv: literals_pb2.Literal
) - > literals_pb2.Literal
Retrieves the actual Literal content from an offloaded literal by downloading it from its URI.
Parameters
| Name | Type | Description |
|---|
| lv | literals_pb2.Literal | The Literal object, potentially containing offloaded metadata. |
Returns
| Type | Description |
|---|
literals_pb2.Literal | The fully unwrapped Literal object. |
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 value to convert. |
| expected_python_type | Type | The expected Python type for the converted value. |
Returns
| Type | Description |
|---|
typing.Any | The Python value converted from the Flyte Literal. |
to_html()
@classmethod
def to_html(
python_val: typing.Any,
expected_python_type: Type[typing.Any]
) - > str
Converts a Python value to an HTML string representation, utilizing the appropriate transformer or a custom renderer if available.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The Python value to convert to HTML. |
| expected_python_type | Type[typing.Any] | The expected Python type of the value, which may include rendering annotations. |
Returns
| Type | Description |
|---|
str | An HTML string representation of the Python 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 Python NamedTuple to convert. |
Returns
| Type | Description |
|---|
interface_pb2.VariableMap | A Flyte VariableMap representing the structure and types of the NamedTuple. |
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 Flyte LiteralMap containing the input values. |
| python_types | typing.Optional[typing.Dict[str, type]] = None | An optional dictionary mapping input names to their expected Python types. |
| literal_types | typing.Optional[typing.Dict[str, interface_pb2.Variable]] = None | An optional dictionary mapping input names to their Flyte Variable definitions, used to infer Python types if python_types is not provided. |
Returns
| Type | Description |
|---|
typing.Dict[str, typing.Any] | A dictionary of keyword arguments, where keys are string names and values are converted Python objects. |
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 input dictionary with string keys and Python values. |
| type_hints | Optional[typing.Dict[str, type]] = None | An optional dictionary providing explicit Python type hints for the values in d, which take precedence over inferred types. |
Returns
| Type | Description |
|---|
LiteralMap | A Flyte LiteralMap representing the input dictionary, with values converted to Flyte Literal objects. |
@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 the Python types for which transformers are currently registered. |
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 VariableEntry objects, each containing a variable name and its LiteralType. |
Returns
| Type | Description |
|---|
typing.Dict[str, Type[Any]] | A dictionary mapping variable names to their guessed 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 LiteralType for which to guess the corresponding Python type. |
Returns
| Type | Description |
|---|
Type[T] | The guessed Python type corresponding to the given Flyte LiteralType. |