Python pycurl 模块，Curl() 实例源码

我们从Python开源项目中，提取了以下48个代码示例，用于说明如何使用pycurl.Curl()。

项目：fourthgen 作者：hyxc | 项目源码 | 文件源码

def api_login():
    global token
    url = settings.MASTER_API_URL_LOGIN
    ch = pycurl.Curl()    #????pycurl?????
    ch.setopt(ch.URL, url)     #??????url
    info = StringIO.StringIO()     
    ch.setopt(ch.WRITEFUNCTION, info.write)
    ch.setopt(ch.POST, True)
    #???https???????
    ch.setopt(ch.SSL_VERIFYPEER, 0)
    ch.setopt(ch.SSL_VERIFYHOST, 2)
    ch.setopt(ch.HTTPHEADER, ['Accept: application/x-yaml'])
    ch.setopt(ch.POSTFIELDS, 'username=%s&password=%s&eauth=pam' %(settings.SALT_API_AUTH_USER, settings.SALT_API_AUTH_PASS))
    #?????
    #ch.setopt(ch.HEADER, True)
    #??????
    ch.setopt(ch.HEADER,False)
    ch.perform()
    html = info.getvalue()
    #??token
    token = html.split("\n")[-3].replace("\n", '')
    token = token.split(' ')[3]
    info.close()
    ch.close()

项目：fourthgen 作者：hyxc | 项目源码 | 文件源码

def api_key(fun='key.list_all', match='', arg_num=0):
    api_login()
    url = settings.MASTER_API_URL
    ch = pycurl.Curl()
    ch.setopt(ch.URL, url)
    info = StringIO.StringIO()
    ch.setopt(ch.WRITEFUNCTION, info.write)
    ch.setopt(ch.POST, True)
    ch.setopt(ch.SSL_VERIFYPEER, 0)
    ch.setopt(ch.SSL_VERIFYHOST, 2)
    ch.setopt(ch.HTTPHEADER, ['Accept: application/json', "X-Auth-Token: %s" %(token)])
    if arg_num == 0:
        ch.setopt(ch.POSTFIELDS, "client=wheel&fun=%s" %(fun))
    elif arg_num == 1:
        ch.setopt(ch.POSTFIELDS, "client=wheel&fun=%s&match=%s" %(fun, match))
    else:
        pass
    ch.setopt(ch.HEADER,False)
    ch.perform()
    html = info.getvalue()
    info.close()
    ch.close()
    return json.loads(html)

项目：recipebook 作者：dpapathanasiou | 项目源码 | 文件源码

def put (url, data, headers={}):
    """Make a PUT request to the url, using data in the message body,
    with the additional headers, if any"""

    reply = -1 # default, non-http response

    curl = pycurl.Curl()
    curl.setopt(pycurl.URL, url)
    if len(headers) > 0:
        curl.setopt(pycurl.HTTPHEADER, [k+': '+v for k,v in headers.items()])
    curl.setopt(pycurl.PUT, 1)
    curl.setopt(pycurl.INFILESIZE, len(data))
    databuffer = StringIO(data)
    curl.setopt(pycurl.READFUNCTION, databuffer.read)
    try:
        curl.perform()
        reply = curl.getinfo(pycurl.HTTP_CODE)
    except Exception:
        pass
    curl.close()

    return reply

项目：Instagram-API 作者：danleyb2 | 项目源码 | 文件源码

def request(self, endpoint, post=None):
        buffer = BytesIO()

        ch = pycurl.Curl()
        ch.setopt(pycurl.URL, Constants.API_URL + endpoint)
        ch.setopt(pycurl.USERAGENT, self.userAgent)
        ch.setopt(pycurl.WRITEFUNCTION, buffer.write)
        ch.setopt(pycurl.FOLLOWLOCATION, True)
        ch.setopt(pycurl.HEADER, True)
        ch.setopt(pycurl.VERBOSE, False)
        ch.setopt(pycurl.COOKIEFILE, os.path.join(self.IGDataPath, self.username, self.username + "-cookies.dat"))
        ch.setopt(pycurl.COOKIEJAR, os.path.join(self.IGDataPath, self.username, self.username + "-cookies.dat"))

        if post is not None:
            ch.setopt(pycurl.POST, True)
            ch.setopt(pycurl.POSTFIELDS, post)

        if self.proxy:
            ch.setopt(pycurl.PROXY, self.proxyHost)
            if self.proxyAuth:
                ch.setopt(pycurl.PROXYUSERPWD, self.proxyAuth)

        ch.perform()
        resp = buffer.getvalue()
        header_len = ch.getinfo(pycurl.HEADER_SIZE)
        header = resp[0: header_len]
        body = resp[header_len:]

        ch.close()

        if self.debug:
            print("REQUEST: " + endpoint)
            if post is not None:
                if not isinstance(post, list):
                    print("DATA: " + str(post))
            print("RESPONSE: " + body)

        return [header, json_decode(body)]

项目：QQBot 作者：springhack | 项目源码 | 文件源码

def CurlPOST(url, data, cookie):
    c = pycurl.Curl()
    b = StringIO.StringIO()
    c.setopt(pycurl.URL, url)
    c.setopt(pycurl.POST, 1)
    c.setopt(pycurl.HTTPHEADER,['Content-Type: application/json'])
    # c.setopt(pycurl.TIMEOUT, 10)
    c.setopt(pycurl.WRITEFUNCTION, b.write)
    c.setopt(pycurl.COOKIEFILE, cookie)
    c.setopt(pycurl.COOKIEJAR, cookie)
    c.setopt(pycurl.POSTFIELDS, data)
    c.perform()
    html = b.getvalue()
    b.close()
    c.close()
    return html

项目：searx-stats2 作者：dalf | 项目源码 | 文件源码

def get_connection():
    # pycurl initialization
    h = pycurl.Curl()

    # follow redirects
    h.setopt(pycurl.FOLLOWLOCATION, False)

    # enable compression
    h.setopt(pycurl.ENCODING, 'gzip, deflate')

    # certifi
    h.setopt(pycurl.CAINFO, certifi.where())

    # no signal
    h.setopt(pycurl.NOSIGNAL, 1)

    # certificate informations
    h.setopt(pycurl.OPT_CERTINFO, 1)

    return h

项目：pathspider 作者：mami-project | 项目源码 | 文件源码

def ipv6_asn(ifname):
    try:
        c = pycurl.Curl()
        body = BytesIO()
        c.setopt(c.URL, "https://stat.ripe.net/data/prefix-overview/data.json?resource={}"
                 .format(ipv6_address_public(ifname)))
        c.setopt(c.INTERFACE, ifname)
        c.setopt(c.WRITEDATA, body)
        c.perform()
        asns = json.loads((body.getvalue()).decode('utf-8'))['data']['asns']
        if len(asns) == 1:
            return asns[0]['asn']
        else:
            return None
    except pycurl.error:
        return None

项目：xmusic-crawler 作者：rockers7414 | 项目源码 | 文件源码

def __authorize(self):
        headers = ["Authorization: Basic " + self.user_info_base64]
        post_field = urlencode(self.parameter)

        buf = BytesIO()

        client = pycurl.Curl()
        client.setopt(client.URL, self.spotify_authorize_url)
        client.setopt(client.HTTPHEADER, headers)
        client.setopt(client.POSTFIELDS, post_field)
        client.setopt(client.WRITEFUNCTION, buf.write)
        client.setopt(pycurl.SSL_VERIFYPEER, 0)
        client.perform()
        client.close()

        self.response = json.loads(buf.getvalue().decode())
        buf.close()
        self.access_token = self.response.get("access_token")

项目：defcon-workshop 作者：devsecops | 项目源码 | 文件源码

def perform(self):
        self.__performHead=""
        self.__performBody=""
        self.__headersSent=""

        try:
            conn = Request.to_pycurl_object(pycurl.Curl(), self)
            conn.perform()
            self.response_from_conn_object(conn, self.__performHead, self.__performBody)
        except pycurl.error, error:
            errno, errstr = error
            raise ReqRespException(ReqRespException.FATAL, errstr)
        finally:
            conn.close()

    ######### ESTE conjunto de funciones no es necesario para el uso habitual de la clase

项目：landscape-client 作者：CanonicalLtd | 项目源码 | 文件源码

def _curl(self, payload, computer_id, exchange_token, message_api):
        # There are a few "if _PY3" checks below, because for Python 3 we
        # want to convert a number of values from bytes to string, before
        # assigning them to the headers.
        if _PY3 and isinstance(message_api, bytes):
            message_api = message_api.decode("ascii")
        headers = {"X-Message-API": message_api,
                   "User-Agent": "landscape-client/%s" % VERSION,
                   "Content-Type": "application/octet-stream"}
        if computer_id:
            if _PY3 and isinstance(computer_id, bytes):
                computer_id = computer_id.decode("ascii")
            headers["X-Computer-ID"] = computer_id
        if exchange_token:
            if _PY3 and isinstance(exchange_token, bytes):
                exchange_token = exchange_token.decode("ascii")
            headers["X-Exchange-Token"] = str(exchange_token)
        curl = pycurl.Curl()
        return (curl, fetch(self._url, post=True, data=payload,
                            headers=headers, cainfo=self._pubkey, curl=curl))