def import_variables()

in source/generate_dag.py [0:0]

• 28 lines of code
• 8 McCabe index (conditional complexity)

def import_variables(yaml_config: dict) -> dict:
    """
    Generate a dictionary for importing variables into a DAG.

    This function determines how to import variables based on the
    provided YAML configuration. It can import variables from a specified file, or from the YAML configuration. It also checks if 'default_args' is defined and not empty in the variables file.

    Args:
        yaml_config (dict): YAML configuration for the task.

    Returns:
        dict: A dictionary containing:
            - add_variables (bool): Flag indicating whether to add variables.
            - variables (list): A list of variables with preserved indentation.
            - variable_keys (list): A list of unique variable keynames from the imported variables
            - valid_default_args (bool): True if 'default_args' is defined and not empty, False otherwise.
    """
    variables = []
    variable_keys = []
    valid_default_args = False  # Initialize the flag here
    try:
        task_variables = yaml_config['task_variables']
        if not isinstance(task_variables, dict):
            raise TypeError("'task_variables' should be a dictionary.")

        if task_variables.get('variables_file_path'):
            file_path = task_variables['variables_file_path']
            variable_keys = extract_keys_from_py_file(file_path)
            print(f"Importing variables: reading variables from file {file_path}")

            # Check for valid 'default_args' here
            valid_default_args = has_valid_default_args(file_path)

            with open(file_path, 'r') as file:
                file_content = file.readlines()  # Read all lines into a list
                variables = [line for line in file_content if '# type: ignore' not in line]

    except (KeyError, TypeError) as e:
        print(f"Importing Variables: Error processing variables file: {e}")

    if variables:
        return {
            "add_variables": True,
            "variables": variables,
            "variable_keys": variable_keys,
            "valid_default_args": valid_default_args
        }
    else:
        print("Importing variables: Skipping variable import.")
        return {"add_variables": False}