scaffold.flyte.flyte_utils

Attributes

logger

Classes

FlyteRemoteHelper

Instantiates a flyteRemote object internally.

Functions

apply_runtime_cfg(→ None)

Apply runtime configuration (usually inside a Flyte pod).

run_local_workflow(→ None)

Execute a Flyte workflow from Hydra's main() function.

runtime_task(→ Callable[[Callable[Ellipsis, Any]], Any])

Drop-in replacement for @task that applies runtime_cfg before execution.

zen_call(→ Any)

Invoke a zen-wrapped function with config plus optional runtime kwargs.

Module Contents

class scaffold.flyte.flyte_utils.FlyteRemoteHelper(domain: str, admin_endpoint: str, insecure: bool = True, project: str = 'default')

Instantiates a flyteRemote object internally.

Parameters:

  • domain (str) – In which flyte domain we operate, usually one of staging, development, production

  • admin_endpoint (str) – Connection to Flyte admin. Usually ‘flyteadmin.flyte.svc.cluster.local:81’ within the cluster, or localhost:30081 if forwarded via kubectl port-forward –address 0.0.0.0 svc/flyteadmin 30081:81 -n flyte

  • insecure (bool) – If true, no SSL is used for the connection. If the connection is just within cluster or through a port-forward done by kubectl it can be set to True.

  • project (str) – flyte project to operate in. Defaults to ‘default’

execute_flyte_launchplan(launchplan_name: str, input_args: dict, wait: bool = False) → Any

Fetches a launchplan using the connection from init and executes it with the given arguments.

Parameters:

  • launchplan_name (str) – Name of the launchplan. Note that launchplans registered via scaffold usually have a _0 suffix.

  • input_args (dict) – A dictionary mapping from workflow input arguments as string keys to the values with which the workflow gets executed. All arguments that do not have a default argument in the launchplan need to be provided. Note that cfg does have a default argument in launchplans registered using the flyte launcher.

  • wait (bool) – If true, this function will only return once the workflow finishes. Defaults to False.

Returns:

The execution object returned by the remote client.

Return type:

FlyteWorkflowExecution

fetch_launchplan(launchplan_name: str) → flytekit.remote.FlyteLaunchPlan

Fetches a registered, remote launchplan by name.

flyte_remote
scaffold.flyte.flyte_utils.apply_runtime_cfg(runtime_cfg: omegaconf.DictConfig) → None

Apply runtime configuration (usually inside a Flyte pod).

Currently configures Python logging via Hydra’s configure_log when logging_cfg is present in runtime_cfg.

Parameters:

runtime_cfg (DictConfig) – A DictConfig that conforms to RuntimeConf (fields: logging_cfg, verbose).

scaffold.flyte.flyte_utils.run_local_workflow(workflow_fn: Any, cfg: Dict, **kwargs: Any) → None

Execute a Flyte workflow from Hydra’s main() function.

Builds runtime_cfg from the active HydraConfig and maps all other config keys to workflow inputs by name.

Parameters:

  • workflow_fn (WorkflowBase) – A flytekit @workflow-decorated function.

  • cfg (DictConfig) – The DictConfig received by the Hydra main() function.

  • **kwargs (Any) – Additional keyword arguments to pass to the workflow function. This is useful for passing in runtime values that are not part of the config

Example

def main(cfg: DictConfig) -> None:
    run_workflow(pipeline, cfg)
scaffold.flyte.flyte_utils.runtime_task(*, runtime_cfg_key: str = RUNTIME_CFG_KEY, **task_kwargs: Any) → Callable[[Callable[Ellipsis, Any]], Any]

Drop-in replacement for @task that applies runtime_cfg before execution.

In addition to calling the wrapped function, this decorator:

  1. Calls apply_runtime_cfg with the runtime_cfg keyword argument so that logging (and other runtime settings) are configured at the start of every remote task execution.

  2. Automatically adds runtime_cfg to Cache.ignored_inputs so that changes to runtime-only settings (e.g. log verbosity) never invalidate the Flyte task cache.

Parameters:

  • runtime_cfg_key (str) – Name of the DictConfig parameter that carries runtime settings. Defaults to RUNTIME_CFG_KEY.

  • **task_kwargs (Any) – All keyword arguments accepted by flytekit’s @task decorator (e.g. requests, cache, container_image).

Example

@runtime_task(requests=DEFAULT_RESOURCES, cache=Cache(version="1"))
def my_task(cfg: DictConfig, runtime_cfg: DictConfig) -> str:
    return zen(my_function)(cfg)
scaffold.flyte.flyte_utils.zen_call(fn: Callable[Ellipsis, Any], cfg: Dict, **kwargs: Any) → Any

Invoke a zen-wrapped function with config plus optional runtime kwargs.

A convenience over the two patterns that appear in task bodies:

  • zen(fn)(cfg) — when all inputs come from the config.

  • zen(functools.partial(fn, **kwargs))(cfg) — when some inputs come from previous task outputs and cannot be expressed as config fields.

Parameters:

  • fn (Callable[..., Any]) – Plain Python function to wrap with zen().

  • cfg (Dict) – Per-task DictConfig passed in from the workflow.

  • **kwargs (Any) – Runtime values to pre-fill on fn before zen() processes the config. When empty, no partial is created.

Returns:

The return value of the zen-wrapped function call.

Return type:

Any

scaffold.flyte.flyte_utils.logger