def fetch_data()

in otava/graphite.py [0:0]

• 26 lines of code
• 4 McCabe index (conditional complexity)

    def fetch_data(self, target_paths: List[str], selector: DataSelector) -> List[TimeSeries]:
        """
        Connects to Graphite server and downloads interesting series with the
        given prefix. The series to be downloaded are picked from SUFFIXES list.
        """
        try:
            info("Fetching data from Graphite...")
            result = []

            from_time = to_graphite_time(selector.since_time, "-365d")
            until_time = to_graphite_time(selector.until_time, "now")
            target_paths = compress_target_paths(target_paths)
            targets = ""
            for path in target_paths:
                targets += f"target={path}&"
            targets = targets.strip("&")

            url = (
                f"{self.__url}render"
                f"?{targets}"
                f"&format=json"
                f"&from={from_time}"
                f"&until={until_time}"
            )

            data_str = urllib.request.urlopen(url).read()
            data_as_json = json.loads(data_str)

            for s in data_as_json:
                series = TimeSeries(path=s["target"], points=decode_graphite_datapoints(s))
                result.append(series)

            return result

        except IOError as err:
            raise GraphiteError(f"Failed to fetch data from Graphite: {str(err)}")