def target_throughput()

in esrally/track/track.py [0:0]

• 32 lines of code
• 11 McCabe index (conditional complexity)

    def target_throughput(self):
        def numeric(v):
            # While booleans can be converted to a number (False -> 0, True -> 1), we don't want to allow that here
            return isinstance(v, numbers.Number) and not isinstance(v, bool)

        target_throughput = self.params.get("target-throughput")
        target_interval = self.params.get("target-interval")

        if target_interval is not None and target_throughput is not None:
            raise exceptions.InvalidSyntax(
                f"Task [{self}] specifies target-interval [{target_interval}] and "
                f"target-throughput [{target_throughput}] but only one of them is allowed."
            )

        value = None
        unit = "ops/s"

        if target_interval:
            if not numeric(target_interval):
                raise exceptions.InvalidSyntax(f"Target interval [{target_interval}] for task [{self}] must be numeric.")
            value = 1 / float(target_interval)
        elif target_throughput:
            if isinstance(target_throughput, str):
                matches = re.match(Task.THROUGHPUT_PATTERN, target_throughput)
                if matches:
                    value = float(matches.group("value"))
                    unit = matches.group("unit")
                else:
                    raise exceptions.InvalidSyntax(f"Task [{self}] specifies invalid target throughput [{target_throughput}].")
            elif numeric(target_throughput):
                value = float(target_throughput)
            else:
                raise exceptions.InvalidSyntax(f"Target throughput [{target_throughput}] for task [{self}] must be string or numeric.")

        if value:
            return Throughput(value, unit)
        else:
            return None