我们从Python开源项目中，提取了以下13个代码示例，用于说明如何使用pycurl.READFUNCTION。
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
def test_post_data(self): curl = CurlStub(b"result") result = fetch("http://example.com", post=True, data="data", curl=curl) self.assertEqual(result, b"result") self.assertEqual(curl.options[pycurl.READFUNCTION](), b"data") self.assertEqual(curl.options, {pycurl.URL: b"http://example.com", pycurl.FOLLOWLOCATION: 1, pycurl.MAXREDIRS: 5, pycurl.CONNECTTIMEOUT: 30, pycurl.LOW_SPEED_LIMIT: 1, pycurl.LOW_SPEED_TIME: 600, pycurl.NOSIGNAL: 1, pycurl.WRITEFUNCTION: Any(), pycurl.POST: True, pycurl.POSTFIELDSIZE: 4, pycurl.READFUNCTION: Any(), pycurl.DNS_CACHE_TIMEOUT: 0, pycurl.ENCODING: b"gzip,deflate"})
def __init__(self, uploader): self.handle = pycurl.Curl() self.response_headers = {} self.output = six.StringIO() self.status_code = None self.handle.setopt(pycurl.CAINFO, certifi.where()) self.handle.setopt(pycurl.URL, uploader.url) self.handle.setopt(pycurl.HEADERFUNCTION, self._prepare_response_header) self.handle.setopt(pycurl.UPLOAD, 1) self.handle.setopt(pycurl.CUSTOMREQUEST, 'PATCH') self.file = uploader.get_file_stream() self.file.seek(uploader.offset) self.handle.setopt(pycurl.READFUNCTION, self.file.read) self.handle.setopt(pycurl.WRITEFUNCTION, self.output.write) self.handle.setopt(pycurl.INFILESIZE, uploader.request_length) headers = ["upload-offset: {}".format(uploader.offset), "Content-Type: application/offset+octet-stream"] + uploader.headers_as_list self.handle.setopt(pycurl.HTTPHEADER, headers)
def transfer(ipaddr, username, password, commandfile): #transfers commandfile to camera storage = StringIO() c = pycurl.Curl() c.setopt(c.URL, 'http://' + ipaddr + '/admin/remoteconfig') c.setopt(c.POST, 1) c.setopt(c.CONNECTTIMEOUT, 5) c.setopt(c.TIMEOUT, TIMEOUT) filesize = os.path.getsize(commandfile) f = open(commandfile, 'rb') c.setopt(c.FAILONERROR, True) c.setopt(pycurl.POSTFIELDSIZE, filesize) c.setopt(pycurl.READFUNCTION, FileReader(f).read_callback) c.setopt(c.WRITEFUNCTION, storage.write) c.setopt(pycurl.HTTPHEADER, ["application/x-www-form-urlencoded"]) c.setopt(c.VERBOSE, VERBOSE) c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC) c.setopt(pycurl.USERPWD, username + ':' + password) try: c.perform() except pycurl.error, error: errno, errstr = error print 'An error occurred: ', errstr return False, '' c.close() content = storage.getvalue() f.close() return True, content # *************************************************************** # *** Main program *** # ***************************************************************
def request(self, url, method, body, headers): c = pycurl.Curl() c.setopt(pycurl.URL, url) if 'proxy_host' in self.proxy: c.setopt(pycurl.PROXY, self.proxy['proxy_host']) if 'proxy_port' in self.proxy: c.setopt(pycurl.PROXYPORT, self.proxy['proxy_port']) if 'proxy_user' in self.proxy: c.setopt(pycurl.PROXYUSERPWD, "%(proxy_user)s:%(proxy_pass)s" % self.proxy) self.buf = StringIO() c.setopt(pycurl.WRITEFUNCTION, self.buf.write) #c.setopt(pycurl.READFUNCTION, self.read) #self.body = StringIO(body) #c.setopt(pycurl.HEADERFUNCTION, self.header) if self.cacert: c.setopt(c.CAINFO, self.cacert) c.setopt(pycurl.SSL_VERIFYPEER, self.cacert and 1 or 0) c.setopt(pycurl.SSL_VERIFYHOST, self.cacert and 2 or 0) c.setopt(pycurl.CONNECTTIMEOUT, self.timeout) c.setopt(pycurl.TIMEOUT, self.timeout) if method == 'POST': c.setopt(pycurl.POST, 1) c.setopt(pycurl.POSTFIELDS, body) if headers: hdrs = ['%s: %s' % (k, v) for k, v in headers.items()] log.debug(hdrs) c.setopt(pycurl.HTTPHEADER, hdrs) c.perform() c.close() return {}, self.buf.getvalue()
def request(self, url, method, body, headers): c = pycurl.Curl() c.setopt(pycurl.URL, url) if 'proxy_host' in self.proxy: c.setopt(pycurl.PROXY, self.proxy['proxy_host']) if 'proxy_port' in self.proxy: c.setopt(pycurl.PROXYPORT, self.proxy['proxy_port']) if 'proxy_user' in self.proxy: c.setopt(pycurl.PROXYUSERPWD, "%(proxy_user)s:%(proxy_pass)s" % self.proxy) self.buf = StringIO() c.setopt(pycurl.WRITEFUNCTION, self.buf.write) #c.setopt(pycurl.READFUNCTION, self.read) #self.body = StringIO(body) #c.setopt(pycurl.HEADERFUNCTION, self.header) if self.cacert: c.setopt(c.CAINFO, self.cacert) c.setopt(pycurl.SSL_VERIFYPEER, self.cacert and 1 or 0) c.setopt(pycurl.SSL_VERIFYHOST, self.cacert and 2 or 0) c.setopt(pycurl.CONNECTTIMEOUT, self.timeout / 6) c.setopt(pycurl.TIMEOUT, self.timeout) if method == 'POST': c.setopt(pycurl.POST, 1) c.setopt(pycurl.POSTFIELDS, body) if headers: hdrs = ['%s: %s' % (k, v) for k, v in headers.items()] log.debug(hdrs) c.setopt(pycurl.HTTPHEADER, hdrs) c.perform() c.close() return {}, self.buf.getvalue()
def transfer(ipaddr, username, password, commandfile, trackprogress): #transfers commandfile to camera storage = StringIO() c = pycurl.Curl() c.setopt(c.URL, 'http://' + ipaddr + '/admin/remoteconfig') c.setopt(c.POST, 1) c.setopt(c.CONNECTTIMEOUT, 5) c.setopt(c.TIMEOUT, TIMEOUT) filesize = os.path.getsize(commandfile) f = open(commandfile, 'rb') c.setopt(c.FAILONERROR, True) c.setopt(pycurl.POSTFIELDSIZE, filesize) c.setopt(pycurl.READFUNCTION, FileReader(f).read_callback) if trackprogress: c.setopt(pycurl.NOPROGRESS, 0) c.setopt(pycurl.PROGRESSFUNCTION, progresscallback) starttime = time.time() else: c.setopt(pycurl.NOPROGRESS, 1) c.setopt(c.WRITEFUNCTION, storage.write) c.setopt(pycurl.HTTPHEADER, ["application/x-www-form-urlencoded"]) c.setopt(c.VERBOSE, VERBOSE) c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC) c.setopt(pycurl.USERPWD, username + ':' + password) try: c.perform() except pycurl.error, error: errno, errstr = error print 'An error occurred: ', errstr return False, '' c.close() content = storage.getvalue() f.close() return True, content
def transfer(ipaddr, username, password, commandfile): #transfers commandfile to camera storage = StringIO() c = pycurl.Curl() c.setopt(c.URL, 'http://' + ipaddr + '/admin/remoteconfig') c.setopt(c.POST, 1) c.setopt(c.CONNECTTIMEOUT, 5) c.setopt(c.TIMEOUT, 60) filesize = os.path.getsize(commandfile) f = open(commandfile, 'rb') c.setopt(c.FAILONERROR, True) c.setopt(pycurl.POSTFIELDSIZE, filesize) c.setopt(pycurl.READFUNCTION, FileReader(f).read_callback) c.setopt(c.WRITEFUNCTION, storage.write) c.setopt(pycurl.HTTPHEADER, ["application/x-www-form-urlencoded"]) c.setopt(c.VERBOSE, 0) c.setopt(pycurl.HTTPAUTH, pycurl.HTTPAUTH_BASIC) c.setopt(pycurl.USERPWD, username + ':' + password) try: c.perform() except pycurl.error, error: errno, errstr = error print 'An error occurred: ', errstr return False, '' c.close() content = storage.getvalue() f.close() return True, content # *************************************************************** # *** Main program *** # ***************************************************************
