def urlopen()

in aliyun-python-sdk-core/aliyunsdkcore/vendored/requests/packages/urllib3/contrib/appengine.py [0:0]

• 74 lines of code
• 23 McCabe index (conditional complexity)

    def urlopen(self, method, url, body=None, headers=None,
                retries=None, redirect=True, timeout=Timeout.DEFAULT_TIMEOUT,
                **response_kw):

        retries = self._get_retries(retries, redirect)

        try:
            follow_redirects = (
                    redirect and
                    retries.redirect != 0 and
                    retries.total)
            response = urlfetch.fetch(
                url,
                payload=body,
                method=method,
                headers=headers or {},
                allow_truncated=False,
                follow_redirects=self.urlfetch_retries and follow_redirects,
                deadline=self._get_absolute_timeout(timeout),
                validate_certificate=self.validate_certificate,
            )
        except urlfetch.DeadlineExceededError as e:
            raise TimeoutError(self, e)

        except urlfetch.InvalidURLError as e:
            if 'too large' in str(e):
                raise AppEnginePlatformError(
                    "URLFetch request too large, URLFetch only "
                    "supports requests up to 10mb in size.", e)
            raise ProtocolError(e)

        except urlfetch.DownloadError as e:
            if 'Too many redirects' in str(e):
                raise MaxRetryError(self, url, reason=e)
            raise ProtocolError(e)

        except urlfetch.ResponseTooLargeError as e:
            raise AppEnginePlatformError(
                "URLFetch response too large, URLFetch only supports"
                "responses up to 32mb in size.", e)

        except urlfetch.SSLCertificateError as e:
            raise SSLError(e)

        except urlfetch.InvalidMethodError as e:
            raise AppEnginePlatformError(
                "URLFetch does not support method: %s" % method, e)

        http_response = self._urlfetch_response_to_http_response(
            response, retries=retries, **response_kw)

        # Handle redirect?
        redirect_location = redirect and http_response.get_redirect_location()
        if redirect_location:
            # Check for redirect response
            if (self.urlfetch_retries and retries.raise_on_redirect):
                raise MaxRetryError(self, url, "too many redirects")
            else:
                if http_response.status == 303:
                    method = 'GET'

                try:
                    retries = retries.increment(method, url, response=http_response, _pool=self)
                except MaxRetryError:
                    if retries.raise_on_redirect:
                        raise MaxRetryError(self, url, "too many redirects")
                    return http_response

                retries.sleep_for_retry(http_response)
                log.debug("Redirecting %s -> %s", url, redirect_location)
                redirect_url = urljoin(url, redirect_location)
                return self.urlopen(
                    method, redirect_url, body, headers,
                    retries=retries, redirect=redirect,
                    timeout=timeout, **response_kw)

        # Check if we should retry the HTTP response.
        has_retry_after = bool(http_response.getheader('Retry-After'))
        if retries.is_retry(method, http_response.status, has_retry_after):
            retries = retries.increment(
                method, url, response=http_response, _pool=self)
            log.debug("Retry: %s", url)
            retries.sleep(http_response)
            return self.urlopen(
                method, url,
                body=body, headers=headers,
                retries=retries, redirect=redirect,
                timeout=timeout, **response_kw)

        return http_response