Skip to main content

TypeEngine

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​

  • lazy_import_lock: typing.ClassVar[threading.Lock] = threading.Lock()
    • 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

Constructors​

  • 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

Methods​

def register(transformer: [TypeTransformer](src_flyte_types__type_engine_typetransformer), additional_types: Optional[typing.List[Type]] = [])

  • This should be used for all types that respond with the right type annotation when you use type(...) function

  • Parameters

    • transformer: TypeTransformer
      • The transformer to register.
    • additional_types: Optional[typing.List[Type]]
      • A list of additional types that this transformer should handle.
def register_restricted_type(name: str, type: Type[T])

  • Registers a type transformer for a restricted type.

  • Parameters

    • name: str
      • The name of the restricted type.
    • type: Type[T]
      • The restricted type.
def register_additional_type(transformer: [TypeTransformer](src_flyte_types__type_engine_typetransformer)[T], additional_type: Type[T], override: bool = False)

  • Registers a type transformer for an additional type.

  • Parameters

    • transformer: TypeTransformer[T]
      • The type transformer to register.
    • additional_type: Type[T]
      • The additional type to register.
    • override: bool
      • Whether to override an existing transformer.
def get_transformer(python_type: Type) - > [TypeTransformer](src_flyte_types__type_engine_typetransformer)

  • Implements a recursive search for the transformer.

  • Parameters

    • python_type: Type
      • The Python type to get the transformer for.

  • Return Value: TypeTransformer

    • The type transformer for the given Python type.
def lazy_import_transformers()
  • Only load the transformers if needed.
def to_literal_type(python_type: Type[T]) - > LiteralType

  • Converts a python type into a flyte specific LiteralType

  • Parameters

    • python_type: Type[T]
      • The Python type to convert.

  • Return Value: LiteralType

    • The Flyte LiteralType representation of the Python type.
def to_literal_checks(python_val: typing.Any, python_type: Type[T], expected: LiteralType)

  • Performs checks before converting a Python value to a Flyte literal.

  • Parameters

    • 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.
def to_literal(python_val: typing.Any, python_type: Type[T], expected: types_pb2.LiteralType) - > literals_pb2.Literal

  • Converts a Python value to a Flyte literal.

  • Parameters

    • 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.

  • Return Value: literals_pb2.Literal

    • The Flyte Literal representation of the Python value.
def unwrap_offloaded_literal(lv: literals_pb2.Literal) - > literals_pb2.Literal

  • Unwraps an offloaded Flyte literal.

  • Parameters

    • lv: literals_pb2.Literal
      • The Flyte literal to unwrap.

  • Return Value: literals_pb2.Literal

    • The unwrapped Flyte literal.
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

    • lv: Literal
      • The Flyte literal to convert.
    • expected_python_type: Type
      • The expected Python type.

  • Return Value: typing.Any

    • The Python value.
def to_html(python_val: typing.Any, expected_python_type: Type[typing.Any]) - > str

  • Converts a Python value to an HTML string representation.

  • Parameters

    • python_val: typing.Any
      • The Python value to convert.
    • expected_python_type: Type[typing.Any]
      • The expected Python type.

  • Return Value: str

    • The HTML string representation.
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

    • t: typing.NamedTuple
      • The NamedTuple to convert.

  • Return Value: interface_pb2.VariableMap

    • A Flyte VariableMap representing the NamedTuple.
def literal_map_to_kwargs(lm: LiteralMap, python_types: typing.Optional[typing.Dict[str, type]], literal_types: typing.Optional[typing.Dict[str, interface_pb2.Variable]]) - > typing.Dict[str, typing.Any]

  • Given a LiteralMap (usually an input into a task - intermediate), convert to kwargs for the task

  • Parameters

    • lm: LiteralMap
      • The LiteralMap to convert.
    • python_types: typing.Optional[typing.Dict[str, type]]
      • A dictionary of expected Python types for the keys in the LiteralMap.
    • literal_types: typing.Optional[typing.Dict[str, interface_pb2.Variable]]
      • A dictionary of Flyte Variable objects for the keys in the LiteralMap.

  • Return Value: typing.Dict[str, typing.Any]

    • A dictionary of keyword arguments.
def dict_to_literal_map(d: typing.Dict[str, typing.Any], type_hints: Optional[typing.Dict[str, type]] = {}) - > 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

    • d: typing.Dict[str, typing.Any]
      • The dictionary to convert.
    • type_hints: Optional[typing.Dict[str, type]]
      • A dictionary of type hints for the keys in the dictionary.

  • Return Value: LiteralMap

    • A Flyte LiteralMap.
def get_available_transformers()

  • Returns all python types for which transformers are available

  • Return Value: typing.KeysView[Type]

    • A view of the keys (Python types) for which transformers are registered.
def guess_python_types(flyte_variable_dict: typing.Dict[str, interface_pb2.Variable]) - > typing.Dict[str, Type[Any]]

  • Transforms a dictionary of flyte-specific Variable objects to a dictionary of regular python values.

  • Parameters

    • flyte_variable_dict: typing.Dict[str, interface_pb2.Variable]
      • A dictionary of Flyte Variable objects.

  • Return Value: typing.Dict[str, Type[Any]]

    • A dictionary mapping string keys to Python types.
def guess_python_type(flyte_type: LiteralType) - > Type[T]

  • Transforms a flyte-specific LiteralType to a regular python value.

  • Parameters

    • flyte_type: LiteralType
      • The Flyte LiteralType to guess the Python type for.

  • Return Value: Type[T]

    • The guessed Python type.