def apply_config_parameters_to_all_queries()

in tasks.py [0:0]

• 15 lines of code
• 5 McCabe index (conditional complexity)

def apply_config_parameters_to_all_queries(c, env_name="prod"):
    """
    Applies parameters from a configuration file to .sqlx query template files,
    generating new .sql files for each template.

    Args:
        c (object): An object providing context for task execution (e.g., task runner).
        env_name (str, optional): Name of the environment to use for configuration. Defaults to "prod".

    Raises:
        FileNotFoundError: If the specified configuration file is not found.
        ValueError: If the configuration file is invalid or missing required keys.
        TemplateError: If an error occurs during template rendering.

    Example usage:
        apply_config_parameters_to_all_queries(my_task_context, env_name="dev")
    
    Customization:
        1) Add a new parameter to .sqlx template file: Add the placeholder: 
            a) Insert the new parameter's placeholder within the template file, surrounded 
            by Jinja2's delimiters (e.g., {{ new_parameter }}) where you want its value to appear 
            in the rendered output.
            b) Add a new key-value pair for the new parameter under the appropriate section of 
            your config.yaml.tftpl YAML configuration file. Ensure the key matches the placeholder 
            name in the template.
            c) No code changes required: The existing code already retrieves configuration values 
            and renders them into templates, so it should handle the new parameter without modification.
    """
    # Load configuration file according to environment name
    current_path = Path(__file__).parent.resolve()
    conf = yaml.safe_load(Path.joinpath(current_path,"config", "{}.yaml".format(env_name)).read_text())
    
    # Fetch all query configs <key,value> to be applied
    query_dict = conf['bigquery']['query']

    # Locate file path for all templates to be used
    template_path = Path.joinpath(current_path,"sql","query")
    templateLoader = FileSystemLoader(searchpath=template_path)
    templateEnv = Environment(loader=templateLoader)
    
    # For each file name, apply the config values to the template
    for template_file in template_path.iterdir(): 
        if template_file.is_file() and template_file.resolve().suffix == '.sqlx':
            template = templateEnv.get_template(template_file.name)
            new_sql = template.render(query_dict[template_file.stem])
            rendered_sql_file = Path.joinpath(template_path, template_file.resolve().stem+".sql")
            with rendered_sql_file.open("w+", encoding ="utf-8") as f:
                f.write(new_sql)
            print("New SQL file rendered at {}".format(rendered_sql_file))