flytekit.interaction.click_types

Directory

Classes

flytekit.interaction.click_types.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.interaction.click_types.BlobType

This type represents offloaded data and is typically used for things like files.

def BlobType(
    format,
    dimensionality,
):

Methods

from_flyte_idl()

def from_flyte_idl(
    proto,
):

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.interaction.click_types.DataClassJsonMixin

DataClassJsonMixin is an ABC that functions as a Mixin.

As with other ABCs, it should not be instantiated directly.

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.interaction.click_types.DateTimeType

The DateTime type converts date strings into datetime objects.

The format strings which are checked are configurable, but default to some common (non-timezone aware) ISO 8601 formats.

When specifying DateTime formats, you should only pass a list or a tuple. Other iterables, like generators, may lead to surprising results.

The format strings are processed using datetime.strptime, and this consequently defines the format strings which are allowed.

Parsing is tried using each format, in order, and the first format which parses successfully is used.

def DateTimeType()

Methods

convert()

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

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call :meth:fail with a descriptive message.

fail()

def fail(
    message: str,
    param: typing.Optional[ForwardRef('Parameter')],
    ctx: typing.Optional[ForwardRef('Context')],
):

Helper method to fail with an invalid value message.

get_metavar()

def get_metavar(
    param: Parameter,
):

Returns the metavar default for this param if it provides one.

get_missing_message()

def get_missing_message(
    param: Parameter,
):

Optionally might return extra information about a missing parameter.

.. versionadded:: 2.0

shell_complete()

def shell_complete(
    ctx: Context,
    param: Parameter,
    incomplete: str,
):

Return a list of :class:~click.shell_completion.CompletionItem objects for the incomplete value. Most types do not provide completions, but some do, and this allows custom types to provide custom completions as well.

split_envvar_value()

def split_envvar_value(
    rv: str,
):

Given a value from an environment variable this splits it up into small chunks depending on the defined envvar list splitter.

If the splitter is set to None, which means that whitespace splits, then leading and trailing whitespace is ignored. Otherwise, leading and trailing splitters usually lead to empty items being included.

to_info_dict()

def to_info_dict()

Gather information that could be useful for a tool generating user-facing documentation.

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

.. versionadded:: 8.0

flytekit.interaction.click_types.DirParamType

Represents the type of a parameter. Validates and converts values from the command line or Python into the correct type.

To implement a custom type, subclass and implement at least the following:

• The :attr:name class attribute must be set.
• Calling an instance of the type with None must return None. This is already implemented by default.
• :meth:convert must convert string values to the correct type.
• :meth:convert must accept values that are already the correct type.
• It must be able to convert a value if the ctx and param arguments are None. This can occur when converting prompt input.

Methods

convert()

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

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call :meth:fail with a descriptive message.

fail()

def fail(
    message: str,
    param: typing.Optional[ForwardRef('Parameter')],
    ctx: typing.Optional[ForwardRef('Context')],
):

Helper method to fail with an invalid value message.

get_metavar()

def get_metavar(
    param: Parameter,
):

Returns the metavar default for this param if it provides one.

get_missing_message()

def get_missing_message(
    param: Parameter,
):

Optionally might return extra information about a missing parameter.

.. versionadded:: 2.0

shell_complete()

def shell_complete(
    ctx: Context,
    param: Parameter,
    incomplete: str,
):

Return a list of :class:~click.shell_completion.CompletionItem objects for the incomplete value. Most types do not provide completions, but some do, and this allows custom types to provide custom completions as well.

split_envvar_value()

def split_envvar_value(
    rv: str,
):

Given a value from an environment variable this splits it up into small chunks depending on the defined envvar list splitter.

If the splitter is set to None, which means that whitespace splits, then leading and trailing whitespace is ignored. Otherwise, leading and trailing splitters usually lead to empty items being included.

to_info_dict()

def to_info_dict()

Gather information that could be useful for a tool generating user-facing documentation.

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

.. versionadded:: 8.0

flytekit.interaction.click_types.DurationParamType

Represents the type of a parameter. Validates and converts values from the command line or Python into the correct type.

To implement a custom type, subclass and implement at least the following:

• The :attr:name class attribute must be set.
• Calling an instance of the type with None must return None. This is already implemented by default.
• :meth:convert must convert string values to the correct type.
• :meth:convert must accept values that are already the correct type.
• It must be able to convert a value if the ctx and param arguments are None. This can occur when converting prompt input.

Methods

convert()

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

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call :meth:fail with a descriptive message.

fail()

def fail(
    message: str,
    param: typing.Optional[ForwardRef('Parameter')],
    ctx: typing.Optional[ForwardRef('Context')],
):

Helper method to fail with an invalid value message.

get_metavar()

def get_metavar(
    param: Parameter,
):

Returns the metavar default for this param if it provides one.

get_missing_message()

def get_missing_message(
    param: Parameter,
):

Optionally might return extra information about a missing parameter.

.. versionadded:: 2.0

shell_complete()

def shell_complete(
    ctx: Context,
    param: Parameter,
    incomplete: str,
):

Return a list of :class:~click.shell_completion.CompletionItem objects for the incomplete value. Most types do not provide completions, but some do, and this allows custom types to provide custom completions as well.

split_envvar_value()

def split_envvar_value(
    rv: str,
):

Given a value from an environment variable this splits it up into small chunks depending on the defined envvar list splitter.

If the splitter is set to None, which means that whitespace splits, then leading and trailing whitespace is ignored. Otherwise, leading and trailing splitters usually lead to empty items being included.

to_info_dict()

def to_info_dict()

Gather information that could be useful for a tool generating user-facing documentation.

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

.. versionadded:: 8.0

flytekit.interaction.click_types.EnumParamType

The choice type allows a value to be checked against a fixed set of supported values. All of these values have to be strings.

You should only pass a list or tuple of choices. Other iterables (like generators) may lead to surprising results.

The resulting value will always be one of the originally passed choices regardless of case_sensitive or any ctx.token_normalize_func being specified.

See :ref:choice-opts for an example.

def EnumParamType(
    enum_type: typing.Type[enum.Enum],
):

Methods

convert()

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

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call :meth:fail with a descriptive message.

fail()

def fail(
    message: str,
    param: typing.Optional[ForwardRef('Parameter')],
    ctx: typing.Optional[ForwardRef('Context')],
):

Helper method to fail with an invalid value message.

get_metavar()

def get_metavar(
    param: Parameter,
):

Returns the metavar default for this param if it provides one.

get_missing_message()

def get_missing_message(
    param: Parameter,
):

Optionally might return extra information about a missing parameter.

.. versionadded:: 2.0

shell_complete()

def shell_complete(
    ctx: Context,
    param: Parameter,
    incomplete: str,
):

Complete choices that start with the incomplete value.

split_envvar_value()

def split_envvar_value(
    rv: str,
):

Given a value from an environment variable this splits it up into small chunks depending on the defined envvar list splitter.

If the splitter is set to None, which means that whitespace splits, then leading and trailing whitespace is ignored. Otherwise, leading and trailing splitters usually lead to empty items being included.

to_info_dict()

def to_info_dict()

Gather information that could be useful for a tool generating user-facing documentation.

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

.. versionadded:: 8.0

flytekit.interaction.click_types.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.interaction.click_types.FileParamType

Represents the type of a parameter. Validates and converts values from the command line or Python into the correct type.

To implement a custom type, subclass and implement at least the following:

• The :attr:name class attribute must be set.
• Calling an instance of the type with None must return None. This is already implemented by default.
• :meth:convert must convert string values to the correct type.
• :meth:convert must accept values that are already the correct type.
• It must be able to convert a value if the ctx and param arguments are None. This can occur when converting prompt input.

Methods

convert()

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

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call :meth:fail with a descriptive message.

fail()

def fail(
    message: str,
    param: typing.Optional[ForwardRef('Parameter')],
    ctx: typing.Optional[ForwardRef('Context')],
):

Helper method to fail with an invalid value message.

get_metavar()

def get_metavar(
    param: Parameter,
):

Returns the metavar default for this param if it provides one.

get_missing_message()

def get_missing_message(
    param: Parameter,
):

Optionally might return extra information about a missing parameter.

.. versionadded:: 2.0

shell_complete()

def shell_complete(
    ctx: Context,
    param: Parameter,
    incomplete: str,
):

Return a list of :class:~click.shell_completion.CompletionItem objects for the incomplete value. Most types do not provide completions, but some do, and this allows custom types to provide custom completions as well.

split_envvar_value()

def split_envvar_value(
    rv: str,
):

Given a value from an environment variable this splits it up into small chunks depending on the defined envvar list splitter.

If the splitter is set to None, which means that whitespace splits, then leading and trailing whitespace is ignored. Otherwise, leading and trailing splitters usually lead to empty items being included.

to_info_dict()

def to_info_dict()

Gather information that could be useful for a tool generating user-facing documentation.

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

.. versionadded:: 8.0

flytekit.interaction.click_types.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.interaction.click_types.FlyteDirectory

def FlyteDirectory(
    path: typing.Union[str, os.PathLike],
    downloader: typing.Optional[typing.Callable],
    remote_directory: typing.Optional[typing.Union[os.PathLike, str, typing.Literal[False]]],
):

Methods

crawl()

def crawl(
    maxdepth: typing.Optional[int],
    topdown: bool,
    kwargs,
):

Crawl returns a generator of all files prefixed by any sub-folders under the given “FlyteDirectory”. if details=True is passed, then it will return a dictionary as specified by fsspec.

Example:

list(fd.crawl()) [("/base", “file1”), ("/base", “dir1/file1”), ("/base", “dir2/file1”), ("/base", “dir1/dir/file1”)]

list(x.crawl(detail=True)) [(’/tmp/test’, {‘my-dir/ab.py’: {’name’: ‘/tmp/test/my-dir/ab.py’, ‘size’: 0, ’type’: ‘file’, ‘created’: 1677720780.2318847, ‘islink’: False, ‘mode’: 33188, ‘uid’: 501, ‘gid’: 0, ‘mtime’: 1677720780.2317934, ‘ino’: 1694329, ’nlink’: 1}})]

deserialize_flyte_dir()

def deserialize_flyte_dir(
    args,
    kwargs,
):

download()

def download()

extension()

def extension()

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,
):

from_source()

def from_source(
    source: str | os.PathLike,
):

Create a new FlyteDirectory object with the remote source set to the input

listdir()

def listdir(
    directory: FlyteDirectory,
):

This function will list all files and folders in the given directory, but without downloading the contents. In addition, it will return a list of FlyteFile and FlyteDirectory objects that have ability to lazily download the contents of the file/folder. For example:

.. code-block:: python

entity = FlyteDirectory.listdir(directory) for e in entity: print(“s3 object:”, e.remote_source)

s3 object: s3://test-flytedir/file1.txt

s3 object: s3://test-flytedir/file2.txt

s3 object: s3://test-flytedir/sub_dir

open(entity[0], “r”) # This will download the file to the local disk. open(entity[0], “r”) # flytekit will read data from the local disk if you open it again.

new()

def new(
    dirname: str | os.PathLike,
):

Create a new FlyteDirectory object in current Flyte working directory.

new_dir()

def new_dir(
    name: typing.Optional[str],
):

This will create a new folder under the current folder. If given a name, it will use the name given, otherwise it’ll pick a random string. Collisions are not checked.

new_file()

def new_file(
    name: typing.Optional[str],
):

This will create a new file under the current folder. If given a name, it will use the name given, otherwise it’ll pick a random string. Collisions are not checked.

new_remote()

def new_remote(
    stem: typing.Optional[str],
    alt: typing.Optional[str],
):

Create a new FlyteDirectory object using the currently configured default remote in the context (i.e. the raw_output_prefix configured in the current FileAccessProvider object in the context). This is used if you explicitly have a folder somewhere that you want to create files under. If you want to write a whole folder, you can let your task return a FlyteDirectory object, and let flytekit handle the uploading.

schema()

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

serialize_flyte_dir()

def serialize_flyte_dir(
    args,
    kwargs,
):

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,
):

Properties

flytekit.interaction.click_types.FlyteFile

def FlyteFile(
    path: typing.Union[str, os.PathLike],
    downloader: typing.Callable,
    remote_path: typing.Optional[typing.Union[os.PathLike, str, bool]],
    metadata: typing.Optional[dict[str, str]],
):

FlyteFile’s init method.

Methods

deserialize_flyte_file()

def deserialize_flyte_file(
    args,
    kwargs,
):

download()

def download()

extension()

def extension()

from_dict()

def from_dict(
    d,
    dialect,
):

from_json()

def from_json(
    data: typing.Union[str, bytes, bytearray],
    decoder: collections.abc.Callable[[typing.Union[str, bytes, bytearray]], dict[typing.Any, typing.Any]],
    from_dict_kwargs: typing.Any,
):

from_source()

def from_source(
    source: str | os.PathLike,
):

Create a new FlyteFile object with the remote source set to the input

new()

def new(
    filename: str | os.PathLike,
):

Create a new FlyteFile object in the current Flyte working directory

new_remote_file()

def new_remote_file(
    name: typing.Optional[str],
    alt: typing.Optional[str],
):

Create a new FlyteFile object with a remote path.

open()

def open(
    mode: str,
    cache_type: typing.Optional[str],
    cache_options: typing.Optional[typing.Dict[str, typing.Any]],
):

Returns a streaming File handle

.. code-block:: python

@task def copy_file(ff: FlyteFile) -> FlyteFile: new_file = FlyteFile.new_remote_file() with ff.open(“rb”, cache_type=“readahead”) as r: with new_file.open(“wb”) as w: w.write(r.read()) return new_file

serialize_flyte_file()

def serialize_flyte_file(
    args,
    kwargs,
):

to_dict()

def to_dict()

to_json()

def to_json(
    encoder: collections.abc.Callable[[typing.Any], typing.Union[str, bytes, bytearray]],
    to_dict_kwargs: typing.Any,
):

Properties

flytekit.interaction.click_types.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.interaction.click_types.FlytePathResolver

Methods

add_mapping()

def add_mapping(
    flyte_uri: str,
    remote_path: str,
):

Thread safe method to dd a mapping from a flyte uri to a remote path

resolve_remote_path()

def resolve_remote_path(
    flyte_uri: str,
):

Given a flyte uri, return the remote path if it exists or was created in current session, otherwise return None

flytekit.interaction.click_types.FlytePickleTransformer

Base transformer type that should be implemented for every python native type that can be handled by flytekit

def FlytePickleTransformer()

Methods

assert_type()

def assert_type(
    t: typing.Type[~T],
    v: ~T,
):

async_to_literal()

def async_to_literal(
    ctx: flytekit.core.context_manager.FlyteContext,
    python_val: ~T,
    python_type: typing.Type[~T],
    expected: flytekit.models.types.LiteralType,
):

Converts a given python_val to a Flyte Literal, assuming the given python_val matches the declared python_type. Implementers should refrain from using type(python_val) instead rely on the passed in python_type. If these do not match (or are not allowed) the Transformer implementer should raise an AssertionError, clearly stating what was the mismatch

async_to_python_value()

def async_to_python_value(
    ctx: flytekit.core.context_manager.FlyteContext,
    lv: flytekit.models.literals.Literal,
    expected_python_type: typing.Type[~T],
):

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

from_binary_idl()

def from_binary_idl(
    binary_idl_object: Binary,
    expected_python_type: Type[T],
):

This function primarily handles deserialization for untyped dicts, dataclasses, Pydantic BaseModels, and attribute access.｀

For untyped dict, dataclass, and pydantic basemodel: Life Cycle (Untyped Dict as example): python val -> msgpack bytes -> binary literal scalar -> msgpack bytes -> python val (to_literal) (from_binary_idl)

For attribute access: Life Cycle: python val -> msgpack bytes -> binary literal scalar -> resolved golang value -> binary literal scalar -> msgpack bytes -> python val (to_literal) (propeller attribute access) (from_binary_idl)

from_generic_idl()

def from_generic_idl(
    generic: Struct,
    expected_python_type: Type[T],
):

TODO: Support all Flyte Types. This is for dataclass attribute access from input created from the Flyte Console.

Note:

• This can be removed in the future when the Flyte Console support generate Binary IDL Scalar as input.

get_literal_type()

def get_literal_type(
    t: typing.Type[~T],
):

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: flytekit.models.types.LiteralType,
):

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
):

to_html()

def to_html(
    ctx: FlyteContext,
    python_val: T,
    expected_python_type: Type[T],
):

Converts any python val (dataframe, int, float) to a html string, and it will be wrapped in the HTML div

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: typing.Any,
    python_type: Type[T],
    expected: LiteralType,
):

Converts a given python_val to a Flyte Literal, assuming the given python_val matches the declared python_type. Implementers should refrain from using type(python_val) instead rely on the passed in python_type. If these do not match (or are not allowed) the Transformer implementer should raise an AssertionError, clearly stating what was the mismatch

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
):

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.interaction.click_types.FlyteSchema

def FlyteSchema(
    local_path: typing.Optional[str],
    remote_path: typing.Optional[str],
    supported_mode: SchemaOpenMode,
    downloader: typing.Optional[typing.Callable],
):

Methods

as_readonly()

def as_readonly()

column_names()

def column_names()

columns()

def columns()

deserialize_flyte_schema()

def deserialize_flyte_schema(
    args,
    kwargs,
):

format()

def format()

from_dict()

def from_dict(
    d,
    dialect,
):

from_json()

def from_json(
    data: typing.Union[str, bytes, bytearray],
    decoder: collections.abc.Callable[[typing.Union[str, bytes, bytearray]], dict[typing.Any, typing.Any]],
    from_dict_kwargs: typing.Any,
):

open()

def open(
    dataframe_fmt: typing.Optional[type],
    override_mode: typing.Optional[SchemaOpenMode],
):

Returns a reader or writer depending on the mode of the object when created. This mode can be overridden, but will depend on whether the override can be performed. For example, if the Object was created in a read-mode a “write mode” override is not allowed. if the object was created in write-mode, a read is allowed.

serialize_flyte_schema()

def serialize_flyte_schema(
    args,
    kwargs,
):

to_dict()

def to_dict()

to_json()

def to_json(
    encoder: collections.abc.Callable[[typing.Any], typing.Union[str, bytes, bytearray]],
    to_dict_kwargs: typing.Any,
):

Properties

flytekit.interaction.click_types.JSONIteratorParamType

Represents the type of a parameter. Validates and converts values from the command line or Python into the correct type.

To implement a custom type, subclass and implement at least the following:

• The :attr:name class attribute must be set.
• Calling an instance of the type with None must return None. This is already implemented by default.
• :meth:convert must convert string values to the correct type.
• :meth:convert must accept values that are already the correct type.
• It must be able to convert a value if the ctx and param arguments are None. This can occur when converting prompt input.

Methods

convert()

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

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call :meth:fail with a descriptive message.

fail()

def fail(
    message: str,
    param: typing.Optional[ForwardRef('Parameter')],
    ctx: typing.Optional[ForwardRef('Context')],
):

Helper method to fail with an invalid value message.

get_metavar()

def get_metavar(
    param: Parameter,
):

Returns the metavar default for this param if it provides one.

get_missing_message()

def get_missing_message(
    param: Parameter,
):

Optionally might return extra information about a missing parameter.

.. versionadded:: 2.0

shell_complete()

def shell_complete(
    ctx: Context,
    param: Parameter,
    incomplete: str,
):

Return a list of :class:~click.shell_completion.CompletionItem objects for the incomplete value. Most types do not provide completions, but some do, and this allows custom types to provide custom completions as well.

split_envvar_value()

def split_envvar_value(
    rv: str,
):

Given a value from an environment variable this splits it up into small chunks depending on the defined envvar list splitter.

If the splitter is set to None, which means that whitespace splits, then leading and trailing whitespace is ignored. Otherwise, leading and trailing splitters usually lead to empty items being included.

to_info_dict()

def to_info_dict()

Gather information that could be useful for a tool generating user-facing documentation.

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

.. versionadded:: 8.0

flytekit.interaction.click_types.JSONIteratorTransformer

A JSON iterator that handles conversion between an iterator/generator and a JSONL file.

def JSONIteratorTransformer()

Methods

assert_type()

def assert_type(
    t: Type[T],
    v: T,
):

async_to_literal()

def async_to_literal(
    ctx: flytekit.core.context_manager.FlyteContext,
    python_val: typing.Iterator[typing.Union[typing.Dict[str, typing.Any], typing.List[typing.Any], bool, float, int, str]],
    python_type: typing.Type[typing.Iterator[typing.Union[typing.Dict[str, typing.Any], typing.List[typing.Any], bool, float, int, str]]],
    expected: flytekit.models.types.LiteralType,
):

Converts a given python_val to a Flyte Literal, assuming the given python_val matches the declared python_type. Implementers should refrain from using type(python_val) instead rely on the passed in python_type. If these do not match (or are not allowed) the Transformer implementer should raise an AssertionError, clearly stating what was the mismatch

async_to_python_value()

def async_to_python_value(
    ctx: flytekit.core.context_manager.FlyteContext,
    lv: flytekit.models.literals.Literal,
    expected_python_type: typing.Type[typing.Iterator[typing.Union[typing.Dict[str, typing.Any], typing.List[typing.Any], bool, float, int, str]]],
):

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

from_binary_idl()

def from_binary_idl(
    binary_idl_object: Binary,
    expected_python_type: Type[T],
):

This function primarily handles deserialization for untyped dicts, dataclasses, Pydantic BaseModels, and attribute access.｀

For untyped dict, dataclass, and pydantic basemodel: Life Cycle (Untyped Dict as example): python val -> msgpack bytes -> binary literal scalar -> msgpack bytes -> python val (to_literal) (from_binary_idl)

For attribute access: Life Cycle: python val -> msgpack bytes -> binary literal scalar -> resolved golang value -> binary literal scalar -> msgpack bytes -> python val (to_literal) (propeller attribute access) (from_binary_idl)

from_generic_idl()

def from_generic_idl(
    generic: Struct,
    expected_python_type: Type[T],
):

TODO: Support all Flyte Types. This is for dataclass attribute access from input created from the Flyte Console.

Note:

• This can be removed in the future when the Flyte Console support generate Binary IDL Scalar as input.

get_literal_type()

def get_literal_type(
    t: typing.Type[typing.Iterator[typing.Union[typing.Dict[str, typing.Any], typing.List[typing.Any], bool, float, int, str]]],
):

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: flytekit.models.types.LiteralType,
):

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
):

to_html()

def to_html(
    ctx: FlyteContext,
    python_val: T,
    expected_python_type: Type[T],
):

Converts any python val (dataframe, int, float) to a html string, and it will be wrapped in the HTML div

to_literal()

def to_literal(
    ctx: FlyteContext,
    python_val: typing.Any,
    python_type: Type[T],
    expected: LiteralType,
):

Converts a given python_val to a Flyte Literal, assuming the given python_val matches the declared python_type. Implementers should refrain from using type(python_val) instead rely on the passed in python_type. If these do not match (or are not allowed) the Transformer implementer should raise an AssertionError, clearly stating what was the mismatch

to_python_value()

def to_python_value(
    ctx: FlyteContext,
    lv: Literal,
    expected_python_type: Type[T],
):

Converts the given Literal to a Python Type. If the conversion cannot be done an AssertionError should be raised

Properties

flytekit.interaction.click_types.JsonParamType

Represents the type of a parameter. Validates and converts values from the command line or Python into the correct type.

To implement a custom type, subclass and implement at least the following:

• The :attr:name class attribute must be set.
• Calling an instance of the type with None must return None. This is already implemented by default.
• :meth:convert must convert string values to the correct type.
• :meth:convert must accept values that are already the correct type.
• It must be able to convert a value if the ctx and param arguments are None. This can occur when converting prompt input.

def JsonParamType(
    python_type: typing.Type,
):

Methods

convert()

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

Convert the value to the correct type. This is not called if the value is None (the missing value).

This must accept string values from the command line, as well as values that are already the correct type. It may also convert other compatible types.

The param and ctx arguments may be None in certain situations, such as when converting prompt input.

If the value cannot be converted, call :meth:fail with a descriptive message.

fail()

def fail(
    message: str,
    param: typing.Optional[ForwardRef('Parameter')],
    ctx: typing.Optional[ForwardRef('Context')],
):

Helper method to fail with an invalid value message.

get_metavar()

def get_metavar(
    param: Parameter,
):

Returns the metavar default for this param if it provides one.

get_missing_message()

def get_missing_message(
    param: Parameter,
):

Optionally might return extra information about a missing parameter.

.. versionadded:: 2.0

shell_complete()

def shell_complete(
    ctx: Context,
    param: Parameter,
    incomplete: str,
):

Return a list of :class:~click.shell_completion.CompletionItem objects for the incomplete value. Most types do not provide completions, but some do, and this allows custom types to provide custom completions as well.

split_envvar_value()

def split_envvar_value(
    rv: str,
):

Given a value from an environment variable this splits it up into small chunks depending on the defined envvar list splitter.

If the splitter is set to None, which means that whitespace splits, then leading and trailing whitespace is ignored. Otherwise, leading and trailing splitters usually lead to empty items being included.

to_info_dict()

def to_info_dict()

Gather information that could be useful for a tool generating user-facing documentation.

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

.. versionadded:: 8.0

flytekit.interaction.click_types.Literal

def Literal(
    scalar: typing.Optional[flytekit.models.literals.Scalar],
    collection: typing.Optional[flytekit.models.literals.LiteralCollection],
    map: typing.Optional[flytekit.models.literals.LiteralMap],
    hash: typing.Optional[str],
    metadata: typing.Optional[typing.Dict[str, str]],
    offloaded_metadata: typing.Optional[flytekit.models.literals.LiteralOffloadedMetadata],
):