def fetch_sampling_rules()

in aws_xray_sdk/core/sampling/connector.py [0:0]

• 18 lines of code
• 3 McCabe index (conditional complexity)

    def fetch_sampling_rules(self):
        """
        Use X-Ray botocore client to get the centralized sampling rules
        from X-Ray service. The call is proxied and signed by X-Ray Daemon.
        """
        new_rules = []

        resp = self._xray_client.get_sampling_rules()
        records = resp['SamplingRuleRecords']

        for record in records:
            rule_def = record['SamplingRule']
            if self._is_rule_valid(rule_def):
                rule = SamplingRule(name=rule_def['RuleName'],
                                    priority=rule_def['Priority'],
                                    rate=rule_def['FixedRate'],
                                    reservoir_size=rule_def['ReservoirSize'],
                                    host=rule_def['Host'],
                                    service=rule_def['ServiceName'],
                                    method=rule_def['HTTPMethod'],
                                    path=rule_def['URLPath'],
                                    service_type=rule_def['ServiceType'])
                new_rules.append(rule)

        return new_rules