def get_materializer_settings()

in common/materializer/generate_assets.py [0:0]

• 23 lines of code
• 6 McCabe index (conditional complexity)

def get_materializer_settings(materializer_settings_file: str) -> dict:
    """Parses settings file and returns settings dict after validations."""
    logging.info("Loading Materializer settings file '%s'...",
                 materializer_settings_file)

    with open(materializer_settings_file,
              encoding="utf-8") as materializer_settings_fp:
        materializer_settings = yaml.safe_load(materializer_settings_fp)

    if materializer_settings is None:
        raise ValueError(f"🛑 '{materializer_settings_file}' file is empty.")

    logging.debug("Materializer settings for this module : \n%s",
                  json.dumps(materializer_settings, indent=4))

    # Validate bq object settings.
    # Since this setting file contains two separate bq table setting sections,
    # we validate both of them.
    independent_tables_settings = materializer_settings.get(
        "bq_independent_objects")
    dependent_tables_settings = materializer_settings.get(
        "bq_dependent_objects")

    # At least one of the two sections needs to be present.
    if (independent_tables_settings is None and
            dependent_tables_settings is None):
        raise ValueError(
            "🛑 'bq_independent_objects' and 'bq_dependent_setting' both "
            "can not be empty.")

    for settings in [independent_tables_settings, dependent_tables_settings]:
        if settings:
            bq_materializer.validate_bq_materializer_settings(settings)

    return materializer_settings