flytekit.types.schema.types

Directory

Classes

Methods

Variables

Methods

generate_ordered_files()

def generate_ordered_files(
    directory: os.PathLike,
    n: int,
) -> typing.Generator[str, None, None]

flytekit.types.schema.types.FlyteSchema

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

Properties

Methods

as_readonly()

def as_readonly()

column_names()

def column_names()

columns()

def columns()

deserialize_flyte_schema()

def deserialize_flyte_schema(
    info,
) -> FlyteSchema

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,
) -> ~T

open()

def open(
    dataframe_fmt: typing.Optional[type],
    override_mode: typing.Optional[SchemaOpenMode],
) -> typing.Union[SchemaReader, SchemaWriter]

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()

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,
) -> typing.Union[str, bytes, bytearray]

flytekit.types.schema.types.FlyteSchemaTransformer

def FlyteSchemaTransformer()

Properties

Methods

assert_type()

def assert_type(
    t: Type[FlyteSchema],
    v: typing.Any,
)

async_to_literal()

def async_to_literal(
    ctx: FlyteContext,
    python_val: FlyteSchema,
    python_type: Type[FlyteSchema],
    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: Type[FlyteSchema],
) -> FlyteSchema

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

dict_to_flyte_schema()

def dict_to_flyte_schema(
    dict_obj: typing.Dict[str, str],
    expected_python_type: Type[FlyteSchema],
) -> FlyteSchema

from_binary_idl()

def from_binary_idl(
    binary_idl_object: Binary,
    expected_python_type: Type[FlyteSchema],
) -> FlyteSchema

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: fs: FlyteSchema

@workflow def wf(dc: DC): t_fs(dc.fs)

Note:

• The deserialization is the same as put a flyte schema 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: Type[FlyteSchema],
) -> FlyteSchema

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: fs: FlyteSchema

@workflow def wf(dc: DC): t_fs(dc.fs)

Note:

• The deserialization is the same as put a flyte schema 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_literal_type()

def get_literal_type(
    t: Type[FlyteSchema],
) -> LiteralType

Converts the python type to a Flyte LiteralType

guess_python_type()

def guess_python_type(
    literal_type: LiteralType,
) -> Type[FlyteSchema]

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

flytekit.types.schema.types.LocalIOSchemaReader

class LocalIOSchemaReader(
    from_path: str,
    cols: typing.Optional[typing.Dict[str, type]],
    fmt: SchemaFormat,
)

Properties

Methods

all()

def all(
    kwargs,
) -> T

iter()

def iter(
    kwargs,
) -> typing.Generator[T, None, None]

flytekit.types.schema.types.LocalIOSchemaWriter

class LocalIOSchemaWriter(
    to_local_path: str,
    cols: typing.Optional[typing.Dict[str, type]],
    fmt: SchemaFormat,
)

Properties

Methods

write()

def write(
    dfs,
    kwargs,
)

flytekit.types.schema.types.SchemaEngine

This is the core Engine that handles all schema sub-systems. All schema types needs to be registered with this to allow direct support for that type in FlyteSchema. e.g. of possible supported types are Pandas.DataFrame, Spark.DataFrame, Vaex.DataFrame, etc.

Methods

get_handler()

def get_handler(
    t: Type,
) -> SchemaHandler

register_handler()

def register_handler(
    h: SchemaHandler,
)

Register a new handler that can create a SchemaReader and SchemaWriter for the expected type.

flytekit.types.schema.types.SchemaFormat

Represents the schema storage format (at rest). Currently only parquet is supported

flytekit.types.schema.types.SchemaHandler

class SchemaHandler(
    name: str,
    object_type: Type,
    reader: Type[SchemaReader],
    writer: Type[SchemaWriter],
    handles_remote_io: bool,
)

flytekit.types.schema.types.SchemaOpenMode

flytekit.types.schema.types.SchemaReader

Base SchemaReader to handle any readers (that can manage their own IO or otherwise) Use the simplified base LocalIOSchemaReader for non distributed dataframes

class SchemaReader(
    from_path: str,
    cols: typing.Optional[typing.Dict[str, type]],
    fmt: SchemaFormat,
)

Properties

Methods

all()

def all(
    kwargs,
) -> T

iter()

def iter(
    kwargs,
) -> typing.Generator[T, None, None]

flytekit.types.schema.types.SchemaWriter

class SchemaWriter(
    to_path: str,
    cols: typing.Optional[typing.Dict[str, type]],
    fmt: SchemaFormat,
)

Properties

Methods

write()

def write(
    dfs,
    kwargs,
)