flytekit.extend.backend.base_agent

Directory

Classes

Errors

• FlyteAgentNotFound
• FlyteUserException

flytekit.extend.backend.base_agent.ABC

Helper class that provides a standard way to create an ABC using inheritance.

flytekit.extend.backend.base_agent.Agent

A ProtocolMessage

flytekit.extend.backend.base_agent.AgentBase

Helper class that provides a standard way to create an ABC using inheritance.

def AgentBase(
    task_type_name: str,
    task_type_version: int,
    kwargs,
):

Properties

flytekit.extend.backend.base_agent.AgentRegistry

This is the registry for all agents. The agent service will look up the agent registry based on the task type. The agent metadata service will look up the agent metadata based on the agent name.

Methods

get_agent()

def get_agent(
    task_type_name: str,
    task_type_version: int,
):

get_agent_metadata()

def get_agent_metadata(
    name: str,
):

list_agents()

def list_agents()

register()

def register(
    agent: typing.Union[flytekit.extend.backend.base_agent.AsyncAgentBase, flytekit.extend.backend.base_agent.SyncAgentBase],
    override: bool,
):

flytekit.extend.backend.base_agent.Any

Special type indicating an unconstrained type.

• Any is compatible with every type.
• Any assumed to have all methods.
• All values assumed to be instances of Any.

Note that all the above statements are true from the point of view of static type checkers. At runtime, Any should not be used with instance checks.

flytekit.extend.backend.base_agent.AsyncAgentBase

This is the base class for all async agents. It defines the interface that all agents must implement. The agent service is responsible for invoking agents. The propeller will communicate with the agent service to create tasks, get the status of tasks, and delete tasks.

All the agents should be registered in the AgentRegistry. Agent Service will look up the agent based on the task type. Every task type can only have one agent.

def AsyncAgentBase(
    metadata_type: flytekit.extend.backend.base_agent.ResourceMeta,
    kwargs,
):

Methods

create()

def create(
    task_template: flytekit.models.task.TaskTemplate,
    output_prefix: str,
    inputs: typing.Optional[flytekit.models.literals.LiteralMap],
    task_execution_metadata: typing.Optional[flytekit.models.task.TaskExecutionMetadata],
    kwargs,
):

Return a resource meta that can be used to get the status of the task.

delete()

def delete(
    resource_meta: flytekit.extend.backend.base_agent.ResourceMeta,
    kwargs,
):

Delete the task. This call should be idempotent. It should raise an error if fails to delete the task.

get()

def get(
    resource_meta: flytekit.extend.backend.base_agent.ResourceMeta,
    kwargs,
):

Return the status of the task, and return the outputs in some cases. For example, bigquery job can’t write the structured dataset to the output location, so it returns the output literals to the propeller, and the propeller will write the structured dataset to the blob store.

Properties

flytekit.extend.backend.base_agent.AsyncAgentExecutorMixin

This mixin class is used to run the async task locally, and it’s only used for local execution. Task should inherit from this class if the task can be run in the agent.

Asynchronous tasks are tasks that take a long time to complete, such as running a query.

Methods

agent_signal_handler()

def agent_signal_handler(
    resource_meta: flytekit.extend.backend.base_agent.ResourceMeta,
    signum: int,
    frame: frame,
):

execute()

def execute(
    kwargs,
):

flytekit.extend.backend.base_agent.ExecutionState

This is the context that is active when executing a task or a local workflow. This carries the necessary state to execute. Some required things during execution deal with temporary directories, ExecutionParameters that are passed to the user etc.

Attributes: mode (ExecutionState.Mode): Defines the context in which the task is executed (local, hosted, etc). working_dir (os.PathLike): Specifies the remote, external directory where inputs, outputs and other protobufs are uploaded engine_dir (os.PathLike): branch_eval_mode Optional[BranchEvalMode]: Used to determine whether a branch node should execute. user_space_params Optional[ExecutionParameters]: Provides run-time, user-centric context such as a statsd handler, a logging handler, the current execution id and a working directory.

def ExecutionState(
    working_dir: Union[os.PathLike, str],
    mode: Optional[ExecutionState.Mode],
    engine_dir: Optional[Union[os.PathLike, str]],
    branch_eval_mode: Optional[BranchEvalMode],
    user_space_params: Optional[ExecutionParameters],
):

Methods

branch_complete()

def branch_complete()

Indicates that we are within a conditional / ifelse block and the active branch is not done. Default to SKIPPED

is_local_execution()

def is_local_execution()

take_branch()

def take_branch()

Indicates that we are within an if-else block and the current branch has evaluated to true. Useful only in local execution mode

with_params()

def with_params(
    working_dir: Optional[os.PathLike],
    mode: Optional[Mode],
    engine_dir: Optional[os.PathLike],
    branch_eval_mode: Optional[BranchEvalMode],
    user_space_params: Optional[ExecutionParameters],
):

Produces a copy of the current execution state and overrides the copy’s parameters with passed parameter values.

flytekit.extend.backend.base_agent.FlyteAgentNotFound

Assertion failed.

def FlyteAgentNotFound(
    args,
    timestamp: typing.Optional[float],
):

Properties

flytekit.extend.backend.base_agent.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.extend.backend.base_agent.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.extend.backend.base_agent.FlyteUserException

Common base class for all non-exit exceptions.

def FlyteUserException(
    args,
    timestamp: typing.Optional[float],
):

Properties

flytekit.extend.backend.base_agent.FrameType

flytekit.extend.backend.base_agent.ImageConfig

We recommend you to use ImageConfig.auto(img_name=None) to create an ImageConfig. For example, ImageConfig.auto(img_name=““ghcr.io/flyteorg/flytecookbook:v1.0.0"”) will create an ImageConfig.

ImageConfig holds available images which can be used at registration time. A default image can be specified along with optional additional images. Each image in the config must have a unique name.

Attributes: default_image (Optional[Image]): The default image to be used as a container for task serialization. images (List[Image]): Optional, additional images which can be used in task container definitions.

def ImageConfig(
    default_image: Optional[Image],
    images: Optional[List[Image]],
):

Methods

auto()

def auto(
    config_file: typing.Union[str, ConfigFile, None],
    img_name: Optional[str],
):

Reads from config file or from img_name Note that this function does not take into account the flytekit default images (see the Dockerfiles at the base of this repo). To pick those up, see the auto_default_image function..

auto_default_image()

def auto_default_image()

create_from()

def create_from(
    default_image: Optional[Image],
    other_images: typing.Optional[typing.List[Image]],
):

find_image()

def find_image(
    name,
):

Return an image, by name, if it exists.

from_dict()

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

from_images()

def from_images(
    default_image: str,
    m: typing.Optional[typing.Dict[str, str]],
):

Allows you to programmatically create an ImageConfig. Usually only the default_image is required, unless your workflow uses multiple images

.. code:: python

ImageConfig.from_dict( “ghcr.io/flyteorg/flytecookbook:v1.0.0”, { “spark”: “ghcr.io/flyteorg/myspark:…”, “other”: “…”, } )

urn:

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

validate_image()

def validate_image(
    _: typing.Any,
    param: str,
    values: tuple,
):

Validates the image to match the standard format. Also validates that only one default image is provided. a default image, is one that is specified as default=<image_uri> or just <image_uri>. All other images should be provided with a name, in the format name=<image_uri> This method can be used with the CLI

flytekit.extend.backend.base_agent.LiteralMap

def LiteralMap(
    literals,
):

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.extend.backend.base_agent.OrderedDict

Dictionary that remembers insertion order

flytekit.extend.backend.base_agent.Progress

Renders an auto-updating progress bar(s).

def Progress(
    columns: typing.Union[str, rich.progress.ProgressColumn],
    console: typing.Optional[rich.console.Console],
    auto_refresh: bool,
    refresh_per_second: float,
    speed_estimate_period: float,
    transient: bool,
    redirect_stdout: bool,
    redirect_stderr: bool,
    get_time: typing.Optional[typing.Callable[[], float]],
    disable: bool,
    expand: bool,
):

Methods

add_task()

def add_task(
    description: str,
    start: bool,
    total: typing.Optional[float],
    completed: int,
    visible: bool,
    fields: typing.Any,
):

Add a new ’task’ to the Progress display.

advance()

def advance(
    task_id: rich.progress.TaskID,
    advance: float,
):

Advance task by a number of steps.

get_default_columns()

def get_default_columns()

Get the default columns used for a new Progress instance:

• a text column for the description (TextColumn)
• the bar itself (BarColumn)
• a text column showing completion percentage (TextColumn)
• an estimated-time-remaining column (TimeRemainingColumn) If the Progress instance is created without passing a columns argument, the default columns defined here will be used.

You can also create a Progress instance using custom columns before and/or after the defaults, as in this example:

progress = Progress( SpinnerColumn(), *Progress.get_default_columns(), “Elapsed:”, TimeElapsedColumn(), )

This code shows the creation of a Progress display, containing a spinner to the left, the default columns, and a labeled elapsed time column.

get_renderable()

def get_renderable()

Get a renderable for the progress display.

get_renderables()

def get_renderables()

Get a number of renderables for the progress display.

make_tasks_table()

def make_tasks_table(
    tasks: typing.Iterable[rich.progress.Task],
):

Get a table to render the Progress display.

open()

def open(
    file: typing.Union[str, ForwardRef('PathLike[str]'), bytes],
    mode: typing.Union[typing.Literal['rb'], typing.Literal['rt'], typing.Literal['r']],
    buffering: int,
    encoding: typing.Optional[str],
    errors: typing.Optional[str],
    newline: typing.Optional[str],
    total: typing.Optional[int],
    task_id: typing.Optional[rich.progress.TaskID],
    description: str,
):

Track progress while reading from a binary file.

refresh()

def refresh()

Refresh (render) the progress information.

remove_task()

def remove_task(
    task_id: rich.progress.TaskID,
):

Delete a task if it exists.

reset()

def reset(
    task_id: rich.progress.TaskID,
    start: bool,
    total: typing.Optional[float],
    completed: int,
    visible: typing.Optional[bool],
    description: typing.Optional[str],
    fields: typing.Any,
):

Reset a task so completed is 0 and the clock is reset.

start()

def start()

Start the progress display.

start_task()

def start_task(
    task_id: rich.progress.TaskID,
):

Start a task.

Starts a task (used when calculating elapsed time). You may need to call this manually, if you called add_task with start=False.

stop()

def stop()

Stop the progress display.

stop_task()

def stop_task(
    task_id: rich.progress.TaskID,
):

Stop a task.

This will freeze the elapsed time on the task.

track()

def track(
    sequence: typing.Union[typing.Iterable[~ProgressType], typing.Sequence[~ProgressType]],
    total: typing.Optional[float],
    completed: int,
    task_id: typing.Optional[rich.progress.TaskID],
    description: str,
    update_period: float,
):

Track progress by iterating over a sequence.

update()

def update(
    task_id: rich.progress.TaskID,
    total: typing.Optional[float],
    completed: typing.Optional[float],
    advance: typing.Optional[float],
    description: typing.Optional[str],
    visible: typing.Optional[bool],
    refresh: bool,
    fields: typing.Any,
):

Update information associated with a task.

wrap_file()

def wrap_file(
    file: typing.BinaryIO,
    total: typing.Optional[int],
    task_id: typing.Optional[rich.progress.TaskID],
    description: str,
):

Track progress file reading from a binary file.

Properties

flytekit.extend.backend.base_agent.PythonFunctionTask

A Python Function task should be used as the base for all extensions that have a python function. It will automatically detect interface of the python function and when serialized on the hosted Flyte platform handles the writing execution command to execute the function

It is advised this task is used using the @task decorator as follows

.. code-block: python

@task def my_func(a: int) -> str: …

In the above code, the name of the function, the module, and the interface (inputs = int and outputs = str) will be auto detected.

def PythonFunctionTask(
    task_config: T,
    task_function: Callable,
    task_type,
    ignore_input_vars: Optional[List[str]],
    execution_mode: ExecutionBehavior,
    task_resolver: Optional[TaskResolverMixin],
    node_dependency_hints: Optional[Iterable[Union['PythonFunctionTask', '_annotated_launch_plan.LaunchPlan', WorkflowBase]]],
    pickle_untyped: bool,
    kwargs,
):

Methods

compile()

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

Generates a node that encapsulates this task in a workflow definition.

compile_into_workflow()

def compile_into_workflow(
    ctx: FlyteContext,
    task_function: Callable,
    kwargs,
):

In the case of dynamic workflows, this function will produce a workflow definition at execution time which will then proceed to be executed.

construct_node_metadata()

def construct_node_metadata()

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

dispatch_execute()

def dispatch_execute(
    ctx: flytekit.core.context_manager.FlyteContext,
    input_literal_map: flytekit.models.literals.LiteralMap,
):

This method translates Flyte’s Type system based input values and invokes the actual call to the executor This method is also invoked during runtime.

• VoidPromise is returned in the case when the task itself declares no outputs.
• Literal Map is returned when the task returns either one more outputs in the declaration. Individual outputs may be none
• DynamicJobSpec is returned when a dynamic workflow is executed

dynamic_execute()

def dynamic_execute(
    task_function: Callable,
    kwargs,
):

By the time this function is invoked, the local_execute function should have unwrapped the Promises and Flyte literal wrappers so that the kwargs we are working with here are now Python native literal values. This function is also expected to return Python native literal values.

Since the user code within a dynamic task constitute a workflow, we have to first compile the workflow, and then execute that workflow.

When running for real in production, the task would stop after the compilation step, and then create a file representing that newly generated workflow, instead of executing it.

execute()

def execute(
    kwargs,
):

This method will be invoked to execute the task. If you do decide to override this method you must also handle dynamic tasks or you will no longer be able to use the task as a dynamic task generator.

find_lhs()

def find_lhs()

get_command()

def get_command(
    settings: SerializationSettings,
):

Returns the command which should be used in the container definition for the serialized version of this task registered on a hosted Flyte platform.

get_config()

def get_config(
    settings: SerializationSettings,
):

Returns the task config as a serializable dictionary. This task config consists of metadata about the custom defined for this task.

get_container()

def get_container(
    settings: SerializationSettings,
):

Returns the container definition (if any) that is used to run the task on hosted Flyte.

get_custom()

def get_custom(
    settings: flytekit.configuration.SerializationSettings,
):

Return additional plugin-specific custom data (if any) as a serializable dictionary.

get_default_command()

def get_default_command(
    settings: SerializationSettings,
):

Returns the default pyflyte-execute command used to run this on hosted Flyte platforms.

get_extended_resources()

def get_extended_resources(
    settings: SerializationSettings,
):

Returns the extended resources to allocate to the task on hosted Flyte.

get_image()

def get_image(
    settings: SerializationSettings,
):

Update image spec based on fast registration usage, and return string representing the image

get_input_types()

def get_input_types()

Returns the names and python types as a dictionary for the inputs of this task.

get_k8s_pod()

def get_k8s_pod(
    settings: SerializationSettings,
):

Returns the kubernetes pod definition (if any) that is used to run the task on hosted Flyte.

get_sql()

def get_sql(
    settings: flytekit.configuration.SerializationSettings,
):

Returns the Sql definition (if any) that is used to run the task on hosted Flyte.

get_type_for_input_var()

def get_type_for_input_var(
    k: str,
    v: typing.Any,
):

Returns the python type for an input variable by name.

get_type_for_output_var()

def get_type_for_output_var(
    k: str,
    v: typing.Any,
):

Returns the python type for the specified output variable by name.

local_execute()

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

This function is used only in the local execution path and is responsible for calling dispatch execute. Use this function when calling a task with native values (or Promises containing Flyte literals derived from Python native values).

local_execution_mode()

def local_execution_mode()

post_execute()

def post_execute(
    user_params: typing.Optional[flytekit.core.context_manager.ExecutionParameters],
    rval: typing.Any,
):

Post execute is called after the execution has completed, with the user_params and can be used to clean-up, or alter the outputs to match the intended tasks outputs. If not overridden, then this function is a No-op

pre_execute()

def pre_execute(
    user_params: typing.Optional[flytekit.core.context_manager.ExecutionParameters],
):

This is the method that will be invoked directly before executing the task method and before all the inputs are converted. One particular case where this is useful is if the context is to be modified for the user process to get some user space parameters. This also ensures that things like SparkSession are already correctly setup before the type transformers are called

This should return either the same context of the mutated context

reset_command_fn()

def reset_command_fn()

Resets the command which should be used in the container definition of this task to the default arguments. This is useful when the command line is overridden at serialization time.

sandbox_execute()

def sandbox_execute(
    ctx: flytekit.core.context_manager.FlyteContext,
    input_literal_map: flytekit.models.literals.LiteralMap,
):

Call dispatch_execute, in the context of a local sandbox execution. Not invoked during runtime.

set_command_fn()

def set_command_fn(
    get_command_fn: Optional[Callable[[SerializationSettings], List[str]]],
):

By default, the task will run on the Flyte platform using the pyflyte-execute command. However, it can be useful to update the command with which the task is serialized for specific cases like running map tasks (“pyflyte-map-execute”) or for fast-executed tasks.

set_resolver()

def set_resolver(
    resolver: TaskResolverMixin,
):

By default, flytekit uses the DefaultTaskResolver to resolve the task. This method allows the user to set a custom task resolver. It can be useful to override the task resolver for specific cases like running tasks in the jupyter notebook.

Properties

flytekit.extend.backend.base_agent.PythonTask

Base Class for all Tasks with a Python native Interface. This should be directly used for task types, that do not have a python function to be executed. Otherwise refer to :py:class:flytekit.PythonFunctionTask.

def PythonTask(
    task_type: str,
    name: str,
    task_config: typing.Optional[~T],
    interface: typing.Optional[flytekit.core.interface.Interface],
    environment: typing.Optional[typing.Dict[str, str]],
    disable_deck: typing.Optional[bool],
    enable_deck: typing.Optional[bool],
    deck_fields: typing.Optional[typing.Tuple[flytekit.deck.deck.DeckField, ...]],
    kwargs,
):

Methods

compile()

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

Generates a node that encapsulates this task in a workflow definition.

construct_node_metadata()

def construct_node_metadata()

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

dispatch_execute()

def dispatch_execute(
    ctx: flytekit.core.context_manager.FlyteContext,
    input_literal_map: flytekit.models.literals.LiteralMap,
):

This method translates Flyte’s Type system based input values and invokes the actual call to the executor This method is also invoked during runtime.

• VoidPromise is returned in the case when the task itself declares no outputs.
• Literal Map is returned when the task returns either one more outputs in the declaration. Individual outputs may be none
• DynamicJobSpec is returned when a dynamic workflow is executed

execute()

def execute(
    kwargs,
):

This method will be invoked to execute the task.

find_lhs()

def find_lhs()

get_config()

def get_config(
    settings: flytekit.configuration.SerializationSettings,
):

Returns the task config as a serializable dictionary. This task config consists of metadata about the custom defined for this task.

get_container()

def get_container(
    settings: flytekit.configuration.SerializationSettings,
):

Returns the container definition (if any) that is used to run the task on hosted Flyte.

get_custom()

def get_custom(
    settings: flytekit.configuration.SerializationSettings,
):

Return additional plugin-specific custom data (if any) as a serializable dictionary.

get_extended_resources()

def get_extended_resources(
    settings: flytekit.configuration.SerializationSettings,
):

Returns the extended resources to allocate to the task on hosted Flyte.

get_input_types()

def get_input_types()

Returns the names and python types as a dictionary for the inputs of this task.

get_k8s_pod()

def get_k8s_pod(
    settings: flytekit.configuration.SerializationSettings,
):

Returns the kubernetes pod definition (if any) that is used to run the task on hosted Flyte.

get_sql()

def get_sql(
    settings: flytekit.configuration.SerializationSettings,
):

Returns the Sql definition (if any) that is used to run the task on hosted Flyte.

get_type_for_input_var()

def get_type_for_input_var(
    k: str,
    v: typing.Any,
):

Returns the python type for an input variable by name.

get_type_for_output_var()

def get_type_for_output_var(
    k: str,
    v: typing.Any,
):

Returns the python type for the specified output variable by name.

local_execute()

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

This function is used only in the local execution path and is responsible for calling dispatch execute. Use this function when calling a task with native values (or Promises containing Flyte literals derived from Python native values).

local_execution_mode()

def local_execution_mode()

post_execute()

def post_execute(
    user_params: typing.Optional[flytekit.core.context_manager.ExecutionParameters],
    rval: typing.Any,
):

Post execute is called after the execution has completed, with the user_params and can be used to clean-up, or alter the outputs to match the intended tasks outputs. If not overridden, then this function is a No-op

pre_execute()

def pre_execute(
    user_params: typing.Optional[flytekit.core.context_manager.ExecutionParameters],
):

This is the method that will be invoked directly before executing the task method and before all the inputs are converted. One particular case where this is useful is if the context is to be modified for the user process to get some user space parameters. This also ensures that things like SparkSession are already correctly setup before the type transformers are called

This should return either the same context of the mutated context

sandbox_execute()

def sandbox_execute(
    ctx: flytekit.core.context_manager.FlyteContext,
    input_literal_map: flytekit.models.literals.LiteralMap,
):

Call dispatch_execute, in the context of a local sandbox execution. Not invoked during runtime.

Properties

flytekit.extend.backend.base_agent.Resource

This is the output resource of the job.

def Resource(
    phase: <google.protobuf.internal.enum_type_wrapper.EnumTypeWrapper object at 0x1060829f0>,
    message: typing.Optional[str],
    log_links: typing.Optional[typing.List[flyteidl.core.execution_pb2.TaskLog]],
    outputs: typing.Union[flytekit.models.literals.LiteralMap, typing.Dict[str, typing.Any], NoneType],
    custom_info: typing.Optional[typing.Dict[str, typing.Any]],
):

Methods

from_flyte_idl()

def from_flyte_idl(
    pb2_object: flyteidl.admin.agent_pb2.Resource,
):

to_flyte_idl()

def to_flyte_idl()

This function is async to call the async type engine functions. This is okay to do because this is not a normal model class that inherits from FlyteIdlEntity

flytekit.extend.backend.base_agent.ResourceMeta

This is the metadata for the job. For example, the id of the job.

def ResourceMeta()

Methods

decode()

def decode(
    data: bytes,
):

Decode the resource meta from bytes.

encode()

def encode()

Encode the resource meta to bytes.

flytekit.extend.backend.base_agent.RichHandler

A logging handler that renders output with Rich. The time / level / message and file are displayed in columns. The level is color coded, and the message is syntax highlighted.

Note: Be careful when enabling console markup in log messages if you have configured logging for libraries not under your control. If a dependency writes messages containing square brackets, it may not produce the intended output.

def RichHandler(
    level: typing.Union[int, str],
    console: typing.Optional[rich.console.Console],
    show_time: bool,
    omit_repeated_times: bool,
    show_level: bool,
    show_path: bool,
    enable_link_path: bool,
    highlighter: typing.Optional[rich.highlighter.Highlighter],
    markup: bool,
    rich_tracebacks: bool,
    tracebacks_width: typing.Optional[int],
    tracebacks_code_width: int,
    tracebacks_extra_lines: int,
    tracebacks_theme: typing.Optional[str],
    tracebacks_word_wrap: bool,
    tracebacks_show_locals: bool,
    tracebacks_suppress: typing.Iterable[typing.Union[str, module]],
    tracebacks_max_frames: int,
    locals_max_length: int,
    locals_max_string: int,
    log_time_format: typing.Union[str, typing.Callable[[datetime.datetime], rich.text.Text]],
    keywords: typing.Optional[typing.List[str]],
):

Initializes the instance - basically setting the formatter to None and the filter list to empty.

Methods

acquire()

def acquire()

Acquire the I/O thread lock.

addFilter()

def addFilter(
    filter,
):

Add the specified filter to this handler.

close()

def close()

Tidy up any resources used by the handler.

This version removes the handler from an internal map of handlers, _handlers, which is used for handler lookup by name. Subclasses should ensure that this gets called from overridden close() methods.

createLock()

def createLock()

Acquire a thread lock for serializing access to the underlying I/O.

emit()

def emit(
    record: logging.LogRecord,
):

Invoked by logging.

filter()

def filter(
    record,
):

Determine if a record is loggable by consulting all the filters.

The default is to allow the record to be logged; any filter can veto this by returning a false value. If a filter attached to a handler returns a log record instance, then that instance is used in place of the original log record in any further processing of the event by that handler. If a filter returns any other true value, the original log record is used in any further processing of the event by that handler.

If none of the filters return false values, this method returns a log record. If any of the filters return a false value, this method returns a false value.

.. versionchanged:: 3.2

Allow filters to be just callables.

.. versionchanged:: 3.12 Allow filters to return a LogRecord instead of modifying it in place.

flush()

def flush()

Ensure all logging output has been flushed.

This version does nothing and is intended to be implemented by subclasses.

format()

def format(
    record,
):

Format the specified record.

If a formatter is set, use it. Otherwise, use the default formatter for the module.

get_level_text()

def get_level_text(
    record: logging.LogRecord,
):

Get the level name from the record.

get_name()

def get_name()

handle()

def handle(
    record,
):

Conditionally emit the specified logging record.

Emission depends on filters which may have been added to the handler. Wrap the actual emission of the record with acquisition/release of the I/O thread lock.

Returns an instance of the log record that was emitted if it passed all filters, otherwise a false value is returned.

handleError()

def handleError(
    record,
):

Handle errors which occur during an emit() call.

This method should be called from handlers when an exception is encountered during an emit() call. If raiseExceptions is false, exceptions get silently ignored. This is what is mostly wanted for a logging system - most users will not care about errors in the logging system, they are more interested in application errors. You could, however, replace this with a custom handler if you wish. The record which was being processed is passed in to this method.

release()

def release()

Release the I/O thread lock.

removeFilter()

def removeFilter(
    filter,
):

Remove the specified filter from this handler.

render()

def render(
    record: logging.LogRecord,
    traceback: typing.Optional[rich.traceback.Traceback],
    message_renderable: ConsoleRenderable,
):

Render log for display.

render_message()

def render_message(
    record: logging.LogRecord,
    message: str,
):

Render message text in to Text.

setFormatter()

def setFormatter(
    fmt,
):

Set the formatter for this handler.

setLevel()

def setLevel(
    level,
):

Set the logging level of this handler. level must be an int or a str.

set_name()

def set_name(
    name,
):

Properties

flytekit.extend.backend.base_agent.SerializationSettings

These settings are provided while serializing a workflow and task, before registration. This is required to get runtime information at serialization time, as well as some defaults.

Attributes: project (str): The project (if any) with which to register entities under. domain (str): The domain (if any) with which to register entities under. version (str): The version (if any) with which to register entities under. image_config (ImageConfig): The image config used to define task container images. env (Optional[Dict[str, str]]): Environment variables injected into task container definitions. flytekit_virtualenv_root (Optional[str]): During out of container serialize the absolute path of the flytekit virtualenv at serialization time won’t match the in-container value at execution time. This optional value is used to provide the in-container virtualenv path python_interpreter (Optional[str]): The python executable to use. This is used for spark tasks in out of container execution. entrypoint_settings (Optional[EntrypointSettings]): Information about the command, path and version of the entrypoint program. fast_serialization_settings (Optional[FastSerializationSettings]): If the code is being serialized so that it can be fast registered (and thus omit building a Docker image) this object contains additional parameters for serialization. source_root (Optional[str]): The root directory of the source code.

def SerializationSettings(
    image_config: ImageConfig,
    project: typing.Optional[str],
    domain: typing.Optional[str],
    version: typing.Optional[str],
    env: Optional[Dict[str, str]],
    git_repo: Optional[str],
    python_interpreter: str,
    flytekit_virtualenv_root: Optional[str],
    fast_serialization_settings: Optional[FastSerializationSettings],
    source_root: Optional[str],
):

Methods

default_entrypoint_settings()

def default_entrypoint_settings(
    interpreter_path: str,
):

Assumes the entrypoint is installed in a virtual-environment where the interpreter is

for_image()

def for_image(
    image: str,
    version: str,
    project: str,
    domain: str,
    python_interpreter_path: str,
):

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

def from_transport(
    s: str,
):

new_builder()

def new_builder()

Creates a SerializationSettings.Builder that copies the existing serialization settings parameters and allows for customization.

schema()

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

should_fast_serialize()

def should_fast_serialize()

Whether or not the serialization settings specify that entities should be serialized for fast registration.

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

venv_root_from_interpreter()

def venv_root_from_interpreter(
    interpreter_path: str,
):

Computes the path of the virtual environment root, based on the passed in python interpreter path for example /opt/venv/bin/python3 -> /opt/venv

with_serialized_context()

def with_serialized_context()

Use this method to create a new SerializationSettings that has an environment variable set with the SerializedContext This is useful in transporting SerializedContext to serialized and registered tasks. The setting will be available in the env field with the key SERIALIZED_CONTEXT_ENV_VAR :return: A newly constructed SerializationSettings, or self, if it already has the serializationSettings

Properties

flytekit.extend.backend.base_agent.Struct

A ProtocolMessage

Methods

get_or_create_list()

def get_or_create_list(
    key,
):

Returns a list for this key, creating if it didn’t exist already.

get_or_create_struct()

def get_or_create_struct(
    key,
):

Returns a struct for this key, creating if it didn’t exist already.

items()

def items()

keys()

def keys()

update()

def update(
    dictionary,
):

values()

def values()

flytekit.extend.backend.base_agent.SyncAgentBase

This is the base class for all sync agents. It defines the interface that all agents must implement. The agent service is responsible for invoking agents. Propeller sends a request to agent service, and gets a response in the same call.

All the agents should be registered in the AgentRegistry. Agent Service will look up the agent based on the task type. Every task type can only have one agent.

def SyncAgentBase(
    task_type_name: str,
    task_type_version: int,
    kwargs,
):

Methods

do()

def do(
    task_template: flytekit.models.task.TaskTemplate,
    output_prefix: str,
    inputs: typing.Optional[flytekit.models.literals.LiteralMap],
    kwargs,
):

This is the method that the agent will run.

Properties

flytekit.extend.backend.base_agent.SyncAgentExecutorMixin

This mixin class is used to run the sync task locally, and it’s only used for local execution. Task should inherit from this class if the task can be run in the agent.

Synchronous tasks run quickly and can return their results instantly. Sending a prompt to ChatGPT and getting a response, or retrieving some metadata from a backend system.

Methods

execute()

def execute(
    kwargs,
):

flytekit.extend.backend.base_agent.TaskCategory

def TaskCategory(
    name: str,
    version: int,
):

Properties

flytekit.extend.backend.base_agent.TaskExecution

A ProtocolMessage

flytekit.extend.backend.base_agent.TaskExecutionMetadata

def TaskExecutionMetadata(
    task_execution_id,
    namespace,
    labels,
    annotations,
    k8s_service_account,
    environment_variables,
    identity,
):

Runtime task execution metadata.

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.extend.backend.base_agent.TaskLog

A ProtocolMessage

flytekit.extend.backend.base_agent.TaskTemplate

def TaskTemplate(
    id,
    type,
    metadata,
    interface,
    custom,
    container,
    task_type_version,
    security_context,
    config,
    k8s_pod,
    sql,
    extended_resources,
):

A task template represents the full set of information necessary to perform a unit of work in the Flyte system. It contains the metadata about what inputs and outputs are consumed or produced. It also contains the metadata necessary for Flyte Propeller to do the appropriate work.

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.extend.backend.base_agent.TypeEngine

Core Extensible TypeEngine of Flytekit. This should be used to extend the capabilities of FlyteKits type system. Users can implement their own TypeTransformers and register them with the TypeEngine. This will allow special handling of user objects

Methods