def send()

in alibabacloud_oss_v2/transport/requests_client.py [0:0]

• 41 lines of code
• 7 McCabe index (conditional complexity)

    def send(self, request: HttpRequest, **kwargs) -> HttpResponse:
        self.open()
        error: Optional[Exception] = None
        resp: _RequestsHttpResponseImpl = None

        try:
            # api's configuration
            connect_timeout = kwargs.pop("connect_timeout", self._connect_timeout)
            read_timeout = kwargs.pop("readwrite_timeout", self._read_timeout)
            stream = kwargs.pop("stream", False)

            # tell requests not to add 'Accept-Encoding: gzip, deflate' by default
            if 'accept-encoding' not in request.headers:
                request.headers.update({'Accept-Encoding': None})

            response = self.session.request(
                request.method,
                request.url,
                headers=request.headers,
                data=request.body,
                verify=self._verify,
                timeout=(connect_timeout, read_timeout),
                allow_redirects=self._allow_redirects,
                proxies=self._proxies,
                stream=stream,
                **kwargs
            )

            resp = _RequestsHttpResponseImpl(
                request=request,
                internal_response=response,
                block_size=self._block_size
            )

            if not stream:
                _ = resp.read()
                resp.close()

        except (NewConnectionError, ConnectTimeoutError) as err:
            error = exceptions.RequestError(error=err)
        except requests.exceptions.ConnectionError as err:
            error = exceptions.RequestError(error=err)
        except requests.exceptions.ReadTimeout as err:
            error = exceptions.ResponseError(error=err)
        except requests.RequestException as err:
            error = exceptions.RequestError(error=err)

        if error:
            raise error

        return resp