def get_config()

in elasticapm/transport/http.py [0:0]

• 33 lines of code
• 8 McCabe index (conditional complexity)

    def get_config(self, current_version=None, keys=None):
        """
        Gets configuration from a remote APM Server

        :param current_version: version of the current configuration
        :param keys: a JSON-serializable dict to identify this instance, e.g.
                {
                    "service": {
                        "name": "foo",
                        "environment": "bar"
                    }
                }
        :return: a three-tuple of new version, config dictionary and validity in seconds.
            Any element of the tuple can be None.
        """
        url = self._config_url
        data = json_encoder.dumps(keys).encode("utf-8")
        headers = self._headers.copy()
        headers[b"Content-Type"] = "application/json"
        headers.update(self.auth_headers)
        max_age = 300
        if current_version:
            headers["If-None-Match"] = current_version
        try:
            response = self.http.urlopen(
                "POST", url, body=data, headers=headers, timeout=self._timeout, preload_content=False
            )
        except (urllib3.exceptions.RequestError, urllib3.exceptions.HTTPError) as e:
            logger.debug("HTTP error while fetching remote config: %s", str(e))
            return current_version, None, max_age
        body = response.read()

        max_age = self._get_cache_control_max_age(response.headers) or max_age

        if response.status == 304:
            # config is unchanged, return
            logger.debug("Configuration unchanged")
            return current_version, None, max_age
        elif response.status >= 400:
            return None, None, max_age

        if not body:
            logger.debug("APM Server answered with empty body and status code %s", response.status)
            return current_version, None, max_age
        body = body.decode("utf-8")
        try:
            data = json_encoder.loads(body)
            return response.headers.get("Etag"), data, max_age
        except json.JSONDecodeError:
            logger.warning("Failed decoding APM Server response as JSON: %s", body)
            return current_version, None, max_age