def _process_bq_object_settings()

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

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

def _process_bq_object_settings(global_settings: dict,
                                bq_objects_settings: list,
                                task_dep_objs: dict[str, dag_types.BqObject],
                                wait_for_prev_step: bool) -> list:
    """Creates a list containing build file settings from bq_object settings."""
    build_file_settings = []
    for bq_object_setting in bq_objects_settings:
        logging.debug("bq_object_setting = '%s'", bq_object_setting)

        object_type = bq_object_setting.get("type")

        if object_type in ["table", "view", "script"]:
            sql_file = bq_object_setting.get("sql_file")
            build_file_settings.append({
                "sql_file": sql_file,
                "wait_for_prev_step": wait_for_prev_step,
                "is_task_dep": sql_file in task_dep_objs,
                # NOTE: Using json.dumps to handle double quotes in settings
                "bq_object_setting": json.dumps(bq_object_setting)
            })
        elif object_type == "k9_dawg":
            k9_id = bq_object_setting.get("k9_id")
            manifest = global_settings["k9_manifest"]
            try:
                k9_definition = manifest[k9_id]
            except Exception as e:
                raise ValueError(f"K9 {k9_id} not found in "
                                 "local manifest.") from e

            k9_settings = bq_object_setting.get("parameters", {})

            build_file_settings.append({
                "k9_id": k9_id,
                "wait_for_prev_step": wait_for_prev_step,
                # NOTE: Using json.dumps to handle double quotes in settings
                "k9_definition": json.dumps(k9_definition),
                "k9_settings": json.dumps(k9_settings)
            })
    return build_file_settings