python类HTTPSConnectionPool()的实例源码

def __https_pool(self):
        """
        Create HTTP connection pool
        :raise HttpsRequestError
        :return: urllib3.HTTPConnectionPool
        """

        try:
            pool = HTTPSConnectionPool(
                    host=self.__cfg.host,
                    port=self.__cfg.port,
                    maxsize=self.__cfg.threads,
                    timeout=Timeout(self.__cfg.timeout, read=self.__cfg.timeout),
                    block=True)
            if self._HTTP_DBG_LEVEL <= self.__debug.level:
                self.__debug.debug_connection_pool('https_pool_start', pool)

            return pool
        except Exception as error:
            raise HttpsRequestError(str(error))

def get_etcd_connection(self):
        with self._etcd_url_lock:
            port = self._etcd_url_parts.port or 2379
            if self._etcd_url_parts.scheme == "https":
                _log.debug("Getting new HTTPS connection to %s:%s",
                           self._etcd_url_parts.hostname, port)
                pool = HTTPSConnectionPool(self._etcd_url_parts.hostname,
                                           port,
                                           key_file=self._etcd_key_file,
                                           cert_file=self._etcd_cert_file,
                                           ca_certs=self._etcd_ca_file,
                                           maxsize=1)
            else:
                _log.debug("Getting new HTTP connection to %s:%s",
                           self._etcd_url_parts.hostname, port)
                pool = HTTPConnectionPool(self._etcd_url_parts.hostname,
                                          port,
                                          maxsize=1)
            return pool

def __init__(self, host='localhost', port=9200, http_auth=None,
            use_ssl=False, verify_certs=False, ca_certs=None, client_cert=None,
            ssl_version=None, ssl_assert_hostname=None, ssl_assert_fingerprint=None,
            maxsize=10, **kwargs):

        super(Urllib3HttpConnection, self).__init__(host=host, port=port, **kwargs)
        self.headers = urllib3.make_headers(keep_alive=True)
        if http_auth is not None:
            if isinstance(http_auth, (tuple, list)):
                http_auth = ':'.join(http_auth)
            self.headers.update(urllib3.make_headers(basic_auth=http_auth))

        pool_class = urllib3.HTTPConnectionPool
        kw = {}
        if use_ssl:
            pool_class = urllib3.HTTPSConnectionPool
            kw.update({
                'ssl_version': ssl_version,
                'assert_hostname': ssl_assert_hostname,
                'assert_fingerprint': ssl_assert_fingerprint,
            })

            if verify_certs:
                kw.update({
                    'cert_reqs': 'CERT_REQUIRED',
                    'ca_certs': ca_certs,
                    'cert_file': client_cert,
                })
            elif ca_certs:
                raise ImproperlyConfigured("You cannot pass CA certificates when verify SSL is off.")
            else:
                warnings.warn(
                    'Connecting to %s using SSL with verify_certs=False is insecure.' % host)

        self.pool = pool_class(host, port=port, timeout=self.timeout, maxsize=maxsize, **kw)

def _get_connection_pool():
    if DEBUG:
        return urllib3.HTTPConnectionPool('127.0.0.1:8000')
    else:
        return urllib3.HTTPSConnectionPool(
            'api.nivad.io',
            cert_reqs='CERT_REQUIRED',
            ca_certs=certifi.where()
        )

def _get_connection_pool():
    if DEBUG:
        return urllib3.HTTPConnectionPool('127.0.0.1:8000')
    else:
        return urllib3.HTTPSConnectionPool(
            'api.nivad.io',
            cert_reqs='CERT_REQUIRED',
            ca_certs=certifi.where()
        )

def get(self, host, params=()):
        # type: (object, object) -> object
        """Get metadata by url"""

        self.__is_server_online(host)
        self.__disable_verbose()
        self.__parse_params(params)
        scheme, host = urlparse(host).scheme, urlparse(host).netloc
        self.DEFAULT_HTTP_PROTOCOL = scheme + "://"
        self.urls = self.__get_urls(host)
        response = {}
        self.HEADER['user-agent'] = self.reader.get_random_user_agent()
        log.info("user-agent : " + self.HEADER['user-agent'])
        log.info('Thread num : ' + str(self.threads))

        try:
            httplib.HTTPConnection.debuglevel = self.debug

            if hasattr(urllib3, 'disable_warnings'):
                urllib3.disable_warnings()
            if scheme == "http":
                self.http = urllib3.HTTPConnectionPool(host.split(':')[0],
                                                       port=80 if len(host.split(':')) == 1 else int(
                                                           host.split(':')[1]), block=True, maxsize=10)
            elif scheme == "https":
                self.http = urllib3.HTTPSConnectionPool(host.split(':')[0],
                                                        port=443 if len(host.split(':')) == 1 else int(
                                                            host.split(':')[1]), block=True, maxsize=10)
            else:
                log.critical("not support http protocl, Exit now ")
                sys.exit(1);
            pool = threadpool.ThreadPool(self.threads)
            requests = threadpool.makeRequests(self.request, self.urls)
            for req in requests:
                pool.putRequest(req)
            time.sleep(1)
            pool.wait()
        except exceptions.AttributeError as e:
            log.critical(e.message)
        except KeyboardInterrupt:
            log.warning('Session canceled')
            sys.exit()

        self.counter['total'] = self.urls.__len__()
        self.counter['pools'] = pool.workers.__len__()

        response['count'] = self.counter
        response['result'] = self.result

        return response