scaffold.flyte.core =================== .. py:module:: scaffold.flyte.core Attributes ---------- .. autoapisummary:: scaffold.flyte.core.logger Functions --------- .. autoapisummary:: scaffold.flyte.core.build_runtime_cfg scaffold.flyte.core.build_workflow_inputs scaffold.flyte.core.extract_flyte_entities scaffold.flyte.core.get_serialization_settings scaffold.flyte.core.identify_main_workflow scaffold.flyte.core.mxm_register scaffold.flyte.core.temp_flyte_remote Module Contents --------------- .. py:function:: build_runtime_cfg(hydra_cfg: Any) -> omegaconf.DictConfig Build a RuntimeConf DictConfig from Hydra's internal config. :param hydra_cfg: The object returned by ``HydraConfig.get()`` (or the ``cfg.hydra`` sub-tree available inside the launcher). :type hydra_cfg: Any :returns: A config with ``logging_cfg`` and ``verbose`` keys. :rtype: DictConfig .. py:function:: 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.``. 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 :param workflow: A flytekit ``@workflow``-decorated function. :type workflow: WorkflowBase :param job_cfg: Hydra user config (i.e. ``cfg`` with the ``hydra`` key removed). :type job_cfg: DictConfig :param hydra_cfg: Hydra internal config (``cfg.hydra`` or ``HydraConfig.get()``). :type hydra_cfg: Any :param runtime_cfg_key: Name of the runtime config parameter. :type runtime_cfg_key: str :returns: Mapping of workflow parameter names to their resolved values. :rtype: dict[str, Any] .. py:function:: extract_flyte_entities(new_entities: Set[Union[flytekit.core.base_task.PythonTask, flytekit.core.workflow.WorkflowBase]], entities: Set[Union[flytekit.core.base_task.PythonTask, flytekit.core.workflow.WorkflowBase]]) -> Set[Union[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. :param new_entities: flyte entities used to recursively extract sub-entities :type new_entities: List[Union[PythonTask, WorkflowBase]] :param entities: already discovered entities not to recurse on :type entities: Set[Union[PythonTask, WorkflowBase]] :returns: A set of all children nodes to be registered (including subworkflows). :rtype: Set[Union[PythonTask, WorkflowBase]] .. py:function:: 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. :param default_image: the tag of the image used by default by the flyte tasks :type default_image: str :param extra_images: 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}}"`. :type extra_images: Dict[str, str] :param fast_serialization_settings: Details of image injections in fast serialisation mode :type fast_serialization_settings: FastSerializationSettings :param project: the flyte project :type project: str :param domain: domain, normally one of ``development``, ``staging`` or ``production`` :type domain: str :returns: Flyte SerializationSettings for registering flyte entities. .. py:function:: identify_main_workflow(module_name: str) -> Tuple[flytekit.core.workflow.WorkflowBase, List[Union[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. :param module_name: the python module :type module_name: str :returns: The main workflow and a list of contained flyte entities to be registered :rtype: Tuple[WorkflowBase, List[Union[PythonTask, WorkflowBase]]] :raises ValueError: If the module does not contain exactly one workflow. .. py:function:: mxm_register(nodes: List[Union[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.) :param nodes: A list of tasks and workflows to be registered :type nodes: List[Union[PythonTask, WorkflowBase]] :returns: The original workflow :rtype: Union[PythonTask, WorkflowBase] .. py:function:: 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. :param project: the flyte project :type project: str :param domain: one of `development`, `staging` or `production` :type domain: str :param endpoint: the Flyte platform endpoint to connect to :type endpoint: str :Yields: *Generator[Tuple[FlyteRemote, SerializationSettings], None, None]* -- FlyteRemote: flyte remote object to make register calls. .. py:data:: logger