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 during the lazy loading of type transformer plugins. |
Methods
register()
@classmethod
def register(
cls,
transformer: 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 | The transformer instance to be registered 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 |
register_restricted_type()
@classmethod
def register_restricted_type(
cls,
name: str,
type: Type[T],
): ...
Registers a Python type as restricted within the Flyte engine, preventing its use in certain contexts.
Parameters
| Name | Type | Description |
|---|
| name | str | The display name or identifier for the restricted type |
| type | Type[T] | The Python class to be added to the restricted types list |
register_additional_type()
@classmethod
def register_additional_type(cls, transformer: TypeTransformer[T], additional_type: Type[T], override=False): ...
Maps an existing transformer to an additional Python type, optionally overriding any existing registration.
Parameters
| Name | Type | Description |
|---|
| transformer | TypeTransformer[T] | The transformer to associate with the new type |
| additional_type | Type[T] | The new Python type to be supported by the provided transformer |
| override | bool = False | If true, allows replacing an existing transformer registration for this type |
@classmethod
def get_transformer(cls, python_type: Type) -> 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 | The transformer capable of handling the type, falling back to a pickle transformer if no specific match is found |
@classmethod
def lazy_import_transformers(cls): ...
Only load the transformers if needed.
to_literal_type()
@classmethod
def to_literal_type(cls, 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-specific type representation used for interface definitions |
to_literal_checks()
@classmethod
def to_literal_checks(cls, python_val: typing.Any, python_type: Type[T], expected: LiteralType): ...
Validates that a Python value is compatible with the expected Flyte LiteralType and Python type hint.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The runtime value to validate |
| python_type | Type[T] | The expected Python type hint |
| expected | LiteralType | The Flyte LiteralType definition to check against |
to_literal()
@classmethod
async def to_literal(
cls, python_val: typing.Any, python_type: Type[T], expected: types_pb2.LiteralType
) -> literals_pb2.Literal: ...
Converts a Python value into a Flyte Literal protobuf object using the appropriate transformer.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The Python object to convert |
| python_type | Type[T] | The Python type hint for the value |
| expected | types_pb2.LiteralType | The target Flyte type definition |
Returns
| Type | Description |
|---|
literals_pb2.Literal | The serialized Flyte Literal representation of the input value |
unwrap_offloaded_literal()
@classmethod
async def unwrap_offloaded_literal(cls, lv: literals_pb2.Literal) -> literals_pb2.Literal: ...
Downloads and deserializes a literal that has been offloaded to remote storage.
Parameters
| Name | Type | Description |
|---|
| lv | literals_pb2.Literal | The literal containing offloaded metadata |
Returns
| Type | Description |
|---|
literals_pb2.Literal | The actual Literal content retrieved from the offloaded URI |
to_python_value()
@classmethod
async def to_python_value(cls, 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 Python type the caller expects to receive |
Returns
| Type | Description |
|---|
typing.Any | The deserialized Python object |
to_html()
@classmethod
def to_html(cls, python_val: typing.Any, expected_python_type: Type[typing.Any]) -> str: ...
Generates an HTML representation of a Python value for visualization in the Flyte UI.
Parameters
| Name | Type | Description |
|---|
| python_val | typing.Any | The value to render |
| expected_python_type | Type[typing.Any] | The type hint used to determine the rendering logic |
Returns
| Type | Description |
|---|
str | A string containing the HTML markup for the value |
named_tuple_to_variable_map()
@classmethod
def named_tuple_to_variable_map(cls, 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 for field definitions |
Returns
| Type | Description |
|---|
interface_pb2.VariableMap | A map of variable names to their corresponding Flyte types |
literal_map_to_kwargs()
@classmethod
async def literal_map_to_kwargs(
cls,
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 values keyed by argument names |
dict_to_literal_map()
@classmethod
async def dict_to_literal_map(
cls,
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 containers |
Returns
| Type | Description |
|---|
LiteralMap | A Flyte LiteralMap containing the converted values |
@classmethod
def get_available_transformers(cls) -> typing.KeysView[Type]: ...
Returns all python types for which transformers are available
Returns
| Type | Description |
|---|
typing.KeysView[Type] | A view of all Python types currently registered in the engine |
guess_python_types()
@classmethod
def guess_python_types(
cls, 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 variable definitions to process |
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(cls, flyte_type: LiteralType) -> Type[T]: ...
Transforms a flyte-specific LiteralType to a regular python value.
Parameters
| Name | Type | Description |
|---|
| flyte_type | LiteralType | The Flyte type definition to reverse-engineer |
Returns
| Type | Description |
|---|
Type[T] | The Python type that best matches the provided Flyte LiteralType |