Python urllib3 模块，make_headers() 实例源码

def configure_http_pool():

    global gl_http_pool

    if gl_args.mode == 'auto-scan' or gl_args.mode == 'file-scan':
        timeout = Timeout(connect=1.0, read=3.0)
    else:
        timeout = Timeout(connect=gl_args.timeout, read=6.0)

    if gl_args.proxy:
        # when using proxy, protocol should be informed
        if (gl_args.host is not None and 'http' not in gl_args.host) or 'http' not in gl_args.proxy:
            print_and_flush(RED + " * When using proxy, you must specify the http or https protocol"
                                  " (eg. http://%s).\n\n" %(gl_args.host if 'http' not in gl_args.host else gl_args.proxy) +ENDC)
            logging.critical('Protocol not specified')
            exit(1)

        try:
            if gl_args.proxy_cred:
                headers = make_headers(proxy_basic_auth=gl_args.proxy_cred)
                gl_http_pool = ProxyManager(proxy_url=gl_args.proxy, proxy_headers=headers, timeout=timeout, cert_reqs='CERT_NONE')
            else:
                gl_http_pool = ProxyManager(proxy_url=gl_args.proxy, timeout=timeout, cert_reqs='CERT_NONE')
        except:
            print_and_flush(RED + " * An error occurred while setting the proxy. Please see log for details..\n\n" +ENDC)
            logging.critical('Error while setting the proxy', exc_info=traceback)
            exit(1)
    else:
        gl_http_pool = PoolManager(timeout=timeout, cert_reqs='CERT_NONE')

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_headers(self):
        if self.username and self.password:
            credentials = ':'.join((self.username, self.password))
            return urllib3.make_headers(basic_auth=credentials)
        return {}

def __init__(self,
                 con_pool_size=1,
                 proxy_url=None,
                 urllib3_proxy_kwargs=None,
                 connect_timeout=5.,
                 read_timeout=5.):
        if urllib3_proxy_kwargs is None:
            urllib3_proxy_kwargs = dict()

        self._connect_timeout = connect_timeout

        kwargs = dict(
            maxsize=con_pool_size,
            cert_reqs='CERT_REQUIRED',
            ca_certs=certifi.where(),
            socket_options=HTTPConnection.default_socket_options + [
                (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
            ],
            timeout=urllib3.Timeout(
                connect=self._connect_timeout, read=read_timeout, total=None))

        # Set a proxy according to the following order:
        # * proxy defined in proxy_url (+ urllib3_proxy_kwargs)
        # * proxy set in `HTTPS_PROXY` env. var.
        # * proxy set in `https_proxy` env. var.
        # * None (if no proxy is configured)

        if not proxy_url:
            proxy_url = os.environ.get('HTTPS_PROXY') or os.environ.get('https_proxy')

        if not proxy_url:
            if urllib3.contrib.appengine.is_appengine_sandbox():
                # Use URLFetch service if running in App Engine
                mgr = urllib3.contrib.appengine.AppEngineManager()
            else:
                mgr = urllib3.PoolManager(**kwargs)
        else:
            kwargs.update(urllib3_proxy_kwargs)
            if proxy_url.startswith('socks'):
                try:
                    from urllib3.contrib.socks import SOCKSProxyManager
                except ImportError:
                    raise RuntimeError('PySocks is missing')
                mgr = SOCKSProxyManager(proxy_url, **kwargs)
            else:
                mgr = urllib3.proxy_from_url(proxy_url, **kwargs)
                if mgr.proxy.auth:
                    # TODO: what about other auth types?
                    auth_hdrs = urllib3.make_headers(proxy_basic_auth=mgr.proxy.auth)
                    mgr.proxy_headers.update(auth_hdrs)

        self._con_pool = mgr

def clear(self):
        '''
         couchbase-cli
           To clear all data in bucket. use command couchbase-cli on Couchbase Server, 
           /opt/couchbase/bin/couchbase-cli bucket-flush -u admin -p password -c 127.0.0.1:8091 -b bucket --force

         Couchbase Command Line tools 
           folder  /opt/couchbase/lib/python could be copied to other host
           or get from GitHub
             > git clone https://github.com/couchbase/couchbase-cli.git 
           execute:
             > python couchbase-cli bucket-flush -u admin -p password -c 192.168.12.13:8091 -b bucket --force

        import os,os.path
        #if self._couchbase_cli == '':
        if True:
            log.error( 'CouchbaseError: please using couchbase-cli instead\n'
                    'couchbase-cli on GitHub\n'
                    'https://github.com/couchbase/couchbase-cli.git' )
            return False
        else:
            # Always 'Permission denied' ?
            return os.system( 'python %s bucket-flush -u %s -p %s -c %s -b %s --force'%
                        ( path.normpath( self._couchbase_cli + '/couchbaase-cli' ),
                        self._options.get( 'admin','' ),
                        self._options.get( 'admin-pwd', '' ),
                        self._server[0], self._bucket ) )== 0
        '''
        import urllib3
        conn = urllib3.connection_from_url(self._server[0], block=True, maxsize=100)
        endpoint = '/pools/default/buckets/%s/controller/doFlush' % self._bucket
        res = conn.urlopen(url=endpoint, method='POST', headers=urllib3.make_headers(basic_auth = self._options.get( 'admin:pwd', '' ),))

        if len(res.data) == 0:
            return True
        else:
            # '{"_":"Flush is disabled for the bucket"}'
            # 'Requested resource not found.\r\n'  REST error
            try:
                j = json.loads( res.data )
                msg = j.get( '-' )
            except Exception as e:
                msg = res.data

            log.error( "CouchbaseError: clear fail..: %s" % ( msg ) )
            return False