flytekit.clis.sdk_in_container.run

Directory

Classes

Errors

• FlyteEntityNotFoundException
• FlyteSystemException

flytekit.clis.sdk_in_container.run.Annotations

def Annotations(
    values,
):

Annotation values to be applied to a workflow execution resource.

Methods

from_flyte_idl()

def from_flyte_idl(
    pb2_object,
):

serialize_to_string()

def serialize_to_string()

short_string()

def short_string()

to_flyte_idl()

def to_flyte_idl()

verbose_string()

def verbose_string()

Properties

flytekit.clis.sdk_in_container.run.ArtifactQuery

def ArtifactQuery(
    artifact: Artifact,
    name: str,
    project: Optional[str],
    domain: Optional[str],
    time_partition: Optional[TimePartition],
    partitions: Optional[Partitions],
    tag: Optional[str],
):

Methods

get_partition_str()

def get_partition_str(
    kwargs,
):

get_str()

def get_str(
    kwargs,
):

get_time_partition_str()

def get_time_partition_str(
    kwargs,
):

to_flyte_idl()

def to_flyte_idl(
    kwargs,
):

Properties

flytekit.clis.sdk_in_container.run.Context

The context is a special internal object that holds state relevant for the script execution at every single level. It’s normally invisible to commands unless they opt-in to getting access to it.

The context is useful as it can pass internal objects around and can control special execution features such as reading data from environment variables.

A context can be used as context manager in which case it will call :meth:close on teardown.

def Context(
    command: Command,
    parent: typing.Optional[ForwardRef('Context')],
    info_name: typing.Optional[str],
    obj: typing.Optional[typing.Any],
    auto_envvar_prefix: typing.Optional[str],
    default_map: typing.Optional[typing.MutableMapping[str, typing.Any]],
    terminal_width: typing.Optional[int],
    max_content_width: typing.Optional[int],
    resilient_parsing: bool,
    allow_extra_args: typing.Optional[bool],
    allow_interspersed_args: typing.Optional[bool],
    ignore_unknown_options: typing.Optional[bool],
    help_option_names: typing.Optional[typing.List[str]],
    token_normalize_func: typing.Optional[typing.Callable[[str], str]],
    color: typing.Optional[bool],
    show_default: typing.Optional[bool],
):

Methods

abort()

def abort()

Aborts the script.

call_on_close()

def call_on_close(
    f: typing.Callable[..., typing.Any],
):

Register a function to be called when the context tears down.

This can be used to close resources opened during the script execution. Resources that support Python’s context manager protocol which would be used in a with statement should be registered with :meth:with_resource instead.

close()

def close()

Invoke all close callbacks registered with :meth:call_on_close, and exit all context managers entered with :meth:with_resource.

ensure_object()

def ensure_object(
    object_type: typing.Type[~V],
):

Like :meth:find_object but sets the innermost object to a new instance of object_type if it does not exist.

exit()

def exit(
    code: int,
):

Exits the application with a given exit code.

fail()

def fail(
    message: str,
):

Aborts the execution of the program with a specific error message.

find_object()

def find_object(
    object_type: typing.Type[~V],
):

Finds the closest object of a given type.

find_root()

def find_root()

Finds the outermost context.

forward()

def forward(
    _Context__self,
    _Context__cmd: Command,
    args: `*args`,
    kwargs: `**kwargs`,
):

Similar to :meth:invoke but fills in default keyword arguments from the current context if the other command expects it. This cannot invoke callbacks directly, only other commands.

.. versionchanged:: 8.0 All kwargs are tracked in :attr:params so they will be passed if forward is called at multiple levels.

get_help()

def get_help()

Helper method to get formatted help page for the current context and command.

get_parameter_source()

def get_parameter_source(
    name: str,
):

Get the source of a parameter. This indicates the location from which the value of the parameter was obtained.

This can be useful for determining when a user specified a value on the command line that is the same as the default value. It will be :attr:~click.core.ParameterSource.DEFAULT only if the value was actually taken from the default.

get_usage()

def get_usage()

Helper method to get formatted usage string for the current context and command.

invoke()

def invoke(
    _Context__self,
    _Context__callback: typing.Union[ForwardRef('Command'), ForwardRef('t.Callable[..., V]')],
    args: `*args`,
    kwargs: `**kwargs`,
):

Invokes a command callback in exactly the way it expects. There are two ways to invoke this method:

1. the first argument can be a callback and all other arguments and keyword arguments are forwarded directly to the function.
2. the first argument is a click command object. In that case all arguments are forwarded as well but proper click parameters (options and click arguments) must be keyword arguments and Click will fill in defaults.

Note that before Click 3.2 keyword arguments were not properly filled in against the intention of this code and no context was created. For more information about this change and why it was done in a bugfix release see :ref:upgrade-to-3.2.

.. versionchanged:: 8.0 All kwargs are tracked in :attr:params so they will be passed if :meth:forward is called at multiple levels.

lookup_default()

def lookup_default(
    name: str,
    call: bool,
):

Get the default for a parameter from :attr:default_map.

make_formatter()

def make_formatter()

Creates the :class:~click.HelpFormatter for the help and usage output.

To quickly customize the formatter class used without overriding this method, set the :attr:formatter_class attribute.

.. versionchanged:: 8.0 Added the :attr:formatter_class attribute.

scope()

def scope(
    cleanup: bool,
):

This helper method can be used with the context object to promote it to the current thread local (see :func:get_current_context). The default behavior of this is to invoke the cleanup functions which can be disabled by setting cleanup to False. The cleanup functions are typically used for things such as closing file handles.

If the cleanup is intended the context object can also be directly used as a context manager.

Example usage::

with ctx.scope(): assert get_current_context() is ctx

This is equivalent::

with ctx: assert get_current_context() is ctx

.. versionadded:: 5.0

set_parameter_source()

def set_parameter_source(
    name: str,
    source: <enum 'ParameterSource'>,
):

Set the source of a parameter. This indicates the location from which the value of the parameter was obtained.

to_info_dict()

def to_info_dict()

Gather information that could be useful for a tool generating user-facing documentation. This traverses the entire CLI structure.

.. code-block:: python

with Context(cli) as ctx: info = ctx.to_info_dict()

.. versionadded:: 8.0

with_resource()

def with_resource(
    context_manager: typing.ContextManager[~V],
):

Register a resource as if it were used in a with statement. The resource will be cleaned up when the context is popped.

Uses :meth:contextlib.ExitStack.enter_context. It calls the resource’s __enter__() method and returns the result. When the context is popped, it closes the stack, which calls the resource’s __exit__() method.

To register a cleanup function for something that isn’t a context manager, use :meth:call_on_close. Or use something from :mod:contextlib to turn it into a context manager first.

.. code-block:: python

@click.group() @click.option("–name") @click.pass_context def cli(ctx): ctx.obj = ctx.with_resource(connect_db(name))

Properties

flytekit.clis.sdk_in_container.run.CopyFileDetection

Create a collection of name/value pairs.

Example enumeration:

class Color(Enum): … RED = 1 … BLUE = 2 … GREEN = 3

Access them by:

• attribute access:

Color.RED <Color.RED: 1>

• value lookup:

Color(1) <Color.RED: 1>

• name lookup:

Color[‘RED’] <Color.RED: 1>

Enumerations can be iterated over, and know how many members they have:

len(Color) 3

list(Color) [<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

flytekit.clis.sdk_in_container.run.DefaultImages

We may want to load the default images from remote - maybe s3 location etc?

Methods

default_image()

def default_image()

find_image_for()

def find_image_for(
    python_version: typing.Optional[flytekit.configuration.default_images.PythonVersion],
    flytekit_version: typing.Optional[str],
):

get_version_suffix()

def get_version_suffix()

flytekit.clis.sdk_in_container.run.DynamicEntityLaunchCommand

This is a dynamic command that is created for each launch plan. This is used to execute a launch plan. It will fetch the launch plan from remote and create parameters from all the inputs of the launch plan.

def DynamicEntityLaunchCommand(
    name: str,
    h: str,
    entity_name: str,
    launcher: str,
    kwargs,
):

Create Rich Command instance.

Methods

collect_usage_pieces()

def collect_usage_pieces(
    ctx: click.core.Context,
):

Returns all the pieces that go into the usage line and returns it as a list of strings.

format_epilog()

def format_epilog(
    ctx: rich_click.rich_context.RichContext,
    formatter: rich_click.rich_help_formatter.RichHelpFormatter,
):

Writes the epilog into the formatter if it exists.

format_help()

def format_help(
    ctx: rich_click.rich_context.RichContext,
    formatter: rich_click.rich_help_formatter.RichHelpFormatter,
):

Writes the help into the formatter if it exists.

This is a low-level method called by :meth:get_help.

This calls the following methods:

• :meth:format_usage
• :meth:format_help_text
• :meth:format_options
• :meth:format_epilog

format_help_text()

def format_help_text(
    ctx: rich_click.rich_context.RichContext,
    formatter: rich_click.rich_help_formatter.RichHelpFormatter,
):

Writes the help text to the formatter if it exists.

format_options()

def format_options(
    ctx: click.core.Context,
    formatter: click.formatting.HelpFormatter,
):

Writes all the options into the formatter if they exist.

format_usage()

def format_usage(
    ctx: click.core.Context,
    formatter: click.formatting.HelpFormatter,
):

Writes the usage line into the formatter.

This is a low-level method called by :meth:get_usage.

get_help()

def get_help(
    ctx: click.core.Context,
):

Formats the help into a string and returns it.

Calls :meth:format_help internally.

get_help_option()

def get_help_option(
    ctx: click.core.Context,
):

Returns the help option object.

Unless add_help_option is False.

.. versionchanged:: 8.1.8 The help option is now cached to avoid creating it multiple times.

get_help_option_names()

def get_help_option_names(
    ctx: click.core.Context,
):

Returns the names for the help option.

get_params()

def get_params(
    ctx: click.core.Context,
):

get_short_help_str()

def get_short_help_str(
    limit: int,
):

Gets short help for the command or makes it by shortening the long help string.

get_usage()

def get_usage(
    ctx: click.core.Context,
):

Formats the usage line into a string and returns it.

Calls :meth:format_usage internally.

invoke()

def invoke(
    ctx: click.core.Context,
):

Default or None values should be ignored. Only values that are provided by the user should be passed to the remote execution.

main()

def main(
    args: `*args`,
    prog_name: typing.Optional[str],
    complete_var: typing.Optional[str],
    standalone_mode: bool,
    windows_expand_args: bool,
    extra: typing.Any,
):

This is the way to invoke a script with all the bells and whistles as a command line application. This will always terminate the application after a call. If this is not wanted, SystemExit needs to be caught.

This method is also available by directly calling the instance of a :class:Command.

make_context()

def make_context(
    info_name: typing.Optional[str],
    args: `*args`,
    parent: typing.Optional[click.core.Context],
    extra: typing.Any,
):

This function when given an info name and arguments will kick off the parsing and create a new :class:Context. It does not invoke the actual command callback though.

To quickly customize the context class used without overriding this method, set the :attr:context_class attribute.

make_parser()

def make_parser(
    ctx: click.core.Context,
):

Creates the underlying option parser for this command.

parse_args()

def parse_args(
    ctx: click.core.Context,
    args: `*args`,
):

Given a context and a list of arguments this creates the parser and parses the arguments, then modifies the context as necessary. This is automatically invoked by :meth:make_context.

shell_complete()

def shell_complete(
    ctx: click.core.Context,
    incomplete: str,
):

Return a list of completions for the incomplete value. Looks at the names of options and chained multi-commands.

to_info_dict()

def to_info_dict(
    ctx: click.core.Context,
):

Gather information that could be useful for a tool generating user-facing documentation. This traverses the entire structure below this command.

Use :meth:click.Context.to_info_dict to traverse the entire CLI structure.

Properties

flytekit.clis.sdk_in_container.run.Entities

NamedTuple to group all entities in a file

Methods

all()

def all()

matching_lp()

def matching_lp(
    lp_name: str,
):

Returns the variable name of the launch plan in the file

flytekit.clis.sdk_in_container.run.FastPackageOptions

FastPackageOptions is used to set configuration options when packaging files.

def FastPackageOptions(
    ignores: list[Ignore],
    keep_default_ignores: bool,
    copy_style: Optional[CopyFileDetection],
    show_files: bool,
):

flytekit.clis.sdk_in_container.run.FastSerializationSettings

This object hold information about settings necessary to serialize an object so that it can be fast-registered.

def FastSerializationSettings(
    enabled: bool,
    destination_dir: Optional[str],
    distribution_location: Optional[str],
):

Methods

from_dict()

def from_dict(
    kvs: typing.Union[dict, list, str, int, float, bool, NoneType],
    infer_missing,
):

from_json()

def from_json(
    s: typing.Union[str, bytes, bytearray],
    parse_float,
    parse_int,
    parse_constant,
    infer_missing,
    kw,
):

schema()

def schema(
    infer_missing: bool,
    only,
    exclude,
    many: bool,
    context,
    load_only,
    dump_only,
    partial: bool,
    unknown,
):

to_dict()

def to_dict(
    encode_json,
):

to_json()

def to_json(
    skipkeys: bool,
    ensure_ascii: bool,
    check_circular: bool,
    allow_nan: bool,
    indent: typing.Union[int, str, NoneType],
    separators: typing.Tuple[str, str],
    default: typing.Callable,
    sort_keys: bool,
    kw,
):

flytekit.clis.sdk_in_container.run.FileAccessProvider

This is the class that is available through the FlyteContext and can be used for persisting data to the remote durable store.

def FileAccessProvider(
    local_sandbox_dir: typing.Union[str, os.PathLike],
    raw_output_prefix: str,
    data_config: typing.Optional[flytekit.configuration.DataConfig],
    execution_metadata: typing.Optional[dict],
):

Methods

async_get_data()

def async_get_data(
    remote_path: str,
    local_path: str,
    is_multipart: bool,
    kwargs,
):

async_put_data()

def async_put_data(
    local_path: typing.Union[str, os.PathLike],
    remote_path: str,
    is_multipart: bool,
    kwargs,
):

The implication here is that we’re always going to put data to the remote location, so we .remote to ensure we don’t use the true local proxy if the remote path is a file://

async_put_raw_data()

def async_put_raw_data(
    lpath: typing.Union[str, os.PathLike, pathlib.Path, bytes, _io.BufferedReader, _io.BytesIO, _io.StringIO],
    upload_prefix: typing.Optional[str],
    file_name: typing.Optional[str],
    read_chunk_size_bytes: int,
    encoding: str,
    skip_raw_data_prefix: bool,
    kwargs,
):

This is a more flexible version of put that accepts a file-like object or a string path. Writes to the raw output prefix only. If you want to write to another fs use put_data or get the fsspec file system directly. FYI: Currently the raw output prefix set by propeller is already unique per retry and looks like s3://my-s3-bucket/data/o4/feda4e266c748463a97d-n0-0

If lpath is a folder, then recursive will be set. If lpath is a streamable, then it can only be a single file.

Writes to: {raw output prefix}/{upload_prefix}/{file_name}

download()

def download(
    remote_path: str,
    local_path: str,
    kwargs,
):

Downloads from remote to local

download_directory()

def download_directory(
    remote_path: str,
    local_path: str,
    kwargs,
):

Downloads directory from given remote to local path

exists()

def exists(
    path: str,
):

generate_new_custom_path()

def generate_new_custom_path(
    fs: typing.Optional[fsspec.spec.AbstractFileSystem],
    alt: typing.Optional[str],
    stem: typing.Optional[str],
):

Generates a new path with the raw output prefix and a random string appended to it. Optionally, you can provide an alternate prefix and a stem. If stem is provided, it will be appended to the path instead of a random string. If alt is provided, it will replace the first part of the output prefix, e.g. the S3 or GCS bucket.

If wanting to write to a non-random prefix in a non-default S3 bucket, this can be called with alt=“my-alt-bucket” and stem=“my-stem” to generate a path like s3://my-alt-bucket/default-prefix-part/my-stem

get()

def get(
    from_path: str,
    to_path: str,
    recursive: bool,
    kwargs,
):

get_async_filesystem_for_path()

def get_async_filesystem_for_path(
    path: str,
    anonymous: bool,
    kwargs,
):

get_data()

def get_data(
    remote_path: str,
    local_path: str,
    is_multipart: bool,
    kwargs,
):

get_file_tail()

def get_file_tail(
    file_path_or_file_name: str,
):

get_filesystem()

def get_filesystem(
    protocol: typing.Optional[str],
    anonymous: bool,
    path: typing.Optional[str],
    kwargs,
):

get_filesystem_for_path()

def get_filesystem_for_path(
    path: str,
    anonymous: bool,
    kwargs,
):

get_random_local_directory()

def get_random_local_directory()

get_random_local_path()

def get_random_local_path(
    file_path_or_file_name: typing.Optional[str],
):

Use file_path_or_file_name, when you want a random directory, but want to preserve the leaf file name

get_random_remote_directory()

def get_random_remote_directory()

get_random_remote_path()

def get_random_remote_path(
    file_path_or_file_name: typing.Optional[str],
):

get_random_string()

def get_random_string()

is_remote()

def is_remote(
    path: typing.Union[str, os.PathLike],
):

Deprecated. Let’s find a replacement

join()

def join(
    args: `*args`,
    unstrip: bool,
    fs: typing.Optional[fsspec.spec.AbstractFileSystem],
):

put_data()

def put_data(
    local_path: typing.Union[str, os.PathLike],
    remote_path: str,
    is_multipart: bool,
    kwargs,
):

The implication here is that we’re always going to put data to the remote location, so we .remote to ensure we don’t use the true local proxy if the remote path is a file://

put_raw_data()

def put_raw_data(
    lpath: typing.Union[str, os.PathLike, pathlib.Path, bytes, _io.BufferedReader, _io.BytesIO, _io.StringIO],
    upload_prefix: typing.Optional[str],
    file_name: typing.Optional[str],
    read_chunk_size_bytes: int,
    encoding: str,
    skip_raw_data_prefix: bool,
    kwargs,
):

This is a more flexible version of put that accepts a file-like object or a string path. Writes to the raw output prefix only. If you want to write to another fs use put_data or get the fsspec file system directly. FYI: Currently the raw output prefix set by propeller is already unique per retry and looks like s3://my-s3-bucket/data/o4/feda4e266c748463a97d-n0-0

If lpath is a folder, then recursive will be set. If lpath is a streamable, then it can only be a single file.

Writes to: {raw output prefix}/{upload_prefix}/{file_name}

recursive_paths()

def recursive_paths(
    f: str,
    t: str,
):

sep()

def sep(
    file_system: typing.Optional[fsspec.spec.AbstractFileSystem],
):

strip_file_header()

def strip_file_header(
    path: str,
    trim_trailing_sep: bool,
):

Drops file:// if it exists from the file

upload()

def upload(
    file_path: str,
    to_path: str,
    kwargs,
):

upload_directory()

def upload_directory(
    local_path: str,
    remote_path: str,
    kwargs,
):

Properties

flytekit.clis.sdk_in_container.run.FlyteContext

This is an internal-facing context object, that most users will not have to deal with. It’s essentially a globally available grab bag of settings and objects that allows flytekit to do things like convert complex types, run and compile workflows, serialize Flyte entities, etc.

Even though this object as a current_context function on it, it should not be called directly. Please use the :py:class:flytekit.FlyteContextManager object instead.

Please do not confuse this object with the :py:class:flytekit.ExecutionParameters object.

def FlyteContext(
    file_access: FileAccessProvider,
    level: int,
    flyte_client: Optional['friendly_client.SynchronousFlyteClient'],
    compilation_state: Optional[CompilationState],
    execution_state: Optional[ExecutionState],
    serialization_settings: Optional[SerializationSettings],
    in_a_condition: bool,
    origin_stackframe: Optional[traceback.FrameSummary],
    output_metadata_tracker: Optional[OutputMetadataTracker],
    worker_queue: Optional[Controller],
):

Methods

current_context()

def current_context()

This method exists only to maintain backwards compatibility. Please use FlyteContextManager.current_context() instead.

Users of flytekit should be wary not to confuse the object returned from this function with :py:func:flytekit.current_context

enter_conditional_section()

def enter_conditional_section()

get_deck()

def get_deck()

Returns the deck that was created as part of the last execution.

The return value depends on the execution environment. In a notebook, the return value is compatible with IPython.display and should be rendered in the notebook.

.. code-block:: python

with flytekit.new_context() as ctx: my_task(…) ctx.get_deck()

OR if you wish to explicitly display

.. code-block:: python

from IPython import display display(ctx.get_deck())

get_origin_stackframe_repr()

def get_origin_stackframe_repr()

new_builder()

def new_builder()

new_compilation_state()

def new_compilation_state(
    prefix: str,
):

Creates and returns a default compilation state. For most of the code this should be the entrypoint of compilation, otherwise the code should always uses - with_compilation_state

new_execution_state()

def new_execution_state(
    working_dir: Optional[os.PathLike],
):

Creates and returns a new default execution state. This should be used at the entrypoint of execution, in all other cases it is preferable to use with_execution_state

set_stackframe()

def set_stackframe(
    s: traceback.FrameSummary,
):

with_client()

def with_client(
    c: SynchronousFlyteClient,
):

with_compilation_state()

def with_compilation_state(
    c: CompilationState,
):

with_execution_state()

def with_execution_state(
    es: ExecutionState,
):

with_file_access()

def with_file_access(
    fa: FileAccessProvider,
):

with_new_compilation_state()

def with_new_compilation_state()

with_output_metadata_tracker()

def with_output_metadata_tracker(
    t: OutputMetadataTracker,
):

with_serialization_settings()

def with_serialization_settings(
    ss: SerializationSettings,
):

with_worker_queue()

def with_worker_queue(
    wq: Controller,
):

Properties

flytekit.clis.sdk_in_container.run.FlyteContextManager

FlyteContextManager manages the execution context within Flytekit. It holds global state of either compilation or Execution. It is not thread-safe and can only be run as a single threaded application currently. Context’s within Flytekit is useful to manage compilation state and execution state. Refer to CompilationState and ExecutionState for more information. FlyteContextManager provides a singleton stack to manage these contexts.

Typical usage is

.. code-block:: python

FlyteContextManager.initialize() with FlyteContextManager.with_context(o) as ctx: pass

If required - not recommended you can use

FlyteContextManager.push_context()

but correspondingly a pop_context should be called

FlyteContextManager.pop_context()

Methods

add_signal_handler()

def add_signal_handler(
    handler: typing.Callable[[int, FrameType], typing.Any],
):

current_context()

def current_context()

get_origin_stackframe()

def get_origin_stackframe(
    limit,
):

initialize()

def initialize()

Re-initializes the context and erases the entire context

pop_context()

def pop_context()

push_context()

def push_context(
    ctx: FlyteContext,
    f: Optional[traceback.FrameSummary],
):

size()

def size()

with_context()

def with_context(
    b: FlyteContext.Builder,
):

flytekit.clis.sdk_in_container.run.FlyteEntityNotFoundException

Inappropriate argument value (of correct type).

def FlyteEntityNotFoundException(
    module_name: str,
    entity_name: str,
):

Properties

flytekit.clis.sdk_in_container.run.FlyteLaunchPlan

A class encapsulating a remote Flyte launch plan.

def FlyteLaunchPlan(
    id,
    args,
    kwargs,
):

Methods

compile()

def compile(
    ctx: FlyteContext,
    args,
    kwargs,
):

construct_node_metadata()

def construct_node_metadata()

Used when constructing the node that encapsulates this task as part of a broader workflow definition.

execute()

def execute(
    kwargs,
):

from_flyte_idl()

def from_flyte_idl(
    pb2,
):

local_execute()

def local_execute(
    ctx: flytekit.core.context_manager.FlyteContext,
    kwargs,
):

local_execution_mode()

def local_execution_mode()

promote_from_model()

def promote_from_model(
    id: id_models.Identifier,
    model: _launch_plan_models.LaunchPlanSpec,
):

serialize_to_string()

def serialize_to_string()

short_string()

def short_string()

to_flyte_idl()

def to_flyte_idl()

verbose_string()

def verbose_string()

Properties

flytekit.clis.sdk_in_container.run.FlyteLiteralConverter

def FlyteLiteralConverter(
    flyte_ctx: flytekit.core.context_manager.FlyteContext,
    literal_type: flytekit.models.types.LiteralType,
    python_type: typing.Type,
    is_remote: bool,
):

Methods

convert()

def convert(
    ctx: click.core.Context,
    param: typing.Optional[click.core.Parameter],
    value: typing.Any,
):

Convert the value to a Flyte Literal or a python native type. This is used by click to convert the input.

is_bool()

def is_bool()

Properties

flytekit.clis.sdk_in_container.run.FlyteRemote

Main entrypoint for programmatically accessing a Flyte remote backend.

The term ‘remote’ is synonymous with ‘backend’ or ‘deployment’ and refers to a hosted instance of the Flyte platform, which comes with a Flyte Admin server on some known URI.

def FlyteRemote(
    config: Config,
    default_project: typing.Optional[str],
    default_domain: typing.Optional[str],
    data_upload_location: str,
    interactive_mode_enabled: typing.Optional[bool],
    kwargs,
):

Initialize a FlyteRemote object.

:type kwargs: All arguments that can be passed to create the SynchronousFlyteClient. These are usually grpc parameters, if you want to customize credentials, ssl handling etc.

Methods

activate_launchplan()

def activate_launchplan(
    ident: Identifier,
):

Given a launchplan, activate it, all previous versions are deactivated.

approve()

def approve(
    signal_id: str,
    execution_name: str,
    project: str,
    domain: str,
):

auto()

def auto(
    config_file: typing.Union[str, ConfigFile],
    default_project: typing.Optional[str],
    default_domain: typing.Optional[str],
    data_upload_location: str,
    interactive_mode_enabled: bool,
    kwargs,
):

download()

def download(
    data: typing.Union[LiteralsResolver, Literal, LiteralMap],
    download_to: str,
    recursive: bool,
):

Download the data to the specified location. If the data is a LiteralsResolver, LiteralMap and if recursive is specified, then all file like objects will be recursively downloaded (e.g. FlyteFile/Dir (blob), StructuredDataset etc).

Note: That it will use your sessions credentials to access the remote location. For sandbox, this should be automatically configured, assuming you are running sandbox locally. For other environments, you will need to configure your credentials appropriately.

execute()

def execute(
    entity: typing.Union[FlyteTask, FlyteLaunchPlan, FlyteWorkflow, PythonTask, WorkflowBase, LaunchPlan, ReferenceEntity],
    inputs: typing.Dict[str, typing.Any],
    project: str,
    domain: str,
    name: str,
    version: str,
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    image_config: typing.Optional[ImageConfig],
    options: typing.Optional[Options],
    wait: bool,
    type_hints: typing.Optional[typing.Dict[str, typing.Type]],
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
    serialization_settings: typing.Optional[SerializationSettings],
):

Execute a task, workflow, or launchplan, either something that’s been declared locally, or a fetched entity.

This method supports:

• Flyte{Task, Workflow, LaunchPlan} remote module objects.
• @task-decorated functions and TaskTemplate tasks.
• @workflow-decorated functions.
• LaunchPlan objects.

For local entities, this code will attempt to find the entity first, and if missing, will compile and register the object.

Not all arguments are relevant in all circumstances. For example, there’s no reason to use the serialization settings for entities that have already been registered on Admin.

execute_local_launch_plan()

def execute_local_launch_plan(
    entity: LaunchPlan,
    inputs: typing.Dict[str, typing.Any],
    version: str,
    project: typing.Optional[str],
    domain: typing.Optional[str],
    name: typing.Optional[str],
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    options: typing.Optional[Options],
    wait: bool,
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
    serialization_settings: typing.Optional[SerializationSettings],
):

Execute a locally defined LaunchPlan.

execute_local_task()

def execute_local_task(
    entity: PythonTask,
    inputs: typing.Dict[str, typing.Any],
    project: str,
    domain: str,
    name: str,
    version: str,
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    image_config: typing.Optional[ImageConfig],
    wait: bool,
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
    options: typing.Optional[Options],
    serialization_settings: typing.Optional[SerializationSettings],
):

Execute a @task-decorated function or TaskTemplate task.

execute_local_workflow()

def execute_local_workflow(
    entity: WorkflowBase,
    inputs: typing.Dict[str, typing.Any],
    project: str,
    domain: str,
    name: str,
    version: str,
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    image_config: typing.Optional[ImageConfig],
    options: typing.Optional[Options],
    wait: bool,
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
    serialization_settings: typing.Optional[SerializationSettings],
):

Execute an @workflow decorated function.

execute_reference_launch_plan()

def execute_reference_launch_plan(
    entity: ReferenceLaunchPlan,
    inputs: typing.Dict[str, typing.Any],
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    options: typing.Optional[Options],
    wait: bool,
    type_hints: typing.Optional[typing.Dict[str, typing.Type]],
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
):

Execute a ReferenceLaunchPlan.

execute_reference_task()

def execute_reference_task(
    entity: ReferenceTask,
    inputs: typing.Dict[str, typing.Any],
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    options: typing.Optional[Options],
    wait: bool,
    type_hints: typing.Optional[typing.Dict[str, typing.Type]],
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
):

Execute a ReferenceTask.

execute_reference_workflow()

def execute_reference_workflow(
    entity: ReferenceWorkflow,
    inputs: typing.Dict[str, typing.Any],
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    options: typing.Optional[Options],
    wait: bool,
    type_hints: typing.Optional[typing.Dict[str, typing.Type]],
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
):

Execute a ReferenceWorkflow.

execute_remote_task_lp()

def execute_remote_task_lp(
    entity: typing.Union[FlyteTask, FlyteLaunchPlan],
    inputs: typing.Dict[str, typing.Any],
    project: str,
    domain: str,
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    options: typing.Optional[Options],
    wait: bool,
    type_hints: typing.Optional[typing.Dict[str, typing.Type]],
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
):

Execute a FlyteTask, or FlyteLaunchplan.

NOTE: the name and version arguments are currently not used and only there consistency in the function signature

execute_remote_wf()

def execute_remote_wf(
    entity: FlyteWorkflow,
    inputs: typing.Dict[str, typing.Any],
    project: str,
    domain: str,
    execution_name: typing.Optional[str],
    execution_name_prefix: typing.Optional[str],
    options: typing.Optional[Options],
    wait: bool,
    type_hints: typing.Optional[typing.Dict[str, typing.Type]],
    overwrite_cache: typing.Optional[bool],
    interruptible: typing.Optional[bool],
    envs: typing.Optional[typing.Dict[str, str]],
    tags: typing.Optional[typing.List[str]],
    cluster_pool: typing.Optional[str],
    execution_cluster_label: typing.Optional[str],
):

Execute a FlyteWorkflow.

NOTE: the name and version arguments are currently not used and only there consistency in the function signature

fast_package()

def fast_package(
    root: os.PathLike,
    deref_symlinks: bool,
    output: str,
    options: typing.Optional[FastPackageOptions],
):

Packages the given paths into an installable zip and returns the md5_bytes and the URL of the uploaded location

fast_register_workflow()

def fast_register_workflow(
    entity: WorkflowBase,
    serialization_settings: typing.Optional[SerializationSettings],
    version: typing.Optional[str],
    default_launch_plan: typing.Optional[bool],
    options: typing.Optional[Options],
    fast_package_options: typing.Optional[FastPackageOptions],
):

Use this method to register a workflow with zip mode.

fetch_active_launchplan()

def fetch_active_launchplan(
    project: str,
    domain: str,
    name: str,
):

Returns the active version of the launch plan if it exists or returns None

fetch_execution()

def fetch_execution(
    project: str,
    domain: str,
    name: str,
):

Fetch a workflow execution entity from flyte admin.

fetch_launch_plan()

def fetch_launch_plan(
    project: str,
    domain: str,
    name: str,
    version: str,
):

Fetch a launchplan entity from flyte admin.

fetch_task()

def fetch_task(
    project: str,
    domain: str,
    name: str,
    version: str,
):

Fetch a task entity from flyte admin.

fetch_task_lazy()

def fetch_task_lazy(
    project: str,
    domain: str,
    name: str,
    version: str,
):

Similar to fetch_task, just that it returns a LazyEntity, which will fetch the workflow lazily.

fetch_workflow()

def fetch_workflow(
    project: str,
    domain: str,
    name: str,
    version: str,
):

Fetch a workflow entity from flyte admin.

fetch_workflow_lazy()

def fetch_workflow_lazy(
    project: str,
    domain: str,
    name: str,
    version: str,
):

Similar to fetch_workflow, just that it returns a LazyEntity, which will fetch the workflow lazily.