docs/detailed-documentation/utils/generate_yaml.html

<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1, minimum-scale=1" />
<meta name="generator" content="pdoc 0.10.0" />
<title>codeflare_sdk.utils.generate_yaml API documentation</title>
<meta name="description" content="This sub-module exists primarily to be used internally by the Cluster object
(in the cluster sub-module) for AppWrapper generation." />
<link rel="preload stylesheet" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/11.0.1/sanitize.min.css" integrity="sha256-PK9q560IAAa6WVRRh76LtCaI8pjTJ2z11v0miyNNjrs=" crossorigin>
<link rel="preload stylesheet" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/10up-sanitize.css/11.0.1/typography.min.css" integrity="sha256-7l/o7C8jubJiy74VsKTidCy1yBkRtiUGbVkYBylBqUg=" crossorigin>
<link rel="stylesheet preload" as="style" href="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/10.1.1/styles/github.min.css" crossorigin>
<style>:root{--highlight-color:#fe9}.flex{display:flex !important}body{line-height:1.5em}#content{padding:20px}#sidebar{padding:30px;overflow:hidden}#sidebar > *:last-child{margin-bottom:2cm}.http-server-breadcrumbs{font-size:130%;margin:0 0 15px 0}#footer{font-size:.75em;padding:5px 30px;border-top:1px solid #ddd;text-align:right}#footer p{margin:0 0 0 1em;display:inline-block}#footer p:last-child{margin-right:30px}h1,h2,h3,h4,h5{font-weight:300}h1{font-size:2.5em;line-height:1.1em}h2{font-size:1.75em;margin:1em 0 .50em 0}h3{font-size:1.4em;margin:25px 0 10px 0}h4{margin:0;font-size:105%}h1:target,h2:target,h3:target,h4:target,h5:target,h6:target{background:var(--highlight-color);padding:.2em 0}a{color:#058;text-decoration:none;transition:color .3s ease-in-out}a:hover{color:#e82}.title code{font-weight:bold}h2[id^="header-"]{margin-top:2em}.ident{color:#900}pre code{background:#f8f8f8;font-size:.8em;line-height:1.4em}code{background:#f2f2f1;padding:1px 4px;overflow-wrap:break-word}h1 code{background:transparent}pre{background:#f8f8f8;border:0;border-top:1px solid #ccc;border-bottom:1px solid #ccc;margin:1em 0;padding:1ex}#http-server-module-list{display:flex;flex-flow:column}#http-server-module-list div{display:flex}#http-server-module-list dt{min-width:10%}#http-server-module-list p{margin-top:0}.toc ul,#index{list-style-type:none;margin:0;padding:0}#index code{background:transparent}#index h3{border-bottom:1px solid #ddd}#index ul{padding:0}#index h4{margin-top:.6em;font-weight:bold}@media (min-width:200ex){#index .two-column{column-count:2}}@media (min-width:300ex){#index .two-column{column-count:3}}dl{margin-bottom:2em}dl dl:last-child{margin-bottom:4em}dd{margin:0 0 1em 3em}#header-classes + dl > dd{margin-bottom:3em}dd dd{margin-left:2em}dd p{margin:10px 0}.name{background:#eee;font-weight:bold;font-size:.85em;padding:5px 10px;display:inline-block;min-width:40%}.name:hover{background:#e0e0e0}dt:target .name{background:var(--highlight-color)}.name > span:first-child{white-space:nowrap}.name.class > span:nth-child(2){margin-left:.4em}.inherited{color:#999;border-left:5px solid #eee;padding-left:1em}.inheritance em{font-style:normal;font-weight:bold}.desc h2{font-weight:400;font-size:1.25em}.desc h3{font-size:1em}.desc dt code{background:inherit}.source summary,.git-link-div{color:#666;text-align:right;font-weight:400;font-size:.8em;text-transform:uppercase}.source summary > *{white-space:nowrap;cursor:pointer}.git-link{color:inherit;margin-left:1em}.source pre{max-height:500px;overflow:auto;margin:0}.source pre code{font-size:12px;overflow:visible}.hlist{list-style:none}.hlist li{display:inline}.hlist li:after{content:',\2002'}.hlist li:last-child:after{content:none}.hlist .hlist{display:inline;padding-left:1em}img{max-width:100%}td{padding:0 .5em}.admonition{padding:.1em .5em;margin-bottom:1em}.admonition-title{font-weight:bold}.admonition.note,.admonition.info,.admonition.important{background:#aef}.admonition.todo,.admonition.versionadded,.admonition.tip,.admonition.hint{background:#dfd}.admonition.warning,.admonition.versionchanged,.admonition.deprecated{background:#fd4}.admonition.error,.admonition.danger,.admonition.caution{background:lightpink}</style>
<style media="screen and (min-width: 700px)">@media screen and (min-width:700px){#sidebar{width:30%;height:100vh;overflow:auto;position:sticky;top:0}#content{width:70%;max-width:100ch;padding:3em 4em;border-left:1px solid #ddd}pre code{font-size:1em}.item .name{font-size:1em}main{display:flex;flex-direction:row-reverse;justify-content:flex-end}.toc ul ul,#index ul{padding-left:1.5em}.toc > ul > li{margin-top:.5em}}</style>
<style media="print">@media print{#sidebar h1{page-break-before:always}.source{display:none}}@media print{*{background:transparent !important;color:#000 !important;box-shadow:none !important;text-shadow:none !important}a[href]:after{content:" (" attr(href) ")";font-size:90%}a[href][title]:after{content:none}abbr[title]:after{content:" (" attr(title) ")"}.ir a:after,a[href^="javascript:"]:after,a[href^="#"]:after{content:""}pre,blockquote{border:1px solid #999;page-break-inside:avoid}thead{display:table-header-group}tr,img{page-break-inside:avoid}img{max-width:100% !important}@page{margin:0.5cm}p,h2,h3{orphans:3;widows:3}h1,h2,h3,h4,h5,h6{page-break-after:avoid}}</style>
<script defer src="https://cdnjs.cloudflare.com/ajax/libs/highlight.js/10.1.1/highlight.min.js" integrity="sha256-Uv3H6lx7dJmRfRvH8TH6kJD1TSK1aFcwgx+mdg3epi8=" crossorigin></script>
<script>window.addEventListener('DOMContentLoaded', () => hljs.initHighlighting())</script>
</head>
<body>
<main>
<article id="content">
<header>
<h1 class="title">Module <code>codeflare_sdk.utils.generate_yaml</code></h1>
</header>
<section id="section-intro">
<p>This sub-module exists primarily to be used internally by the Cluster object
(in the cluster sub-module) for AppWrapper generation.</p>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python"># Copyright 2022 IBM, Red Hat
#
# Licensed under the Apache License, Version 2.0 (the &#34;License&#34;);
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an &#34;AS IS&#34; BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

&#34;&#34;&#34;
This sub-module exists primarily to be used internally by the Cluster object
(in the cluster sub-module) for AppWrapper generation.
&#34;&#34;&#34;

from typing import Optional
import typing
import yaml
import sys
import os
import argparse
import uuid
from kubernetes import client, config
from .kube_api_helpers import _kube_api_error_handling
from ..cluster.auth import api_config_handler, config_check
from os import urandom
from base64 import b64encode
from urllib3.util import parse_url


def read_template(template):
    with open(template, &#34;r&#34;) as stream:
        try:
            return yaml.safe_load(stream)
        except yaml.YAMLError as exc:
            print(exc)


def gen_names(name):
    if not name:
        gen_id = str(uuid.uuid4())
        appwrapper_name = &#34;appwrapper-&#34; + gen_id
        cluster_name = &#34;cluster-&#34; + gen_id
        return appwrapper_name, cluster_name
    else:
        return name, name


# Check if the routes api exists
def is_openshift_cluster():
    try:
        config_check()
        for api in client.ApisApi(api_config_handler()).get_api_versions().groups:
            for v in api.versions:
                if &#34;route.openshift.io/v1&#34; in v.group_version:
                    return True
        else:
            return False
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)


def is_kind_cluster():
    try:
        config_check()
        v1 = client.CoreV1Api()
        label_selector = &#34;kubernetes.io/hostname=kind-control-plane&#34;
        nodes = v1.list_node(label_selector=label_selector)
        # If we find one or more nodes with the label, assume it&#39;s a KinD cluster
        return len(nodes.items) &gt; 0
    except Exception as e:
        print(f&#34;Error checking if cluster is KinD: {e}&#34;)
        return False


def update_names(yaml, item, appwrapper_name, cluster_name, namespace):
    metadata = yaml.get(&#34;metadata&#34;)
    metadata[&#34;name&#34;] = appwrapper_name
    metadata[&#34;namespace&#34;] = namespace
    lower_meta = item.get(&#34;generictemplate&#34;, {}).get(&#34;metadata&#34;)
    lower_meta[&#34;labels&#34;][&#34;workload.codeflare.dev/appwrapper&#34;] = appwrapper_name
    lower_meta[&#34;name&#34;] = cluster_name
    lower_meta[&#34;namespace&#34;] = namespace


def update_labels(yaml, instascale, instance_types):
    metadata = yaml.get(&#34;metadata&#34;)
    if instascale:
        if not len(instance_types) &gt; 0:
            sys.exit(
                &#34;If instascale is set to true, must provide at least one instance type&#34;
            )
        type_str = &#34;&#34;
        for type in instance_types:
            type_str += type + &#34;_&#34;
        type_str = type_str[:-1]
        metadata[&#34;labels&#34;][&#34;orderedinstance&#34;] = type_str
    else:
        metadata.pop(&#34;labels&#34;)


def update_priority(yaml, item, dispatch_priority, priority_val):
    spec = yaml.get(&#34;spec&#34;)
    if dispatch_priority is not None:
        if priority_val:
            spec[&#34;priority&#34;] = priority_val
        else:
            raise ValueError(
                &#34;AW generation error: Priority value is None, while dispatch_priority is defined&#34;
            )
        head = item.get(&#34;generictemplate&#34;).get(&#34;spec&#34;).get(&#34;headGroupSpec&#34;)
        worker = item.get(&#34;generictemplate&#34;).get(&#34;spec&#34;).get(&#34;workerGroupSpecs&#34;)[0]
        head[&#34;template&#34;][&#34;spec&#34;][&#34;priorityClassName&#34;] = dispatch_priority
        worker[&#34;template&#34;][&#34;spec&#34;][&#34;priorityClassName&#34;] = dispatch_priority
    else:
        spec.pop(&#34;priority&#34;)


def update_custompodresources(
    item,
    min_cpu,
    max_cpu,
    min_memory,
    max_memory,
    gpu,
    workers,
    head_cpus,
    head_memory,
    head_gpus,
):
    if &#34;custompodresources&#34; in item.keys():
        custompodresources = item.get(&#34;custompodresources&#34;)
        for i in range(len(custompodresources)):
            resource = custompodresources[i]
            if i == 0:
                # Leave head node resources as template default
                resource[&#34;requests&#34;][&#34;cpu&#34;] = head_cpus
                resource[&#34;limits&#34;][&#34;cpu&#34;] = head_cpus
                resource[&#34;requests&#34;][&#34;memory&#34;] = head_memory
                resource[&#34;limits&#34;][&#34;memory&#34;] = head_memory
                resource[&#34;requests&#34;][&#34;nvidia.com/gpu&#34;] = head_gpus
                resource[&#34;limits&#34;][&#34;nvidia.com/gpu&#34;] = head_gpus

            else:
                for k, v in resource.items():
                    if k == &#34;replicas&#34; and i == 1:
                        resource[k] = workers
                    if k == &#34;requests&#34; or k == &#34;limits&#34;:
                        for spec, _ in v.items():
                            if spec == &#34;cpu&#34;:
                                if k == &#34;limits&#34;:
                                    resource[k][spec] = max_cpu
                                else:
                                    resource[k][spec] = min_cpu
                            if spec == &#34;memory&#34;:
                                if k == &#34;limits&#34;:
                                    resource[k][spec] = max_memory
                                else:
                                    resource[k][spec] = min_memory
                            if spec == &#34;nvidia.com/gpu&#34;:
                                if i == 0:
                                    resource[k][spec] = 0
                                else:
                                    resource[k][spec] = gpu
    else:
        sys.exit(&#34;Error: malformed template&#34;)


def update_affinity(spec, appwrapper_name, instascale):
    if instascale:
        node_selector_terms = (
            spec.get(&#34;affinity&#34;)
            .get(&#34;nodeAffinity&#34;)
            .get(&#34;requiredDuringSchedulingIgnoredDuringExecution&#34;)
            .get(&#34;nodeSelectorTerms&#34;)
        )
        node_selector_terms[0][&#34;matchExpressions&#34;][0][&#34;values&#34;][0] = appwrapper_name
        node_selector_terms[0][&#34;matchExpressions&#34;][0][&#34;key&#34;] = appwrapper_name
    else:
        spec.pop(&#34;affinity&#34;)


def update_image(spec, image):
    containers = spec.get(&#34;containers&#34;)
    for container in containers:
        container[&#34;image&#34;] = image


def update_image_pull_secrets(spec, image_pull_secrets):
    template_secrets = spec.get(&#34;imagePullSecrets&#34;, [])
    spec[&#34;imagePullSecrets&#34;] = template_secrets + [
        {&#34;name&#34;: x} for x in image_pull_secrets
    ]


def update_env(spec, env):
    containers = spec.get(&#34;containers&#34;)
    for container in containers:
        if env:
            if &#34;env&#34; in container:
                container[&#34;env&#34;].extend(env)
            else:
                container[&#34;env&#34;] = env


def update_resources(spec, min_cpu, max_cpu, min_memory, max_memory, gpu):
    container = spec.get(&#34;containers&#34;)
    for resource in container:
        requests = resource.get(&#34;resources&#34;).get(&#34;requests&#34;)
        if requests is not None:
            requests[&#34;cpu&#34;] = min_cpu
            requests[&#34;memory&#34;] = min_memory
            requests[&#34;nvidia.com/gpu&#34;] = gpu
        limits = resource.get(&#34;resources&#34;).get(&#34;limits&#34;)
        if limits is not None:
            limits[&#34;cpu&#34;] = max_cpu
            limits[&#34;memory&#34;] = max_memory
            limits[&#34;nvidia.com/gpu&#34;] = gpu


def update_nodes(
    item,
    appwrapper_name,
    min_cpu,
    max_cpu,
    min_memory,
    max_memory,
    gpu,
    workers,
    image,
    instascale,
    env,
    image_pull_secrets,
    head_cpus,
    head_memory,
    head_gpus,
):
    if &#34;generictemplate&#34; in item.keys():
        head = item.get(&#34;generictemplate&#34;).get(&#34;spec&#34;).get(&#34;headGroupSpec&#34;)
        head[&#34;rayStartParams&#34;][&#34;num-gpus&#34;] = str(int(head_gpus))

        worker = item.get(&#34;generictemplate&#34;).get(&#34;spec&#34;).get(&#34;workerGroupSpecs&#34;)[0]
        # Head counts as first worker
        worker[&#34;replicas&#34;] = workers
        worker[&#34;minReplicas&#34;] = workers
        worker[&#34;maxReplicas&#34;] = workers
        worker[&#34;groupName&#34;] = &#34;small-group-&#34; + appwrapper_name
        worker[&#34;rayStartParams&#34;][&#34;num-gpus&#34;] = str(int(gpu))

        for comp in [head, worker]:
            spec = comp.get(&#34;template&#34;).get(&#34;spec&#34;)
            update_affinity(spec, appwrapper_name, instascale)
            update_image_pull_secrets(spec, image_pull_secrets)
            update_image(spec, image)
            update_env(spec, env)
            if comp == head:
                # TODO: Eventually add head node configuration outside of template
                update_resources(
                    spec, head_cpus, head_cpus, head_memory, head_memory, head_gpus
                )
            else:
                update_resources(spec, min_cpu, max_cpu, min_memory, max_memory, gpu)


def del_from_list_by_name(l: list, target: typing.List[str]) -&gt; list:
    return [x for x in l if x[&#34;name&#34;] not in target]


def write_user_appwrapper(user_yaml, output_file_name):
    # Create the directory if it doesn&#39;t exist
    directory_path = os.path.dirname(output_file_name)
    if not os.path.exists(directory_path):
        os.makedirs(directory_path)

    with open(output_file_name, &#34;w&#34;) as outfile:
        yaml.dump(user_yaml, outfile, default_flow_style=False)

    print(f&#34;Written to: {output_file_name}&#34;)


def get_default_kueue_name(namespace: str):
    # If the local queue is set, use it. Otherwise, try to use the default queue.
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        local_queues = api_instance.list_namespaced_custom_object(
            group=&#34;kueue.x-k8s.io&#34;,
            version=&#34;v1beta1&#34;,
            namespace=namespace,
            plural=&#34;localqueues&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)
    for lq in local_queues[&#34;items&#34;]:
        if (
            &#34;annotations&#34; in lq[&#34;metadata&#34;]
            and &#34;kueue.x-k8s.io/default-queue&#34; in lq[&#34;metadata&#34;][&#34;annotations&#34;]
            and lq[&#34;metadata&#34;][&#34;annotations&#34;][&#34;kueue.x-k8s.io/default-queue&#34;].lower()
            == &#34;true&#34;
        ):
            return lq[&#34;metadata&#34;][&#34;name&#34;]
    raise ValueError(
        &#34;Default Local Queue with kueue.x-k8s.io/default-queue: true annotation not found please create a default Local Queue or provide the local_queue name in Cluster Configuration&#34;
    )


def local_queue_exists(namespace: str, local_queue_name: str):
    # get all local queues in the namespace
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        local_queues = api_instance.list_namespaced_custom_object(
            group=&#34;kueue.x-k8s.io&#34;,
            version=&#34;v1beta1&#34;,
            namespace=namespace,
            plural=&#34;localqueues&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)
    # check if local queue with the name provided in cluster config exists
    for lq in local_queues[&#34;items&#34;]:
        if lq[&#34;metadata&#34;][&#34;name&#34;] == local_queue_name:
            return True
    return False


def write_components(
    user_yaml: dict,
    output_file_name: str,
    namespace: str,
    local_queue: Optional[str],
    labels: dict,
):
    # Create the directory if it doesn&#39;t exist
    directory_path = os.path.dirname(output_file_name)
    if not os.path.exists(directory_path):
        os.makedirs(directory_path)

    components = user_yaml.get(&#34;spec&#34;, &#34;resources&#34;)[&#34;resources&#34;].get(&#34;GenericItems&#34;)
    open(output_file_name, &#34;w&#34;).close()
    lq_name = local_queue or get_default_kueue_name(namespace)
    cluster_labels = labels
    if not local_queue_exists(namespace, lq_name):
        raise ValueError(
            &#34;local_queue provided does not exist or is not in this namespace. Please provide the correct local_queue name in Cluster Configuration&#34;
        )
    with open(output_file_name, &#34;a&#34;) as outfile:
        for component in components:
            if &#34;generictemplate&#34; in component:
                if (
                    &#34;workload.codeflare.dev/appwrapper&#34;
                    in component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;]
                ):
                    del component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;][
                        &#34;workload.codeflare.dev/appwrapper&#34;
                    ]
                    labels = component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;]
                    labels.update({&#34;kueue.x-k8s.io/queue-name&#34;: lq_name})
                    labels.update(cluster_labels)
                outfile.write(&#34;---\n&#34;)
                yaml.dump(
                    component[&#34;generictemplate&#34;], outfile, default_flow_style=False
                )
    print(f&#34;Written to: {output_file_name}&#34;)


def load_components(
    user_yaml: dict,
    name: str,
    namespace: str,
    local_queue: Optional[str],
    labels: dict,
):
    component_list = []
    components = user_yaml.get(&#34;spec&#34;, &#34;resources&#34;)[&#34;resources&#34;].get(&#34;GenericItems&#34;)
    lq_name = local_queue or get_default_kueue_name(namespace)
    cluster_labels = labels
    if not local_queue_exists(namespace, lq_name):
        raise ValueError(
            &#34;local_queue provided does not exist or is not in this namespace. Please provide the correct local_queue name in Cluster Configuration&#34;
        )
    for component in components:
        if &#34;generictemplate&#34; in component:
            if (
                &#34;workload.codeflare.dev/appwrapper&#34;
                in component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;]
            ):
                del component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;][
                    &#34;workload.codeflare.dev/appwrapper&#34;
                ]
                labels = component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;]
                labels.update({&#34;kueue.x-k8s.io/queue-name&#34;: lq_name})
                labels.update(cluster_labels)
            component_list.append(component[&#34;generictemplate&#34;])

    resources = &#34;---\n&#34; + &#34;---\n&#34;.join(
        [yaml.dump(component) for component in component_list]
    )
    user_yaml = resources
    print(f&#34;Yaml resources loaded for {name}&#34;)
    return user_yaml


def load_appwrapper(user_yaml: dict, name: str):
    user_yaml = yaml.dump(user_yaml)
    print(f&#34;Yaml resources loaded for {name}&#34;)
    return user_yaml


def generate_appwrapper(
    name: str,
    namespace: str,
    head_cpus: int,
    head_memory: int,
    head_gpus: int,
    min_cpu: int,
    max_cpu: int,
    min_memory: int,
    max_memory: int,
    gpu: int,
    workers: int,
    template: str,
    image: str,
    instascale: bool,
    mcad: bool,
    instance_types: list,
    env,
    image_pull_secrets: list,
    dispatch_priority: str,
    priority_val: int,
    write_to_file: bool,
    verify_tls: bool,
    local_queue: Optional[str],
    labels,
):
    user_yaml = read_template(template)
    appwrapper_name, cluster_name = gen_names(name)
    resources = user_yaml.get(&#34;spec&#34;, &#34;resources&#34;)
    item = resources[&#34;resources&#34;].get(&#34;GenericItems&#34;)[0]
    update_names(
        user_yaml,
        item,
        appwrapper_name,
        cluster_name,
        namespace,
    )
    update_labels(user_yaml, instascale, instance_types)
    update_priority(user_yaml, item, dispatch_priority, priority_val)
    update_custompodresources(
        item,
        min_cpu,
        max_cpu,
        min_memory,
        max_memory,
        gpu,
        workers,
        head_cpus,
        head_memory,
        head_gpus,
    )
    update_nodes(
        item,
        appwrapper_name,
        min_cpu,
        max_cpu,
        min_memory,
        max_memory,
        gpu,
        workers,
        image,
        instascale,
        env,
        image_pull_secrets,
        head_cpus,
        head_memory,
        head_gpus,
    )

    directory_path = os.path.expanduser(&#34;~/.codeflare/resources/&#34;)
    outfile = os.path.join(directory_path, appwrapper_name + &#34;.yaml&#34;)

    if write_to_file:
        if mcad:
            write_user_appwrapper(user_yaml, outfile)
        else:
            write_components(user_yaml, outfile, namespace, local_queue, labels)
        return outfile
    else:
        if mcad:
            user_yaml = load_appwrapper(user_yaml, name)
        else:
            user_yaml = load_components(user_yaml, name, namespace, local_queue, labels)
        return user_yaml</code></pre>
</details>
</section>
<section>
</section>
<section>
</section>
<section>
<h2 class="section-title" id="header-functions">Functions</h2>
<dl>
<dt id="codeflare_sdk.utils.generate_yaml.del_from_list_by_name"><code class="name flex">
<span>def <span class="ident">del_from_list_by_name</span></span>(<span>l: list, target: List[str]) ‑> list</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def del_from_list_by_name(l: list, target: typing.List[str]) -&gt; list:
    return [x for x in l if x[&#34;name&#34;] not in target]</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.gen_names"><code class="name flex">
<span>def <span class="ident">gen_names</span></span>(<span>name)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def gen_names(name):
    if not name:
        gen_id = str(uuid.uuid4())
        appwrapper_name = &#34;appwrapper-&#34; + gen_id
        cluster_name = &#34;cluster-&#34; + gen_id
        return appwrapper_name, cluster_name
    else:
        return name, name</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.generate_appwrapper"><code class="name flex">
<span>def <span class="ident">generate_appwrapper</span></span>(<span>name: str, namespace: str, head_cpus: int, head_memory: int, head_gpus: int, min_cpu: int, max_cpu: int, min_memory: int, max_memory: int, gpu: int, workers: int, template: str, image: str, instascale: bool, mcad: bool, instance_types: list, env, image_pull_secrets: list, dispatch_priority: str, priority_val: int, write_to_file: bool, verify_tls: bool, local_queue: Optional[str], labels)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def generate_appwrapper(
    name: str,
    namespace: str,
    head_cpus: int,
    head_memory: int,
    head_gpus: int,
    min_cpu: int,
    max_cpu: int,
    min_memory: int,
    max_memory: int,
    gpu: int,
    workers: int,
    template: str,
    image: str,
    instascale: bool,
    mcad: bool,
    instance_types: list,
    env,
    image_pull_secrets: list,
    dispatch_priority: str,
    priority_val: int,
    write_to_file: bool,
    verify_tls: bool,
    local_queue: Optional[str],
    labels,
):
    user_yaml = read_template(template)
    appwrapper_name, cluster_name = gen_names(name)
    resources = user_yaml.get(&#34;spec&#34;, &#34;resources&#34;)
    item = resources[&#34;resources&#34;].get(&#34;GenericItems&#34;)[0]
    update_names(
        user_yaml,
        item,
        appwrapper_name,
        cluster_name,
        namespace,
    )
    update_labels(user_yaml, instascale, instance_types)
    update_priority(user_yaml, item, dispatch_priority, priority_val)
    update_custompodresources(
        item,
        min_cpu,
        max_cpu,
        min_memory,
        max_memory,
        gpu,
        workers,
        head_cpus,
        head_memory,
        head_gpus,
    )
    update_nodes(
        item,
        appwrapper_name,
        min_cpu,
        max_cpu,
        min_memory,
        max_memory,
        gpu,
        workers,
        image,
        instascale,
        env,
        image_pull_secrets,
        head_cpus,
        head_memory,
        head_gpus,
    )

    directory_path = os.path.expanduser(&#34;~/.codeflare/resources/&#34;)
    outfile = os.path.join(directory_path, appwrapper_name + &#34;.yaml&#34;)

    if write_to_file:
        if mcad:
            write_user_appwrapper(user_yaml, outfile)
        else:
            write_components(user_yaml, outfile, namespace, local_queue, labels)
        return outfile
    else:
        if mcad:
            user_yaml = load_appwrapper(user_yaml, name)
        else:
            user_yaml = load_components(user_yaml, name, namespace, local_queue, labels)
        return user_yaml</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.get_default_kueue_name"><code class="name flex">
<span>def <span class="ident">get_default_kueue_name</span></span>(<span>namespace: str)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def get_default_kueue_name(namespace: str):
    # If the local queue is set, use it. Otherwise, try to use the default queue.
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        local_queues = api_instance.list_namespaced_custom_object(
            group=&#34;kueue.x-k8s.io&#34;,
            version=&#34;v1beta1&#34;,
            namespace=namespace,
            plural=&#34;localqueues&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)
    for lq in local_queues[&#34;items&#34;]:
        if (
            &#34;annotations&#34; in lq[&#34;metadata&#34;]
            and &#34;kueue.x-k8s.io/default-queue&#34; in lq[&#34;metadata&#34;][&#34;annotations&#34;]
            and lq[&#34;metadata&#34;][&#34;annotations&#34;][&#34;kueue.x-k8s.io/default-queue&#34;].lower()
            == &#34;true&#34;
        ):
            return lq[&#34;metadata&#34;][&#34;name&#34;]
    raise ValueError(
        &#34;Default Local Queue with kueue.x-k8s.io/default-queue: true annotation not found please create a default Local Queue or provide the local_queue name in Cluster Configuration&#34;
    )</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.is_kind_cluster"><code class="name flex">
<span>def <span class="ident">is_kind_cluster</span></span>(<span>)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def is_kind_cluster():
    try:
        config_check()
        v1 = client.CoreV1Api()
        label_selector = &#34;kubernetes.io/hostname=kind-control-plane&#34;
        nodes = v1.list_node(label_selector=label_selector)
        # If we find one or more nodes with the label, assume it&#39;s a KinD cluster
        return len(nodes.items) &gt; 0
    except Exception as e:
        print(f&#34;Error checking if cluster is KinD: {e}&#34;)
        return False</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.is_openshift_cluster"><code class="name flex">
<span>def <span class="ident">is_openshift_cluster</span></span>(<span>)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def is_openshift_cluster():
    try:
        config_check()
        for api in client.ApisApi(api_config_handler()).get_api_versions().groups:
            for v in api.versions:
                if &#34;route.openshift.io/v1&#34; in v.group_version:
                    return True
        else:
            return False
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.load_appwrapper"><code class="name flex">
<span>def <span class="ident">load_appwrapper</span></span>(<span>user_yaml: dict, name: str)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def load_appwrapper(user_yaml: dict, name: str):
    user_yaml = yaml.dump(user_yaml)
    print(f&#34;Yaml resources loaded for {name}&#34;)
    return user_yaml</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.load_components"><code class="name flex">
<span>def <span class="ident">load_components</span></span>(<span>user_yaml: dict, name: str, namespace: str, local_queue: Optional[str], labels: dict)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def load_components(
    user_yaml: dict,
    name: str,
    namespace: str,
    local_queue: Optional[str],
    labels: dict,
):
    component_list = []
    components = user_yaml.get(&#34;spec&#34;, &#34;resources&#34;)[&#34;resources&#34;].get(&#34;GenericItems&#34;)
    lq_name = local_queue or get_default_kueue_name(namespace)
    cluster_labels = labels
    if not local_queue_exists(namespace, lq_name):
        raise ValueError(
            &#34;local_queue provided does not exist or is not in this namespace. Please provide the correct local_queue name in Cluster Configuration&#34;
        )
    for component in components:
        if &#34;generictemplate&#34; in component:
            if (
                &#34;workload.codeflare.dev/appwrapper&#34;
                in component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;]
            ):
                del component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;][
                    &#34;workload.codeflare.dev/appwrapper&#34;
                ]
                labels = component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;]
                labels.update({&#34;kueue.x-k8s.io/queue-name&#34;: lq_name})
                labels.update(cluster_labels)
            component_list.append(component[&#34;generictemplate&#34;])

    resources = &#34;---\n&#34; + &#34;---\n&#34;.join(
        [yaml.dump(component) for component in component_list]
    )
    user_yaml = resources
    print(f&#34;Yaml resources loaded for {name}&#34;)
    return user_yaml</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.local_queue_exists"><code class="name flex">
<span>def <span class="ident">local_queue_exists</span></span>(<span>namespace: str, local_queue_name: str)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def local_queue_exists(namespace: str, local_queue_name: str):
    # get all local queues in the namespace
    try:
        config_check()
        api_instance = client.CustomObjectsApi(api_config_handler())
        local_queues = api_instance.list_namespaced_custom_object(
            group=&#34;kueue.x-k8s.io&#34;,
            version=&#34;v1beta1&#34;,
            namespace=namespace,
            plural=&#34;localqueues&#34;,
        )
    except Exception as e:  # pragma: no cover
        return _kube_api_error_handling(e)
    # check if local queue with the name provided in cluster config exists
    for lq in local_queues[&#34;items&#34;]:
        if lq[&#34;metadata&#34;][&#34;name&#34;] == local_queue_name:
            return True
    return False</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.read_template"><code class="name flex">
<span>def <span class="ident">read_template</span></span>(<span>template)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def read_template(template):
    with open(template, &#34;r&#34;) as stream:
        try:
            return yaml.safe_load(stream)
        except yaml.YAMLError as exc:
            print(exc)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_affinity"><code class="name flex">
<span>def <span class="ident">update_affinity</span></span>(<span>spec, appwrapper_name, instascale)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_affinity(spec, appwrapper_name, instascale):
    if instascale:
        node_selector_terms = (
            spec.get(&#34;affinity&#34;)
            .get(&#34;nodeAffinity&#34;)
            .get(&#34;requiredDuringSchedulingIgnoredDuringExecution&#34;)
            .get(&#34;nodeSelectorTerms&#34;)
        )
        node_selector_terms[0][&#34;matchExpressions&#34;][0][&#34;values&#34;][0] = appwrapper_name
        node_selector_terms[0][&#34;matchExpressions&#34;][0][&#34;key&#34;] = appwrapper_name
    else:
        spec.pop(&#34;affinity&#34;)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_custompodresources"><code class="name flex">
<span>def <span class="ident">update_custompodresources</span></span>(<span>item, min_cpu, max_cpu, min_memory, max_memory, gpu, workers, head_cpus, head_memory, head_gpus)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_custompodresources(
    item,
    min_cpu,
    max_cpu,
    min_memory,
    max_memory,
    gpu,
    workers,
    head_cpus,
    head_memory,
    head_gpus,
):
    if &#34;custompodresources&#34; in item.keys():
        custompodresources = item.get(&#34;custompodresources&#34;)
        for i in range(len(custompodresources)):
            resource = custompodresources[i]
            if i == 0:
                # Leave head node resources as template default
                resource[&#34;requests&#34;][&#34;cpu&#34;] = head_cpus
                resource[&#34;limits&#34;][&#34;cpu&#34;] = head_cpus
                resource[&#34;requests&#34;][&#34;memory&#34;] = head_memory
                resource[&#34;limits&#34;][&#34;memory&#34;] = head_memory
                resource[&#34;requests&#34;][&#34;nvidia.com/gpu&#34;] = head_gpus
                resource[&#34;limits&#34;][&#34;nvidia.com/gpu&#34;] = head_gpus

            else:
                for k, v in resource.items():
                    if k == &#34;replicas&#34; and i == 1:
                        resource[k] = workers
                    if k == &#34;requests&#34; or k == &#34;limits&#34;:
                        for spec, _ in v.items():
                            if spec == &#34;cpu&#34;:
                                if k == &#34;limits&#34;:
                                    resource[k][spec] = max_cpu
                                else:
                                    resource[k][spec] = min_cpu
                            if spec == &#34;memory&#34;:
                                if k == &#34;limits&#34;:
                                    resource[k][spec] = max_memory
                                else:
                                    resource[k][spec] = min_memory
                            if spec == &#34;nvidia.com/gpu&#34;:
                                if i == 0:
                                    resource[k][spec] = 0
                                else:
                                    resource[k][spec] = gpu
    else:
        sys.exit(&#34;Error: malformed template&#34;)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_env"><code class="name flex">
<span>def <span class="ident">update_env</span></span>(<span>spec, env)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_env(spec, env):
    containers = spec.get(&#34;containers&#34;)
    for container in containers:
        if env:
            if &#34;env&#34; in container:
                container[&#34;env&#34;].extend(env)
            else:
                container[&#34;env&#34;] = env</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_image"><code class="name flex">
<span>def <span class="ident">update_image</span></span>(<span>spec, image)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_image(spec, image):
    containers = spec.get(&#34;containers&#34;)
    for container in containers:
        container[&#34;image&#34;] = image</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_image_pull_secrets"><code class="name flex">
<span>def <span class="ident">update_image_pull_secrets</span></span>(<span>spec, image_pull_secrets)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_image_pull_secrets(spec, image_pull_secrets):
    template_secrets = spec.get(&#34;imagePullSecrets&#34;, [])
    spec[&#34;imagePullSecrets&#34;] = template_secrets + [
        {&#34;name&#34;: x} for x in image_pull_secrets
    ]</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_labels"><code class="name flex">
<span>def <span class="ident">update_labels</span></span>(<span>yaml, instascale, instance_types)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_labels(yaml, instascale, instance_types):
    metadata = yaml.get(&#34;metadata&#34;)
    if instascale:
        if not len(instance_types) &gt; 0:
            sys.exit(
                &#34;If instascale is set to true, must provide at least one instance type&#34;
            )
        type_str = &#34;&#34;
        for type in instance_types:
            type_str += type + &#34;_&#34;
        type_str = type_str[:-1]
        metadata[&#34;labels&#34;][&#34;orderedinstance&#34;] = type_str
    else:
        metadata.pop(&#34;labels&#34;)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_names"><code class="name flex">
<span>def <span class="ident">update_names</span></span>(<span>yaml, item, appwrapper_name, cluster_name, namespace)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_names(yaml, item, appwrapper_name, cluster_name, namespace):
    metadata = yaml.get(&#34;metadata&#34;)
    metadata[&#34;name&#34;] = appwrapper_name
    metadata[&#34;namespace&#34;] = namespace
    lower_meta = item.get(&#34;generictemplate&#34;, {}).get(&#34;metadata&#34;)
    lower_meta[&#34;labels&#34;][&#34;workload.codeflare.dev/appwrapper&#34;] = appwrapper_name
    lower_meta[&#34;name&#34;] = cluster_name
    lower_meta[&#34;namespace&#34;] = namespace</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_nodes"><code class="name flex">
<span>def <span class="ident">update_nodes</span></span>(<span>item, appwrapper_name, min_cpu, max_cpu, min_memory, max_memory, gpu, workers, image, instascale, env, image_pull_secrets, head_cpus, head_memory, head_gpus)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_nodes(
    item,
    appwrapper_name,
    min_cpu,
    max_cpu,
    min_memory,
    max_memory,
    gpu,
    workers,
    image,
    instascale,
    env,
    image_pull_secrets,
    head_cpus,
    head_memory,
    head_gpus,
):
    if &#34;generictemplate&#34; in item.keys():
        head = item.get(&#34;generictemplate&#34;).get(&#34;spec&#34;).get(&#34;headGroupSpec&#34;)
        head[&#34;rayStartParams&#34;][&#34;num-gpus&#34;] = str(int(head_gpus))

        worker = item.get(&#34;generictemplate&#34;).get(&#34;spec&#34;).get(&#34;workerGroupSpecs&#34;)[0]
        # Head counts as first worker
        worker[&#34;replicas&#34;] = workers
        worker[&#34;minReplicas&#34;] = workers
        worker[&#34;maxReplicas&#34;] = workers
        worker[&#34;groupName&#34;] = &#34;small-group-&#34; + appwrapper_name
        worker[&#34;rayStartParams&#34;][&#34;num-gpus&#34;] = str(int(gpu))

        for comp in [head, worker]:
            spec = comp.get(&#34;template&#34;).get(&#34;spec&#34;)
            update_affinity(spec, appwrapper_name, instascale)
            update_image_pull_secrets(spec, image_pull_secrets)
            update_image(spec, image)
            update_env(spec, env)
            if comp == head:
                # TODO: Eventually add head node configuration outside of template
                update_resources(
                    spec, head_cpus, head_cpus, head_memory, head_memory, head_gpus
                )
            else:
                update_resources(spec, min_cpu, max_cpu, min_memory, max_memory, gpu)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_priority"><code class="name flex">
<span>def <span class="ident">update_priority</span></span>(<span>yaml, item, dispatch_priority, priority_val)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_priority(yaml, item, dispatch_priority, priority_val):
    spec = yaml.get(&#34;spec&#34;)
    if dispatch_priority is not None:
        if priority_val:
            spec[&#34;priority&#34;] = priority_val
        else:
            raise ValueError(
                &#34;AW generation error: Priority value is None, while dispatch_priority is defined&#34;
            )
        head = item.get(&#34;generictemplate&#34;).get(&#34;spec&#34;).get(&#34;headGroupSpec&#34;)
        worker = item.get(&#34;generictemplate&#34;).get(&#34;spec&#34;).get(&#34;workerGroupSpecs&#34;)[0]
        head[&#34;template&#34;][&#34;spec&#34;][&#34;priorityClassName&#34;] = dispatch_priority
        worker[&#34;template&#34;][&#34;spec&#34;][&#34;priorityClassName&#34;] = dispatch_priority
    else:
        spec.pop(&#34;priority&#34;)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.update_resources"><code class="name flex">
<span>def <span class="ident">update_resources</span></span>(<span>spec, min_cpu, max_cpu, min_memory, max_memory, gpu)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def update_resources(spec, min_cpu, max_cpu, min_memory, max_memory, gpu):
    container = spec.get(&#34;containers&#34;)
    for resource in container:
        requests = resource.get(&#34;resources&#34;).get(&#34;requests&#34;)
        if requests is not None:
            requests[&#34;cpu&#34;] = min_cpu
            requests[&#34;memory&#34;] = min_memory
            requests[&#34;nvidia.com/gpu&#34;] = gpu
        limits = resource.get(&#34;resources&#34;).get(&#34;limits&#34;)
        if limits is not None:
            limits[&#34;cpu&#34;] = max_cpu
            limits[&#34;memory&#34;] = max_memory
            limits[&#34;nvidia.com/gpu&#34;] = gpu</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.write_components"><code class="name flex">
<span>def <span class="ident">write_components</span></span>(<span>user_yaml: dict, output_file_name: str, namespace: str, local_queue: Optional[str], labels: dict)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def write_components(
    user_yaml: dict,
    output_file_name: str,
    namespace: str,
    local_queue: Optional[str],
    labels: dict,
):
    # Create the directory if it doesn&#39;t exist
    directory_path = os.path.dirname(output_file_name)
    if not os.path.exists(directory_path):
        os.makedirs(directory_path)

    components = user_yaml.get(&#34;spec&#34;, &#34;resources&#34;)[&#34;resources&#34;].get(&#34;GenericItems&#34;)
    open(output_file_name, &#34;w&#34;).close()
    lq_name = local_queue or get_default_kueue_name(namespace)
    cluster_labels = labels
    if not local_queue_exists(namespace, lq_name):
        raise ValueError(
            &#34;local_queue provided does not exist or is not in this namespace. Please provide the correct local_queue name in Cluster Configuration&#34;
        )
    with open(output_file_name, &#34;a&#34;) as outfile:
        for component in components:
            if &#34;generictemplate&#34; in component:
                if (
                    &#34;workload.codeflare.dev/appwrapper&#34;
                    in component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;]
                ):
                    del component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;][
                        &#34;workload.codeflare.dev/appwrapper&#34;
                    ]
                    labels = component[&#34;generictemplate&#34;][&#34;metadata&#34;][&#34;labels&#34;]
                    labels.update({&#34;kueue.x-k8s.io/queue-name&#34;: lq_name})
                    labels.update(cluster_labels)
                outfile.write(&#34;---\n&#34;)
                yaml.dump(
                    component[&#34;generictemplate&#34;], outfile, default_flow_style=False
                )
    print(f&#34;Written to: {output_file_name}&#34;)</code></pre>
</details>
</dd>
<dt id="codeflare_sdk.utils.generate_yaml.write_user_appwrapper"><code class="name flex">
<span>def <span class="ident">write_user_appwrapper</span></span>(<span>user_yaml, output_file_name)</span>
</code></dt>
<dd>
<div class="desc"></div>
<details class="source">
<summary>
<span>Expand source code</span>
</summary>
<pre><code class="python">def write_user_appwrapper(user_yaml, output_file_name):
    # Create the directory if it doesn&#39;t exist
    directory_path = os.path.dirname(output_file_name)
    if not os.path.exists(directory_path):
        os.makedirs(directory_path)

    with open(output_file_name, &#34;w&#34;) as outfile:
        yaml.dump(user_yaml, outfile, default_flow_style=False)

    print(f&#34;Written to: {output_file_name}&#34;)</code></pre>
</details>
</dd>
</dl>
</section>
<section>
</section>
</article>
<nav id="sidebar">
<h1>Index</h1>
<div class="toc">
<ul></ul>
</div>
<ul id="index">
<li><h3>Super-module</h3>
<ul>
<li><code><a title="codeflare_sdk.utils" href="index.html">codeflare_sdk.utils</a></code></li>
</ul>
</li>
<li><h3><a href="#header-functions">Functions</a></h3>
<ul class="">
<li><code><a title="codeflare_sdk.utils.generate_yaml.del_from_list_by_name" href="#codeflare_sdk.utils.generate_yaml.del_from_list_by_name">del_from_list_by_name</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.gen_names" href="#codeflare_sdk.utils.generate_yaml.gen_names">gen_names</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.generate_appwrapper" href="#codeflare_sdk.utils.generate_yaml.generate_appwrapper">generate_appwrapper</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.get_default_kueue_name" href="#codeflare_sdk.utils.generate_yaml.get_default_kueue_name">get_default_kueue_name</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.is_kind_cluster" href="#codeflare_sdk.utils.generate_yaml.is_kind_cluster">is_kind_cluster</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.is_openshift_cluster" href="#codeflare_sdk.utils.generate_yaml.is_openshift_cluster">is_openshift_cluster</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.load_appwrapper" href="#codeflare_sdk.utils.generate_yaml.load_appwrapper">load_appwrapper</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.load_components" href="#codeflare_sdk.utils.generate_yaml.load_components">load_components</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.local_queue_exists" href="#codeflare_sdk.utils.generate_yaml.local_queue_exists">local_queue_exists</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.read_template" href="#codeflare_sdk.utils.generate_yaml.read_template">read_template</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_affinity" href="#codeflare_sdk.utils.generate_yaml.update_affinity">update_affinity</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_custompodresources" href="#codeflare_sdk.utils.generate_yaml.update_custompodresources">update_custompodresources</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_env" href="#codeflare_sdk.utils.generate_yaml.update_env">update_env</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_image" href="#codeflare_sdk.utils.generate_yaml.update_image">update_image</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_image_pull_secrets" href="#codeflare_sdk.utils.generate_yaml.update_image_pull_secrets">update_image_pull_secrets</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_labels" href="#codeflare_sdk.utils.generate_yaml.update_labels">update_labels</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_names" href="#codeflare_sdk.utils.generate_yaml.update_names">update_names</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_nodes" href="#codeflare_sdk.utils.generate_yaml.update_nodes">update_nodes</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_priority" href="#codeflare_sdk.utils.generate_yaml.update_priority">update_priority</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.update_resources" href="#codeflare_sdk.utils.generate_yaml.update_resources">update_resources</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.write_components" href="#codeflare_sdk.utils.generate_yaml.write_components">write_components</a></code></li>
<li><code><a title="codeflare_sdk.utils.generate_yaml.write_user_appwrapper" href="#codeflare_sdk.utils.generate_yaml.write_user_appwrapper">write_user_appwrapper</a></code></li>
</ul>
</li>
</ul>
</nav>
</main>
<footer id="footer">
<p>Generated by <a href="https://pdoc3.github.io/pdoc" title="pdoc: Python API documentation generator"><cite>pdoc</cite> 0.10.0</a>.</p>
</footer>
</body>
</html>