def fetch_data()

in otava/importer.py [0:0]

• 59 lines of code
• 23 McCabe index (conditional complexity)

    def fetch_data(self, test_conf: TestConfig, selector: DataSelector = DataSelector()) -> Series:

        if not isinstance(test_conf, JsonTestConfig):
            raise ValueError("Expected JsonTestConfig")

        # TODO: refactor. THis is copy pasted from CSV importer
        since_time = selector.since_time
        until_time = selector.until_time

        if since_time.timestamp() > until_time.timestamp():
            raise DataImportError(
                f"Invalid time range: ["
                f"{format_timestamp(int(since_time.timestamp()))}, "
                f"{format_timestamp(int(until_time.timestamp()))}]"
            )

        time = []
        data = OrderedDict()
        metrics = OrderedDict()
        attributes = OrderedDict()

        for name in self.fetch_all_metric_names(test_conf):
            # Ignore metrics if selector.metrics is not None and name is not in selector.metrics
            if selector.metrics is not None and name not in selector.metrics:
                continue
            data[name] = []

        attr_names = self.fetch_all_attribute_names(test_conf)
        for name in attr_names:
            attributes[name] = []

        # If the user specified a branch, only include results from that branch.
        # Otherwise if the test config specifies a branch, only include results from that branch.
        # Else include all results.
        branch = None
        if selector.branch:
            branch = selector.branch
        elif test_conf.base_branch:
            branch = test_conf.base_branch

        objs = self.inputfile(test_conf)
        list_of_json_obj = []
        for o in objs:
            if branch and o["attributes"]["branch"] != branch:
                continue
            list_of_json_obj.append(o)

        for result in list_of_json_obj:
            time.append(result["timestamp"])
            for metric in result["metrics"]:
                # Skip metrics not in selector.metrics if selector.metrics is enabled
                if metric["name"] not in data:
                    continue

                data[metric["name"]].append(metric["value"])
                metrics[metric["name"]] = Metric(1, 1.0)
        for a in attr_names:
            attributes[a] = [o["attributes"][a] for o in list_of_json_obj]

        # Leave last n points:
        time = time[-selector.last_n_points :]
        tmp = data
        data = {}
        for k, v in tmp.items():
            data[k] = v[-selector.last_n_points :]
        tmp = attributes
        attributes = {}
        for k, v in tmp.items():
            attributes[k] = v[-selector.last_n_points :]

        return Series(
            test_conf.name,
            branch=None,
            time=time,
            metrics=metrics,
            data=data,
            attributes=attributes,
        )