def _specs_from_directory()

in common/data_mesh/deploy_data_mesh.py [0:0]

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

def _specs_from_directory(directory: str, spec_type: Type[T],
                          jinja_dict: Dict[str, Any]) -> List[T]:

    logging.info("Loading specs in directory: %s.", directory)

    dir_path = pathlib.Path(directory)
    if not dir_path.exists():
        logging.warning("The following spec directory doesn't exist: %s.",
                        directory)
        return []

    # Load all YAML files from a directory into a single list of specs.
    specs = []
    for file_path in dir_path.iterdir():
        if file_path.suffix != ".yaml":
            logging.info("Ignoring non-YAML file: '%s'", file_path)
            continue

        # TODO: catch cases where config isnt set properly.
        # Can result in missing jinja dict keys. Alternatively, set the jinja
        # dict vars regardless of config options.
        try:
            final_yaml = jinja.apply_jinja_params_dict_to_file(
                file_path, jinja_dict)
            spec_dict = yaml.safe_load(final_yaml)
            specs.append(spec_type.from_dict(spec_dict))  # type: ignore
        except Exception as e:
            raise cortex_exc.CriticalError(
                f"Error while attempting to load {file_path}.") from e

    return specs