def get_data_from_request()

in elasticapm/contrib/django/client.py [0:0]

• 39 lines of code
• 14 McCabe index (conditional complexity)

    def get_data_from_request(self, request, event_type):
        result = {
            "env": dict(get_environ(request.META)),
            "method": request.method,
            "socket": {"remote_address": request.META.get("REMOTE_ADDR")},
            "cookies": dict(request.COOKIES),
        }
        if self.config.capture_headers:
            request_headers = dict(get_headers(request.META))

            for key, value in request_headers.items():
                if isinstance(value, (int, float)):
                    request_headers[key] = str(value)

            result["headers"] = request_headers

        if request.method in constants.HTTP_WITH_BODY:
            capture_body = self.config.capture_body in ("all", event_type)
            if not capture_body:
                result["body"] = "[REDACTED]"
            else:
                content_type = request.META.get("CONTENT_TYPE")
                if content_type == "application/x-www-form-urlencoded":
                    data = compat.multidict_to_dict(request.POST)
                elif content_type and content_type.startswith("multipart/form-data"):
                    data = compat.multidict_to_dict(request.POST)
                    if request.FILES:
                        data["_files"] = {field: file.name for field, file in request.FILES.items()}
                else:
                    try:
                        data = request.body
                    except Exception as e:
                        self.logger.debug("Can't capture request body: %s", str(e))
                        data = "<unavailable>"
                if data is not None:
                    # Can we apply this as a processor instead?
                    # https://github.com/elastic/apm-agent-python/issues/305
                    result["body"] = long_field(data)

        url = get_raw_uri(request)
        try:
            result["url"] = get_url_dict(url)
        except ValueError as exc:
            self.logger.warning(f"URL parsing failed: {exc}")
        return result