def process_kibana_object()

in helper-scripts/export_kibana_config.py [0:0]

• 58 lines of code
• 9 McCabe index (conditional complexity)

    def process_kibana_object(self, obj_type, indexpattern=None):
        """
        Create json from ndjson kibana object to ease diff during commits
        """
        print(f"# Processing kibana object: {obj_type}")

        if obj_type != "index-pattern":
            src_file_name = f"{EXPORT_FILES_PREFIX_KIBANA}{obj_type}"
        else:
            if indexpattern is None:
                for i in INDEX_PATTERNS_FILTER.split("|"):
                    self.process_kibana_object(obj_type, indexpattern=i)
                return
            else:
                src_file_name = f"{EXPORT_FILES_PREFIX_KIBANA}{obj_type}_{indexpattern}"

        src_file = os.path.join(self.export_path, f"{src_file_name}.ndjson")
        diff_file = os.path.join(self.export_path, DIFF_PATH, f"{src_file_name}.json")
        print(f"\tOpening {obj_type}: {src_file}")
        with open(src_file, "r", encoding="utf-8") as src_ndjson_file:
            src_ndjson = ndjson.load(src_ndjson_file)

        for src_ndjson_line in src_ndjson:
            if obj_type == "index-pattern":
                src_ndjson_line["attributes"]["fields"] = sorted(
                    json.loads(src_ndjson_line["attributes"]["fields"]),
                    key=lambda x: x["name"],
                )
            elif obj_type == "search":
                src_ndjson_line["attributes"]["kibanaSavedObjectMeta"][
                    "searchSourceJSON"
                ] = json.loads(
                    src_ndjson_line["attributes"]["kibanaSavedObjectMeta"][
                        "searchSourceJSON"
                    ]
                )
            elif obj_type == "visualization":
                src_ndjson_line["attributes"]["kibanaSavedObjectMeta"][
                    "searchSourceJSON"
                ] = json.loads(
                    src_ndjson_line["attributes"]["kibanaSavedObjectMeta"][
                        "searchSourceJSON"
                    ]
                )
                src_ndjson_line["attributes"]["visState"] = json.loads(
                    src_ndjson_line["attributes"]["visState"]
                )
            elif obj_type == "dashboard":
                src_ndjson_line["attributes"]["kibanaSavedObjectMeta"][
                    "searchSourceJSON"
                ] = json.loads(
                    src_ndjson_line["attributes"]["kibanaSavedObjectMeta"][
                        "searchSourceJSON"
                    ]
                )
                src_ndjson_line["attributes"]["optionsJSON"] = json.loads(
                    src_ndjson_line["attributes"]["optionsJSON"]
                )
                src_ndjson_line["attributes"]["panelsJSON"] = json.loads(
                    src_ndjson_line["attributes"]["panelsJSON"]
                )

        print(f"\tWriting output to: {diff_file}")
        with open(diff_file, "w", encoding="utf-8") as dst_json_file:
            json.dump(src_ndjson, dst_json_file, indent=4, sort_keys=True)