flytekit.core.interface

Directory

Classes

Methods

Variables

Methods

default_output_name()

def default_output_name(
    index: int,
) -> str

detect_artifact()

def detect_artifact(
    ts: typing.Tuple[typing.Any, ...],
) -> Optional[art_id.ArtifactID]

If the user wishes to control how Artifacts are created (i.e. naming them, etc.) this is where we pick it up and store it in the interface.

extract_return_annotation()

def extract_return_annotation(
    return_annotation: Union[Type, Tuple, None],
) -> Dict[str, Type]

The purpose of this function is to sort out whether a function is returning one thing, or multiple things, and to name the outputs accordingly, either by using our default name function, or from a typing.NamedTuple.

# Option 1
nt1 = typing.NamedTuple("NT1", x_str=str, y_int=int)
def t(a: int, b: str) -> nt1: ...

# Option 2
def t(a: int, b: str) -> typing.NamedTuple("NT1", x_str=str, y_int=int): ...

# Option 3
def t(a: int, b: str) -> typing.Tuple[int, str]: ...

# Option 4
def t(a: int, b: str) -> (int, str): ...

# Option 5
def t(a: int, b: str) -> str: ...

# Option 6
def t(a: int, b: str) -> None: ...

# Options 7/8
def t(a: int, b: str) -> List[int]: ...
def t(a: int, b: str) -> Dict[str, int]: ...

Note that Options 1 and 2 are identical, just syntactic sugar. In the NamedTuple case, we’ll use the names in the definition. In all other cases, we’ll automatically generate output names, indexed starting at 0.

output_name_generator()

def output_name_generator(
    length: int,
) -> Generator[str, None, None]

remap_shared_output_descriptions()

def remap_shared_output_descriptions(
    output_descriptions: Dict[str, str],
    outputs: Dict[str, Type],
) -> n: Dict of output variable names mapping to shared output description

Deals with mixed styles of return value descriptions used in docstrings. If the docstring contains a single entry of return value description, that output description is shared by each output variable.

repr_kv()

def repr_kv(
    k: str,
    v: Union[Type, Tuple[Type, Any]],
) -> str

repr_type_signature()

def repr_type_signature(
    io: Union[Dict[str, Tuple[Type, Any]], Dict[str, Type]],
) -> str

Converts an inputs and outputs to a type signature

transform_function_to_interface()

def transform_function_to_interface(
    fn: typing.Callable,
    docstring: Optional[Docstring],
    is_reference_entity: bool,
    pickle_untyped: bool,
) -> Interface

From the annotations on a task function that the user should have provided, and the output names they want to use for each output parameter, construct the TypedInterface object

For now the fancy object, maybe in the future a dumb object.

transform_inputs_to_parameters()

def transform_inputs_to_parameters(
    ctx: context_manager.FlyteContext,
    interface: Interface,
) -> _interface_models.ParameterMap

Transforms the given interface (with inputs) to a Parameter Map with defaults set

transform_interface_to_list_interface()

def transform_interface_to_list_interface(
    interface: Interface,
    bound_inputs: typing.Set[str],
    excluded_inputs: typing.Set[str],
    optional_outputs: bool,
) -> Interface

Takes a single task interface and interpolates it to an array interface - to allow performing distributed python map like functions

transform_interface_to_typed_interface()

def transform_interface_to_typed_interface(
    interface: typing.Optional[Interface],
    allow_partial_artifact_id_binding: bool,
) -> typing.Optional[_interface_models.TypedInterface]

Transform the given simple python native interface to FlyteIDL’s interface

transform_type()

def transform_type(
    x: type,
    description: Optional[str],
) -> _interface_models.Variable

transform_types_to_list_of_type()

def transform_types_to_list_of_type(
    m: Dict[str, type],
    bound_inputs: typing.Set[str],
    list_as_optional: bool,
) -> Dict[str, type]

Converts unbound inputs into the equivalent (optional) collections. This is useful for array jobs / map style code. It will create a collection of types even if any one these types is not a collection type.

transform_variable_map()

def transform_variable_map(
    variable_map: Dict[str, type],
    descriptions: Optional[Dict[str, str]],
) -> Dict[str, _interface_models.Variable]

Given a map of str (names of inputs for instance) to their Python native types, return a map of the name to a Flyte Variable object with that type.

verify_outputs_artifact_bindings()

def verify_outputs_artifact_bindings(
    inputs: Dict[str, type],
    outputs: Dict[str, _interface_models.Variable],
    allow_partial_artifact_id_binding: bool,
)

flytekit.core.interface.Interface

A Python native interface object, like inspect.signature but simpler.

class Interface(
    inputs: Union[Optional[Dict[str, Type]], Optional[Dict[str, Tuple[Type, Any]]]],
    outputs: Union[Optional[Dict[str, Type]], Optional[Dict[str, Optional[Type]]]],
    output_tuple_name: Optional[str],
    docstring: Optional[Docstring],
)

Methods

remove_inputs()

def remove_inputs(
    vars: Optional[List[str]],
) -> Interface

This method is useful in removing some variables from the Flyte backend inputs specification, as these are implicit local only inputs or will be supplied by the library at runtime. For example, spark-session etc It creates a new instance of interface with the requested variables removed

with_inputs()

def with_inputs(
    extra_inputs: Dict[str, Type],
) -> Interface

Use this to add additional inputs to the interface. This is useful for adding additional implicit inputs that are added without the user requesting for them

with_outputs()

def with_outputs(
    extra_outputs: Dict[str, Type],
) -> Interface

This method allows addition of extra outputs are expected from a task specification

Properties