def __call_

def call()

in aws_xray_sdk/ext/django/middleware.py [0:0]
60 lines of code
14 McCabe index (conditional complexity)

    def __call__(self, request):

        sampling_decision = None
        meta = request.META
        xray_header = construct_xray_header(meta)
        # a segment name is required
        name = calculate_segment_name(meta.get(HOST_KEY), xray_recorder)

        sampling_req = {
            'host': meta.get(HOST_KEY),
            'method': request.method,
            'path': request.path,
            'service': name,
        }
        sampling_decision = calculate_sampling_decision(
            trace_header=xray_header,
            recorder=xray_recorder,
            sampling_req=sampling_req,
        )
        if self.in_lambda_ctx:
            segment = xray_recorder.begin_subsegment(name)
            # X-Ray can't search/filter subsegments on URL but it can search annotations
            # So for lambda to be able to filter by annotation we add these as annotations
        else:
            segment = xray_recorder.begin_segment(
                name=name,
                traceid=xray_header.root,
                parent_id=xray_header.parent,
                sampling=sampling_decision,
            )

        segment.save_origin_trace_header(xray_header)
        segment.put_http_meta(http.URL, request.build_absolute_uri())
        segment.put_http_meta(http.METHOD, request.method)
        if self._urls_as_annotation():
            segment.put_annotation(http.URL, request.build_absolute_uri())
            segment.put_annotation(http.METHOD, request.method)

        if meta.get(USER_AGENT_KEY):
            segment.put_http_meta(http.USER_AGENT, meta.get(USER_AGENT_KEY))
            if self._urls_as_annotation():
                segment.put_annotation(http.USER_AGENT, meta.get(USER_AGENT_KEY))
        if meta.get(X_FORWARDED_KEY):
            # X_FORWARDED_FOR may come from untrusted source so we
            # need to set the flag to true as additional information
            segment.put_http_meta(http.CLIENT_IP, meta.get(X_FORWARDED_KEY))
            segment.put_http_meta(http.X_FORWARDED_FOR, True)
            if self._urls_as_annotation():
                segment.put_annotation(http.CLIENT_IP, meta.get(X_FORWARDED_KEY))
                segment.put_annotation(http.X_FORWARDED_FOR, True)
        elif meta.get(REMOTE_ADDR_KEY):
            segment.put_http_meta(http.CLIENT_IP, meta.get(REMOTE_ADDR_KEY))
            if self._urls_as_annotation():
                segment.put_annotation(http.CLIENT_IP, meta.get(REMOTE_ADDR_KEY))

        response = self.get_response(request)
        segment.put_http_meta(http.STATUS, response.status_code)
        if self._urls_as_annotation():
            segment.put_annotation(http.STATUS, response.status_code)

        if response.has_header(CONTENT_LENGTH_KEY):
            length = int(response[CONTENT_LENGTH_KEY])
            segment.put_http_meta(http.CONTENT_LENGTH, length)
            if self._urls_as_annotation():
                segment.put_annotation(http.CONTENT_LENGTH, length)
        response[http.XRAY_HEADER] = prepare_response_header(xray_header, segment)

        if self.in_lambda_ctx:
            xray_recorder.end_subsegment()
        else:
            xray_recorder.end_segment()

        return response
def __call__()

def call()
