def get_variant_metrics()

in lambda/api/experiment_metrics.py [0:0]

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

    def get_variant_metrics(self, endpoint_name):
        """
        Return the strategy and the list of varints, with the counts defaulted to zero if not exist
        """
        table = self.dynamodb.Table(self.metrics_table)
        response = table.get_item(
            Key={
                "endpoint_name": endpoint_name,
            },
            ReturnConsumedCapacity="TOTAL",
        )
        # Return the list of invocation and success counts per variant
        if "Item" not in response:
            raise Exception(f"Endpoint {endpoint_name} not found")

        strategy = response["Item"]["strategy"]
        epsilon = float(response["Item"]["epsilon"])
        warmup = int(response["Item"]["warmup"])
        variant_names = response["Item"]["variant_names"]
        variant_metrics = response["Item"]["variant_metrics"]
        metrics = [
            {
                "endpoint_name": endpoint_name,
                "variant_name": v,
                "initial_variant_weight": float(
                    variant_metrics[v]["initial_variant_weight"]
                ),
                "invocation_count": int(variant_metrics[v].get("invocation_count", 0)),
                "conversion_count": int(variant_metrics[v].get("conversion_count", 0)),
                "reward_sum": float(variant_metrics[v].get("reward_sum", 0)),
            }
            for v in variant_names
        ]
        return strategy, epsilon, warmup, metrics