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 config 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 config to Flyte workflow inputs by parameter name.

For each parameter in the workflow interface:

• If the name matches runtime_cfg_key: injects a RuntimeConf DictConfig built from Hydra’s logging configuration.

• Otherwise: looks up the value as job_cfg.<name>. Emits a warning when no matching key exists (the workflow default is used in that case).

• For legacy compatibility, if the workflow only expects “cfg” or “config”, do not explode the config

Parameters:

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

• job_cfg (DictConfig) – Hydra user config (i.e. cfg with the hydra key removed).

• hydra_cfg (Any) – Hydra internal config (cfg.hydra or HydraConfig.get()).

• runtime_cfg_key (str) – Name of the runtime config parameter.

Returns:

Mapping of workflow parameter names to their resolved values.

Return type:

dict[str, Any]

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