flytekit.types.file.file

Directory

Classes

Methods

Variables

Methods

noop()

def noop()

flytekit.types.file.file.FlyteFile

Parameters

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

FlyteFile’s init method.

Properties

Methods

deserialize_flyte_file()

def deserialize_flyte_file(
    info,
) -> 'FlyteFile'

download()

def download()

extension()

def extension()

from_dict()

def from_dict(
    d,
    dialect = None,
)

from_source()

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

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

new()

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

Create a new FlyteFile object in the current Flyte working directory

new_remote_file()

def new_remote_file(
    name: typing.Optional[str] = None,
    alt: typing.Optional[str] = None,
) -> FlyteFile

Create a new FlyteFile object with a remote path.

open()

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

Returns a streaming File handle

@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

:type mode: str :type cache_type: str, optional :type cache_options: Dict[str, Any], optional

serialize_flyte_file()

def serialize_flyte_file()

to_dict()

def to_dict()

flytekit.types.file.file.FlyteFilePathTransformer

Parameters

def FlyteFilePathTransformer()

Properties

Methods

assert_type()

def assert_type(
    t: typing.Union[typing.Type[FlyteFile], os.PathLike],
    v: typing.Union[FlyteFile, os.PathLike, str],
)

async_to_literal()

def async_to_literal(
    ctx: FlyteContext,
    python_val: typing.Union[FlyteFile, os.PathLike, str],
    python_type: typing.Type[FlyteFile],
    expected: LiteralType,
) -> Literal

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: FlyteContext,
    lv: Literal,
    expected_python_type: typing.Union[typing.Type[FlyteFile], os.PathLike],
) -> FlyteFile

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

dict_to_flyte_file()

def dict_to_flyte_file(
    dict_obj: typing.Dict[str, str],
    expected_python_type: typing.Union[typing.Type[FlyteFile], os.PathLike],
) -> FlyteFile

from_binary_idl()

def from_binary_idl(
    binary_idl_object: Binary,
    expected_python_type: typing.Union[typing.Type[FlyteFile], os.PathLike],
) -> FlyteFile

If the input is from flytekit, the Life Cycle will be as follows:

Life Cycle: binary IDL -> resolved binary -> bytes -> expected Python object (flytekit customized (propeller processing) (flytekit binary IDL) (flytekit customized serialization) deserialization)

Example Code: @dataclass class DC: ff: FlyteFile

@workflow
def wf(dc: DC):
    t_ff(dc.ff)

• The deserialization is the same as put a flyte file in a dataclass, which will deserialize by the mashumaro’s API.

Related PR:

• Title: Override Dataclass Serialization/Deserialization Behavior for FlyteTypes via Mashumaro
• Link: https://github.com/flyteorg/flytekit/pull/2554

from_generic_idl()

def from_generic_idl(
    generic: Struct,
    expected_python_type: typing.Union[typing.Type[FlyteFile], os.PathLike],
) -> FlyteFile

If the input is from Flyte Console, the Life Cycle will be as follows:

Life Cycle: json str -> protobuf struct -> resolved protobuf struct -> expected Python object (console user input) (console output) (propeller) (flytekit customized deserialization)

Example Code: @dataclass class DC: ff: FlyteFile

@workflow def wf(dc: DC): t_ff(dc.ff)

• The deserialization is the same as put a flyte file in a dataclass, which will deserialize by the mashumaro’s API.

Related PR:

• Title: Override Dataclass Serialization/Deserialization Behavior for FlyteTypes via Mashumaro
• Link: https://github.com/flyteorg/flytekit/pull/2554

get_additional_headers()

def get_additional_headers(
    source_path: str | os.PathLike,
) -> typing.Dict[str, str]

get_format()

def get_format(
    t: typing.Union[typing.Type[FlyteFile], os.PathLike],
) -> str

get_literal_type()

def get_literal_type(
    t: typing.Union[typing.Type[FlyteFile], os.PathLike],
) -> LiteralType

Converts the python type to a Flyte LiteralType

get_mime_type_from_extension()

def get_mime_type_from_extension(
    extension: str,
) -> typing.Union[str, typing.Sequence[str]]

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> typing.Type[FlyteFile[typing.Any]]

Converts the Flyte LiteralType to a python object type.

isinstance_generic()

def isinstance_generic(
    obj,
    generic_alias,
)

schema_match()

def schema_match(
    schema: dict,
) -> bool

Check if a JSON schema fragment matches this transformer’s python_type.

For BaseModel subclasses, automatically compares the schema’s title, type, and required fields against the type’s own JSON schema. For other types, returns False by default — override if needed.

to_html()

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

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

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],
) -> Optional[T]

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

validate_file_type()

def validate_file_type(
    python_type: typing.Type[FlyteFile],
    source_path: typing.Union[str, os.PathLike],
)

This method validates the type of the file at source_path against the expected python_type. It uses the magic library to determine the real type of the file. If the magic library is not installed, it logs a debug message and returns. If the actual file does not exist, it returns without raising an error.

Raises