Python urllib3 模块,HTTPResponse() 实例源码

我们从Python开源项目中,提取了以下9个代码示例,用于说明如何使用urllib3.HTTPResponse()

项目:drf-reverse-proxy    作者:danpoland    | 项目源码 | 文件源码
def get_urlopen_mock(body=DEFAULT_BODY_CONTENT, headers=None, status=200):
    mockHttpResponse = Mock(name='httplib.HTTPResponse')

    headers = urllib3.response.HTTPHeaderDict(headers)

    if not hasattr(body, 'read'):
        body = BytesIO(body)

    else:
        body.seek(0)

    urllib3_response = urllib3.HTTPResponse(body,
                                            headers,
                                            status,
                                            preload_content=False,
                                            original_response=mockHttpResponse)

    return MagicMock(return_value=urllib3_response)
项目:Projects    作者:it2school    | 项目源码 | 文件源码
def __init__(self, resp):
        super(RESTResponse, self).__init__()
        # arg: A urllib3.HTTPResponse object
        self.urllib3_response = resp
        self.status = resp.status
        self.version = resp.version
        self.reason = resp.reason
        self.strict = resp.strict
        self.is_closed = False
项目:Projects    作者:it2school    | 项目源码 | 文件源码
def closed(self):
        return self.is_closed

    # ---------------------------------
    # Backwards compat for HTTPResponse
    # ---------------------------------
项目:OpenDoor    作者:stanislav-web    | 项目源码 | 文件源码
def _provide_ssl_auth_required(self):
        """
        Provide ssl auth response
        :return: urllib3.HTTPResponse
        """

        response = HTTPResponse()
        response.status = self.DEFAULT_SSL_CERT_REQUIRED_STATUSES
        response.__setattr__('_body', ' ')
        return response
项目:sharead    作者:strin    | 项目源码 | 文件源码
def __init__(self, resp):
        # arg: A urllib3.HTTPResponse object
        self.urllib3_response = resp
        self.status = resp.status
        self.version = resp.version
        self.reason = resp.reason
        self.strict = resp.strict
        self.is_closed = False
项目:sharead    作者:strin    | 项目源码 | 文件源码
def closed(self):
        return self.is_closed


    # ---------------------------------
    # Backwards compat for HTTPResponse
    # ---------------------------------
项目:sharead    作者:strin    | 项目源码 | 文件源码
def __init__(self, resp):
        # arg: A urllib3.HTTPResponse object
        self.urllib3_response = resp
        self.status = resp.status
        self.version = resp.version
        self.reason = resp.reason
        self.strict = resp.strict
        self.is_closed = False
项目:sharead    作者:strin    | 项目源码 | 文件源码
def closed(self):
        return self.is_closed


    # ---------------------------------
    # Backwards compat for HTTPResponse
    # ---------------------------------
项目:OpenDoor    作者:stanislav-web    | 项目源码 | 文件源码
def request(self, url):
        """
        Client request SSL
        :param str url: request uri
        :return: urllib3.HTTPResponse
        """

        if self._HTTP_DBG_LEVEL <= self.__debug.level:
            self.__debug.debug_request(self._headers, url, self.__cfg.method)

        try:

            disable_warnings(InsecureRequestWarning)

            if self.__cfg.DEFAULT_SCAN == self.__cfg.scan:  # directories requests
                response = self.__pool.request(self.__cfg.method,
                                               helper.parse_url(url).path,
                                               headers=self._headers,
                                               retries=self.__cfg.retries,
                                               assert_same_host=False,
                                               redirect=False)
                self.cookies_middleware(is_accept=self.__cfg.accept_cookies, response=response)
            else:  # subdomains

                response = PoolManager().request(self.__cfg.method, url,
                                                 headers=self._headers,
                                                 retries=self.__cfg.retries,
                                                 assert_same_host=False,
                                                 redirect=False)
            return response

        except MaxRetryError:
            if self.__cfg.DEFAULT_SCAN == self.__cfg.scan:
                self.__tpl.warning(key='max_retry_error', url=helper.parse_url(url).path)

        except HostChangedError as error:
            self.__tpl.warning(key='host_changed_error', details=error)
            pass

        except ReadTimeoutError:
            self.__tpl.warning(key='read_timeout_error', url=url)
            pass

        except ConnectTimeoutError:
            self.__tpl.warning(key='connection_timeout_error', url=url)
            pass

        except SSLError:
            if self.__cfg.DEFAULT_SCAN != self.__cfg.scan:
                return self._provide_ssl_auth_required()