scaffold.flyte ============== .. py:module:: scaffold.flyte Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/scaffold/flyte/core/index /autoapi/scaffold/flyte/flyte_utils/index /autoapi/scaffold/flyte/git/index /autoapi/scaffold/flyte/image_builder/index /autoapi/scaffold/flyte/launcher_conf/index Classes ------- .. autoapisummary:: scaffold.flyte.FlyteRemoteHelper Functions --------- .. autoapisummary:: scaffold.flyte.apply_runtime_cfg scaffold.flyte.run_local_workflow scaffold.flyte.runtime_task scaffold.flyte.zen_call Package Contents ---------------- .. py:class:: FlyteRemoteHelper(domain: str, admin_endpoint: str, insecure: bool = True, project: str = 'default') Instantiates a flyteRemote object internally. :param domain: In which flyte domain we operate, usually one of staging, development, production :type domain: str :param admin_endpoint: 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` :type admin_endpoint: str :param insecure: 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. :type insecure: bool :param project: flyte project to operate in. Defaults to 'default' :type project: str .. py:method:: 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. :param launchplan_name: Name of the launchplan. Note that launchplans registered via scaffold usually have a _0 suffix. :type launchplan_name: str :param input_args: 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. :type input_args: dict :param wait: If true, this function will only return once the workflow finishes. Defaults to False. :type wait: bool :returns: The execution object returned by the remote client. :rtype: FlyteWorkflowExecution .. py:method:: fetch_launchplan(launchplan_name: str) -> flytekit.remote.FlyteLaunchPlan Fetches a registered, remote launchplan by name. .. py:attribute:: flyte_remote .. py:function:: 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*. :param runtime_cfg: A DictConfig that conforms to ``RuntimeConf`` (fields: ``logging_cfg``, ``verbose``). :type runtime_cfg: DictConfig .. py:function:: 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. :param workflow_fn: A flytekit ``@workflow``-decorated function. :type workflow_fn: WorkflowBase :param cfg: The ``DictConfig`` received by the Hydra ``main()`` function. :type cfg: DictConfig :param \*\*kwargs: Additional keyword arguments to pass to the workflow function. This is useful for passing in runtime values that are not part of the config :type \*\*kwargs: Any .. rubric:: Example :: def main(cfg: DictConfig) -> None: run_workflow(pipeline, cfg) .. py:function:: 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. :param runtime_cfg_key: Name of the ``DictConfig`` parameter that carries runtime settings. Defaults to ``RUNTIME_CFG_KEY``. :type runtime_cfg_key: str :param \*\*task_kwargs: All keyword arguments accepted by flytekit's ``@task`` decorator (e.g. ``requests``, ``cache``, ``container_image``). :type \*\*task_kwargs: Any .. rubric:: Example :: @runtime_task(requests=DEFAULT_RESOURCES, cache=Cache(version="1")) def my_task(cfg: DictConfig, runtime_cfg: DictConfig) -> str: return zen(my_function)(cfg) .. py:function:: 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. :param fn: Plain Python function to wrap with ``zen()``. :type fn: Callable[..., Any] :param cfg: Per-task ``DictConfig`` passed in from the workflow. :type cfg: Dict :param \*\*kwargs: Runtime values to pre-fill on *fn* before ``zen()`` processes the config. When empty, no ``partial`` is created. :type \*\*kwargs: Any :returns: The return value of the zen-wrapped function call. :rtype: Any