NativeInterface

A class representing the native interface for a task. This is used to interact with the task and its execution context.

Attributes

• inputs: Dict[str, Tuple[Type, Any]]

  • A dictionary of input names and their types and a value indicating if they have a default value.

• outputs: Dict[str, Type]

  • A dictionary of output names and their types.

• docstring: Optional[Docstring] = None

  • Optional Docstring for the task.

• has_default: ClassVar[Type[_has_default]] = _has_default

  • This can be used to indicate if a specific input has a default value or not, in the case when the default value is not known. An example would be remote tasks.

Constructors

• Initialize a new NativeInterface instance.

   Args:
       inputs: A dictionary mapping input names to tuples of (type, default_value).
               The default_value can be inspect.Parameter.empty or a special _has_default sentinel.
       outputs: A dictionary mapping output names to their types.
       _remote_defaults: An optional dictionary of default values for remote tasks.
   

• Parameters

  • inputs: Dict[str, Tuple[Type, Type[_has_default]| Type[inspect._empty]]]
    • A dictionary mapping input names to tuples of (type, default_value). The default_value can be inspect.Parameter.empty or a special _has_default sentinel.
  • outputs: Dict[str, Type]
    • A dictionary mapping output names to their types.
  • _remote_defaults: Optional[Dict[str, literals_pb2.Literal]]
    • An optional dictionary of default values for remote tasks.

Methods

def has_outputs()

• Check if the task has outputs. This is used to determine if the task has outputs or not.

• Return Value: bool

  • True if the task has outputs, False otherwise.

def required_inputs()

• Get the names of the required inputs for the task. This is used to determine which inputs are required for the task execution.

• Return Value: List[str]

  • A list of required input names.

def num_required_inputs()

• Get the number of required inputs for the task. This is used to determine how many inputs are required for the task execution.

• Return Value: int

  • The number of required inputs.

def from_types(inputs: Dict[str, Tuple[Type, Type[_has_default]| Type[inspect._empty]]], outputs: Dict[str, Type], default_inputs: Optional[Dict[str, literals_pb2.Literal]] = None) - > [NativeInterface](src_flyte_models_nativeinterface)

• Create a new NativeInterface from the given types. This is used to create a native interface for the task.

• Parameters

  • inputs: Dict[str, Tuple[Type, Type[_has_default]| Type[inspect._empty]]]
    • A dictionary of input names and their types and a value indicating if they have a default value.
  • outputs: Dict[str, Type]
    • A dictionary of output names and their types.
  • default_inputs: Optional[Dict[str, literals_pb2.Literal]]
    • Optional dictionary of default inputs for remote tasks.

• Return Value: NativeInterface

  • A NativeInterface object with the given inputs and outputs.

def from_callable(func: Callable) - > [NativeInterface](src_flyte_models_nativeinterface)

• Extract the native interface from the given function. This is used to create a native interface for the task.

• Parameters

  • func: Callable
    • The function to extract the interface from.

• Return Value: NativeInterface

  • A NativeInterface object representing the callable.

def convert_to_kwargs(args: Any, kwargs: Any) - > Dict[str, Any]

• Convert the given arguments to keyword arguments based on the native interface. This is used to convert the arguments to the correct types for the task execution.

• Parameters

  • args: Any
    • Positional arguments.
  • kwargs: Any
    • Keyword arguments.

• Return Value: Dict[str, Any]

  • A dictionary of keyword arguments.

def get_input_types()

• Get the input types for the task. This is used to get the types of the inputs for the task execution.

• Return Value: Dict[str, Type]

  • A dictionary of input names and their types.