# # All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or # its licensors. # # For complete copyright and license terms please see the LICENSE at the root of this # distribution (the "License"). All use of this software is governed by the License, # or, if provided, by the license below or the license accompanying this file. Do not # remove or modify any license notices. This file is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # import os import sys import traceback, errno from az_code_gen.base import * from azcg_extension import * import jinja_extensions import jinja2.exceptions import jinja2.utils # --------------------------------------------------- # Prepare the Jinja2 template engine. def load_jinja2_environment(script_paths): try: from jinja2 import Environment, FileSystemLoader, StrictUndefined, DebugUndefined except Exception as e: print('base.py error loading jinja2 environment: {}'.format(e)) raise # wrap the FileSystemLoader so we can seed it with script paths and record what it requests class AZCGLoader(FileSystemLoader): def __init__(self, scripts): script_paths = scripts.split(',') paths = [] for script_path in script_paths: script_dir = os.path.dirname(script_path) if script_dir not in paths: paths.append(script_dir) # feed the unique script dirs as input dirs to FileSystemLoader super(AZCGLoader, self).__init__(paths) def get_source(self, environment, template): for searchpath in self.searchpath: template_filename = os.path.join(searchpath, template) template_file = jinja2.utils.open_if_exists(template_filename) if template_file is None: continue try: contents = template_file.read().decode(self.encoding) finally: template_file.close() mtime = os.path.getmtime(template_filename) # Record dependency to utility RegisterDependencyFile(template_filename) # Allow relative template paths to this template, if not already a search path template_path = os.path.dirname(template_filename) if template_path not in self.searchpath: self.searchpath.append(template_path) def uptodate(): try: return os.path.getmtime(template_filename) == mtime except OSError: return False return contents, template_filename, uptodate raise jinja2.exceptions.TemplateNotFound(template) jinja_env = Environment(loader=AZCGLoader(script_paths), # strip whitespace from templates as much as possible trim_blocks=True, lstrip_blocks=True, # throws an exception if variables are undefined. Setting this to DebugUndefined will print the error message into the generated file instead undefined=StrictUndefined, extensions=[jinja_extensions.error.RaiseExtension, "jinja2.ext.do"]) jinja_extensions.template.registerExtensions(jinja_env) return jinja_env # --------------------------------------------------- def exception_name(exc): return '{}.{}'.format(type(exc).__module__, type(exc).__name__) def generate_output_writer(output_path, output_files): def output_writer(output_file_name, output_data, should_add_to_build): output_file_path = os.path.abspath(os.path.join(output_path, output_file_name)) output_file_desc = None if os.name is 'nt' and len(output_file_path) >= 260: OutputError('Unable to write generated file output, path exceeds MAX_PATH limit of 260 characters. Please use a shorter root path. Length: {} - Path: {}'.format(len(output_file_path), output_file_path)) return if output_file_path not in output_files: output_file_dir = os.path.dirname(output_file_path) if not os.path.isdir(output_file_dir): try: os.makedirs(output_file_dir) except OSError as ex: # it is possible for multiple codegen processes to attempt to create the same # output directory. If the directory exists, just accept it and move on if ex.errno != errno.EEXIST or not os.path.isdir(output_file_dir): raise ex; output_file_desc = os.open(output_file_path, os.O_CREAT | os.O_TRUNC | os.O_WRONLY) output_files[output_file_path] = output_file_desc else: output_file_desc = output_files[output_file_path] os.write(output_file_desc, str.encode(output_data)) RegisterOutputFile(output_file_path, should_add_to_build) return output_writer # This is the function called by the codegen utility # @return boolean - True if the script execution was successful, False if execution failed # @param rootPath - The path to the source script, necessary to find the modules at runtime # @param script - The script to run, this will load the script as a python module # @param dataObject - The data for the script to operate on, this can be JSON/XML or some custom binary format. # @param inputPath - This is the absolute path where the inputFile is relative to # @param outputPath - This is where the output file are required to go, but we may add folders, if necessary # @param inputFile - This is the relative path to the input file we processed to get the dataObject def run_scripts(scripts, data_object, input_path, output_path, input_file): # We want to catch any exception that happens and get some output to the user try: env = create_environment() # Setup the jinja2 environment env.jinja_env = load_jinja2_environment(scripts) # Track output files output_files = {} # Set the output writer env.output_writer = generate_output_writer(output_path, output_files) script_array = scripts.split(',') was_successful = True for script in script_array: drivers = env.get_drivers_for_script(script) for driver in drivers: result = driver.execute(data_object, input_file) if not result: was_successful = False break # Close open file handles for output_file_path, output_file_desc in output_files.items(): os.close(output_file_desc) OutputPrint("Done - {}".format("Successful" if was_successful else "Failed")) return was_successful except BaseException as ex: tb_list = traceback.extract_tb(sys.exc_info()[2]) for filename, lineno, name, line in tb_list: OutputError('{}({}): error {}: in {}: {}'.format(filename, lineno, exception_name(ex), name, line)) filename, lineno, _, _ = tb_list[-1] OutputError('{}({}): error {}: {}'.format(filename, lineno, exception_name(ex), str(ex))) raise