scaffold.flyte.core

Attributes

logger

Functions

build_runtime_cfg(→ omegaconf.DictConfig)

Build a RuntimeConf DictConfig from Hydra's internal config.

build_workflow_inputs(→ dict[str, Any])

Map a Hydra configuration dictionary to Flyte workflow inputs by parameter name.

extract_flyte_entities(...)

Recursively extract sub-workflows and other flyte entities from a list of nodes or flyte entities.

get_serialization_settings(...)

Create flyte serialization settings for registering tasks and workflows.

identify_main_workflow(...)

Identify the main Flyte workflow and its subcomponents in a python module.

mxm_register(→ Callable)

Wrapper for both functions and workflows to declare additional flyte components for registration.

temp_flyte_remote(...)

Create a flyte remote object with a temporary proxy into the kubernetes cluster.

Module Contents

scaffold.flyte.core.build_runtime_cfg(hydra_cfg: Any) → omegaconf.DictConfig

Build a RuntimeConf DictConfig from Hydra’s internal config.

Parameters:

hydra_cfg (Any) – The object returned by HydraConfig.get() (or the cfg.hydra sub-tree available inside the launcher).

Returns:

A config with logging_cfg and verbose keys.

Return type:

DictConfig

scaffold.flyte.core.build_workflow_inputs(workflow: Any, job_cfg: omegaconf.DictConfig, hydra_cfg: Any, runtime_cfg_key: str = RUNTIME_CFG_KEY) → dict[str, Any]

Map a Hydra configuration dictionary to Flyte workflow inputs by parameter name.

For each input argument of the signature of the workflow function, this function resolves and populates its value using the following rules:

  1. Logging Runtime Configuration: If the parameter name matches runtime_cfg_key, it injects a RuntimeConf configuration built dynamically from Hydra’s internal logging configuration.

  2. Top-Level Configuration: If the parameter name is precisely "cfg" or "config", the entire top-level configuration is injected. This preserves backward compatibility with legacy workflows that accept the full configuration object rather than exploded parameters. Consequently, parameters named "cfg" or "config" cannot represent sub-keys and will always resolve to the top-level configuration.

  3. Exploded Parameters: For all other parameter names, the function attempts to resolve the value from the top level config using the parameter name as a path key (via OmegaConf.select). If a match is found, the value is populated.

Note

This function runs once, locally, when the Hydra launcher plugin registers a LaunchPlan with Flyte. Its return value becomes that launch plan’s default_inputs.

The actual values used for a given execution are decided afterwards, by whichever actor triggers that execution. This will be the flyte launcher plugin for executions started via Hydra, but could also be a CronSchedule firing on its own (which injects e.g. its kickoff_time_input_arg). Any input value such an actor supplies explicitly overrides the default computed here for that one execution. The values from this function only take effect when nothing overrides them.

Parameters:

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

  • job_cfg (DictConfig) – The user-defined Hydra configuration (i.e., the full config with the internal hydra metadata key removed).

  • hydra_cfg (Any) – The internal Hydra configuration block (typically cfg.hydra or obtained via HydraConfig.get()).

  • runtime_cfg_key (str, optional) – The parameter name reserved for the runtime configuration. Defaults to RUNTIME_CFG_KEY.

Returns:

A mapping of workflow parameter names to their resolved input values.

Return type:

dict[str, Any]

Example

Suppose we have a Flyte workflow defined as:

>>> from flytekit import workflow
>>> from omegaconf import DictConfig
>>>
>>> @workflow
>>> def train_pipeline(cfg: DictConfig, lr: float, epochs: int):
...     pass

And we have the following top-level Hydra user configuration (job_cfg):

>>> from omegaconf import OmegaConf
>>> job_cfg = OmegaConf.create({
...     "lr": 0.001,
...     "epochs": 50,
...     "batch_size": 32,  # Not present in workflow arguments; will be ignored
... })

When we call build_workflow_inputs(train_pipeline, job_cfg, hydra_cfg), the returned dictionary maps the workflow inputs as follows:

>>> inputs = build_workflow_inputs(train_pipeline, job_cfg, hydra_cfg)
>>> print(inputs)
{
    "cfg": {
        "lr": 0.001,
        "epochs": 50,
        "batch_size": 32
    },
    "lr": 0.001,
    "epochs": 50
}

Here, cfg receives the complete, top-level job_cfg dictionary, whereas lr and epochs are resolved individually from their matching keys.

scaffold.flyte.core.extract_flyte_entities(new_entities: Set[flytekit.core.base_task.PythonTask | flytekit.core.workflow.WorkflowBase], entities: Set[flytekit.core.base_task.PythonTask | flytekit.core.workflow.WorkflowBase]) → Set[flytekit.core.base_task.PythonTask | flytekit.core.workflow.WorkflowBase]

Recursively extract sub-workflows and other flyte entities from a list of nodes or flyte entities.

Parameters:

  • new_entities (List[Union[PythonTask, WorkflowBase]]) – flyte entities used to recursively extract sub-entities

  • entities (Set[Union[PythonTask, WorkflowBase]]) – already discovered entities not to recurse on

Returns:

A set of all children nodes to be registered (including subworkflows).

Return type:

Set[Union[PythonTask, WorkflowBase]]

scaffold.flyte.core.get_serialization_settings(default_image: str, extra_images: Dict[str, str], fast_serialization_settings: flytekit.configuration.FastSerializationSettings, project: str, domain: str) → flytekit.configuration.SerializationSettings

Create flyte serialization settings for registering tasks and workflows.

Parameters:

  • default_image (str) – the tag of the image used by default by the flyte tasks

  • extra_images (Dict[str, str]) – a dictionary mapping the names of extra images to be used to the respective tags. The format is e.g. {“spark”: “eu.gcr.io/project_name/spark_image:tag”}. In the flyte task decorator, the image is specified e.g. with container_image=”{{.images.spark.fqn}}:{{.images.default.version}}”.

  • fast_serialization_settings (FastSerializationSettings) – Details of image injections in fast serialisation mode

  • project (str) – the flyte project

  • domain (str) – domain, normally one of development, staging or production

Returns:

Flyte SerializationSettings for registering flyte entities.

scaffold.flyte.core.identify_main_workflow(module_name: str) → Tuple[flytekit.core.workflow.WorkflowBase, List[flytekit.core.base_task.PythonTask | flytekit.core.workflow.WorkflowBase]]

Identify the main Flyte workflow and its subcomponents in a python module.

Top-level workflows are all workflows defined in the same module as hydra.main. The main workflow is the only top-level workflow which is not used in any other top-level workflow - it is assumed, that exactly one such workflow exists. The required flyte components that also need to be registered are identified by recursively descending through the workflow’s node tree.

Note: The main workflow is not contained in the returned list itself.

Parameters:

module_name (str) – the python module

Returns:

The main workflow and a list of contained flyte entities to be registered

Return type:

Tuple[WorkflowBase, List[Union[PythonTask, WorkflowBase]]]

Raises:

ValueError – If the module does not contain exactly one workflow.

scaffold.flyte.core.mxm_register(nodes: List[flytekit.core.base_task.PythonTask | flytekit.core.workflow.WorkflowBase]) → Callable

Wrapper for both functions and workflows to declare additional flyte components for registration. This should go outside the flyte decorators (task, dynamic, workflow etc.)

Parameters:

nodes (List[Union[PythonTask, WorkflowBase]]) – A list of tasks and workflows to be registered

Returns:

The original workflow

Return type:

Union[PythonTask, WorkflowBase]

scaffold.flyte.core.temp_flyte_remote(project: str, domain: str, endpoint: str) → Generator[Tuple[flytekit.remote.remote.FlyteRemote, flytekit.configuration.SerializationSettings], None, None]

Create a flyte remote object with a temporary proxy into the kubernetes cluster.

Parameters:

  • project (str) – the flyte project

  • domain (str) – one of development, staging or production

  • endpoint (str) – the Flyte platform endpoint to connect to

Yields:

Generator[Tuple[FlyteRemote, SerializationSettings], None, None] – FlyteRemote: flyte remote object to make register calls.

scaffold.flyte.core.logger