ci/pipelines/render.py

#!/usr/bin/env python3

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); 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 "AS IS" 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.
#

"""This script performs a multi-pass render of a Jinja templated file.
See the argparse help for details."""
import argparse
import jinja2.exceptions
import logging
import os
import pprint
import yaml
from itertools import combinations
from jinja2 import Environment, FileSystemLoader, Undefined
from typing import List, Dict

RENDER_PASS_LIMIT = 5


class RaiseExceptionIfUndefined(Undefined):
    """If a variable is not defined, do not interpolate it to be an empty string per Jinja's default behavior.
    Rather, the render script should fail vocally.

    The actual failure must be raised in __str__, when the renderer attempts to interpolate this value.
    """

    def __str__(self):
        raise jinja2.exceptions.UndefinedError(f"Could not find variable definition for '{self._undefined_name}'.")


def render_template(template_file: str,
                    variables_files: List[str], environment_dirs: List[str],
                    command_line_variable_options: List[Dict],
                    print_rendered: bool, output_file: str):
    env = get_environment(environment_dirs, template_file)
    variables = determine_variables(command_line_variable_options, variables_files)
    template = env.get_template(os.path.basename(template_file))
    rendered_template = multipass_render(template, variables, env)

    if print_rendered:
        print(rendered_template)

    logging.debug(f"Writing to output '{output_file}'")
    with open(output_file, 'w') as pipeline_file:
        pipeline_file.write(rendered_template)


def get_environment(environment_dirs, template_file):
    """Gets the Jinja Environment for the renderer.
    Uses a FileSystemLoader with the specified environment dirs, as well as the directory containing the template_file.
    Raises an error if any variable cannot be resolved."""
    environment_searchpath = [get_absolute_dirname(template_file)] + environment_dirs
    logging.debug(f"Environment search path:\n{pprint.pformat(environment_searchpath)}")
    env = Environment(loader=FileSystemLoader(environment_searchpath),
                      extensions=['jinja2.ext.do'],
                      undefined=RaiseExceptionIfUndefined,
                      )
    return env


def multipass_render(template, variables, environment):
    f"""Performs a multiple-pass render of the template file, allowing variables to reference other variables.
    Performs at most RENDER_PASS_LIMIT (={RENDER_PASS_LIMIT}) passes.
    """
    this_template = template

    for i in range(1, RENDER_PASS_LIMIT + 1):
        logging.debug(f"Performing render pass {i}...")

        this_template_rendered = this_template.render(variables)
        if "{{" not in this_template_rendered:
            return this_template_rendered
        this_template = environment.from_string(this_template_rendered)

    raise RuntimeError(f"Variables not eliminated from template after {RENDER_PASS_LIMIT} passes."
                       f"  Please verify that template variables references to not recurse.")


def determine_variables(command_line_variable_options: List[Dict], variables_files):
    """Merges variables from each specified variable file and any command-line specified variables,
     warning when conflicts exist."""
    # For safety purposes, we first generate a list of tuples for each specified variable,
    # where each tuple is of the form (source, key, value).
    variable_list = [(source, key, value)
                     for source in variables_files
                     for key, value in get_variables_from_file(source).items()]
    # Add any command-line variables last to give them the highest priority
    variable_list.extend([
        ("command-line option", k, v)
        for cmd_opt in command_line_variable_options
        for k, v in cmd_opt.items()
    ])
    # Scan though the variable set to make sure there are no conflicts
    for (s1, k1, v1), (s2, k2, v2) in combinations(variable_list, 2):
        if k1 == k2 and v1 == v2:
            logging.info(f"Key value pair ({k1}: {v1}) is duplicated in variable files {s1} and {s2}")
        if k1 == k2 and v1 != v2:
            logging.warning(f"Conflicting variable specifications provided.\n"
                            f"Key value pair from {s1}:\n"
                            f"  {k1}: {v1}\n"
                            f"has been overwritten by {s2}:\n"
                            f"  {k2}: {v2}")

    # Here as above, we maintain load order so that later-specified variable files will overwrite earlier-specified.
    variables = {key: value for _, key, value in variable_list}

    logging.debug(f"Variables:\n{pprint.pformat(variables)}\n")

    return variables


def get_variables_from_file(absolute_var_file):
    logging.debug(f"Loading variables from file {absolute_var_file}")
    with open(absolute_var_file, 'r') as variablesFromYml:
        return yaml.load(variablesFromYml)


def get_absolute_dirname(some_path):
    """If the path given is a directory, returns the absolute path to that directory.
    If the path given is a file, returns the absolute path to the directory containing that file.
    Errors if the path does not exist or could not otherwise resolve."""
    absolute_path = os.path.abspath(some_path)
    if os.path.isfile(absolute_path):
        return os.path.abspath(os.path.dirname(os.path.abspath(absolute_path)))

    if os.path.isdir(absolute_path):
        return os.path.abspath(absolute_path)

    if not os.path.exists(absolute_path):
        raise FileNotFoundError(f"Path '{absolute_path}' does not exist.")

    raise ValueError(f"Something strange happened attempting to resolve the absolute dirname of '{absolute_path}'")


if __name__ == '__main__':
    # All files / directories specified are converted to absolute paths by the ArgParser
    parser = argparse.ArgumentParser()
    parser.add_argument("template",
                        help="Jinja template file.",
                        type=lambda s: os.path.abspath(s)
                        )
    parser.add_argument("--variable-file",
                        help="Jinja variables file.  Multiple files can be provided."
                             "  You will be warned if any definitions collide."
                             "  Preference will be given to the file specified last.",
                        nargs='+',
                        default=[],
                        type=lambda s: os.path.abspath(s))
    parser.add_argument("--environment",
                        help="Additional directories to pass to the Jinja environment, e.g.,"
                             " a 'shared/' directory containing files referenced by name,"
                             " e.g., '{%% from shared_jinja.yml import x with context %%}'",
                        nargs="+",
                        default=[],
                        type=get_absolute_dirname
                        )
    parser.add_argument("--raw-variable",
                        help="YML or JSON string defining additional variables."
                             "  This option has preference greater than variable files.",
                        nargs="+",
                        default=[],
                        type=lambda s: yaml.load(s))
    parser.add_argument("-o",
                        "--output",
                        help="Output file",
                        default=f"generated-pipeline.yml",
                        type=lambda s: os.path.abspath(s))
    parser.add_argument("--print-to-console",
                        help="Prints the rendered file before writing to --output",
                        action="store_true")
    parser.add_argument("--debug",
                        help="It's debug.  If you have to ask, you'll never know.",
                        action="store_true")
    parser.add_argument("--info",
                        help="It's info.  It's like a lazy --debug.",
                        action="store_true")

    _args = parser.parse_args()

    if _args.debug:
        logging.getLogger().setLevel(logging.DEBUG)
    elif _args.info:
        logging.getLogger().setLevel(logging.INFO)
    else:
        logging.getLogger().setLevel(logging.WARNING)

    # Preprocess file arguments so that we only deal with absolute paths.
    render_template(_args.template,
                    _args.variable_file,
                    _args.environment,
                    _args.raw_variable,
                    _args.print_to_console,
                    _args.output)