def gen_kubeflow_config()

in cli/aws_orbit/remote_files/kubeflow.py [0:0]

• 55 lines of code
• 3 McCabe index (conditional complexity)

def gen_kubeflow_config(context: Context, output_path: str, cluster_name: str) -> None:

    os.makedirs(output_path, exist_ok=True)
    _cleanup_output(output_path=output_path)
    if context.account_id is None:
        raise RuntimeError("context.account_id is None!")
    if context.region is None:
        raise RuntimeError("context.region is None!")

    input = os.path.join(CONFIG_PATH, "kfctl_aws.yaml")
    output = os.path.join(output_path, "kfctl_aws.yaml")

    client = boto3_client(service_name="cognito-idp")
    response: Dict[str, Any] = client.describe_user_pool(UserPoolId=context.user_pool_id)
    domain: str = response["UserPool"].get("Domain")

    with open(input, "r") as file:
        content: str = file.read()

    content = utils.resolve_parameters(
        content,
        dict(
            certArn=context.networking.frontend.ssl_cert_arn,
            cognitoAppClientId=context.user_pool_client_id,
            cognitoUserPoolID=context.user_pool_id,
            account_id=context.account_id,
            region=context.region,
            env_name=context.name,
            cluster_name=cluster_name,
            cognitoUserPoolDomain=domain,
        ),
    )
    _logger.debug("Kubeflow configuration:\n%s", content)
    with open(output, "w") as file:
        file.write(content)

    k8s_context = get_k8s_context(context=context)

    input = os.path.join(CONFIG_PATH, "apply_kf.sh")
    output = os.path.join(output_path, "apply_kf.sh")

    with open(input, "r") as file:
        content = file.read()

    content = utils.resolve_parameters(
        content,
        dict(cluster_name=cluster_name, k8s_context=k8s_context),
    )
    _logger.debug("Kubeflow script:\n%s", content)
    with open(output, "w") as file:
        file.write(content)

    sh.run(f"chmod a+x  {output}")

    input = os.path.join(CONFIG_PATH, "delete_kf.sh")
    output = os.path.join(output_path, "delete_kf.sh")

    with open(input, "r") as file:
        content = file.read()

    content = utils.resolve_parameters(
        content,
        dict(cluster_name=cluster_name, k8s_context=k8s_context),
    )
    _logger.debug("Kubeflow script:\n%s", content)
    with open(output, "w") as file:
        file.write(content)

    sh.run(f"chmod a+x  {output}")