def list_annotations()

in esrally/metrics.py [0:0]

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

    def list_annotations(self):
        environment = self.environment_name
        track = self._track()
        from_date = self._from_date()
        to_date = self._to_date()
        query = {
            "query": {
                "bool": {
                    "filter": [
                        {"term": {"environment": environment}},
                        {"range": {"race-timestamp": {"gte": from_date, "lte": to_date, "format": "basic_date"}}},
                    ]
                }
            }
        }
        if track:
            query["query"]["bool"]["filter"].append({"term": {"track": track}})

        query["sort"] = [{"race-timestamp": "desc"}, {"track": "asc"}, {"chart": "asc"}]
        query["size"] = self._max_results()

        result = self.client.search(index="rally-annotations", body=query)
        annotations = []
        hits = result["hits"]["total"]
        if hits == 0:
            console.println(f"No annotations found in environment [{environment}].")
        else:
            for hit in result["hits"]["hits"]:
                src = hit["_source"]
                annotations.append(
                    [
                        hit["_id"],
                        src["race-timestamp"],
                        src.get("track", ""),
                        src.get("chart", ""),
                        src.get("chart-name", ""),
                        src["message"],
                    ]
                )

            if annotations:
                console.println("\nAnnotations:\n")
                console.println(
                    tabulate.tabulate(
                        annotations,
                        headers=["Annotation Id", "Timestamp", "Track", "Chart Type", "Chart Name", "Message"],
                    )
                )