def update_grafana_annotations()

in otava/main.py [0:0]
65 lines of code
17 McCabe index (conditional complexity)

    def update_grafana_annotations(self, test: GraphiteTestConfig, series: AnalyzedSeries):
        grafana = self.__get_grafana()
        begin = datetime.fromtimestamp(series.time()[0], tz=pytz.UTC)
        end = datetime.fromtimestamp(series.time()[len(series.time()) - 1], tz=pytz.UTC)

        logging.info(f"Fetching Grafana annotations for test {test.name}...")
        tags_to_query = ["otava", "change-point", "test:" + test.name]
        old_annotations_for_test = grafana.fetch_annotations(begin, end, list(tags_to_query))
        logging.info(f"Found {len(old_annotations_for_test)} annotations")

        created_count = 0
        for metric_name, change_points in series.change_points.items():
            path = test.get_path(series.branch_name(), metric_name)
            metric_tag = f"metric:{metric_name}"
            tags_to_create = (
                tags_to_query
                + [metric_tag]
                + test.tags
                + test.annotate
                + test.metrics[metric_name].annotate
            )

            substitutions = {
                "TEST_NAME": test.name,
                "METRIC_NAME": metric_name,
                "GRAPHITE_PATH": [path],
                "GRAPHITE_PATH_COMPONENTS": path.split("."),
                "GRAPHITE_PREFIX": [test.prefix],
                "GRAPHITE_PREFIX_COMPONENTS": test.prefix.split("."),
            }

            tmp_tags_to_create = []
            for t in tags_to_create:
                tmp_tags_to_create += interpolate(t, substitutions)
            tags_to_create = tmp_tags_to_create

            old_annotations = [a for a in old_annotations_for_test if metric_tag in a.tags]
            old_annotation_times = set((a.time for a in old_annotations if a.tags))

            target_annotations = []
            for cp in change_points:
                attributes = series.attributes_at(cp.index)
                annotation_text = get_back_links(attributes)
                target_annotations.append(
                    Annotation(
                        id=None,
                        time=datetime.fromtimestamp(cp.time, tz=pytz.UTC),
                        text=annotation_text,
                        tags=tags_to_create,
                    )
                )
            target_annotation_times = set((a.time for a in target_annotations))

            to_delete = [a for a in old_annotations if a.time not in target_annotation_times]
            if to_delete:
                logging.info(
                    f"Removing {len(to_delete)} annotations "
                    f"for test {test.name} and metric {metric_name}..."
                )
                grafana.delete_annotations(*(a.id for a in to_delete))

            to_create = [a for a in target_annotations if a.time not in old_annotation_times]
            if to_create:
                logging.info(
                    f"Creating {len(to_create)} annotations "
                    f"for test {test.name} and metric {metric_name}..."
                )
                grafana.create_annotations(*to_create)
                created_count += len(to_create)

        if created_count == 0:
            logging.info("All annotations up-to-date. No new annotations needed.")
        else:
            logging.info(f"Created {created_count} annotations.")