def _delete_pod(self, name, namespace):
headers = {'Content-Type': 'application/json'}
con_dest_pod = httplib.HTTPConnection(self.target+":"+self.port)
con_dest_pod.request('DELETE', '/api/'+self.apiversion+'/namespaces/' +
namespace+'/pods/'+name, headers=headers, body=POD_DELETE)
resp = con_dest_pod.getresponse()
reason = resp.reason
status = resp.status
data = resp.read()
con_dest_pod.close()
if not 200 <= status <= 299:
errmsg = "Failed to delete Pod: {} {} - {}".format(
status, reason, data)
raise RuntimeError(errmsg)
for _ in xrange(5):
status, data, reason = self._get_pod(name, namespace)
if status != 404:
time.sleep(1)
continue
break
if status != 404:
errmsg = "Failed to delete Pod: {} {} - {}".format(
status, reason, data)
raise RuntimeError(errmsg)
python类HTTPConnection()的实例源码
def send_http(self, url):
o = urlparse(url)
r = 0
try:
conn = httplib.HTTPConnection(o[1], timeout=self.timeout)
if o[4]:
conn.request('GET', o[2] + o[3] + '?' + o[4], headers=self.headers)
else:
conn.request('GET', o[2] + o[3], headers=self.headers)
r = conn.getresponse()
logger.info('%s %s' % (url, r.status))
time.sleep(self.delay)
except (httplib.HTTPException, socket.timeout, socket.gaierror, Exception), e:
logger.error('url %s is unreachable, Exception %s %s' % (url, e.__class__.__name__, e))
print 'url %s is unreachable, Exception %s %s' % (url.encode('utf-8'), e.__class__.__name__, e)
pass
return r
def DoesUrlExist(url):
"""Determines whether a resource exists at the given URL.
Args:
url: URL to be verified.
Returns:
True if url exists, otherwise False.
"""
parsed = urlparse.urlparse(url)
try:
conn = httplib.HTTPConnection(parsed.netloc)
conn.request('HEAD', parsed.path)
response = conn.getresponse()
except (socket.gaierror, socket.error):
return False
finally:
conn.close()
# Follow both permanent (301) and temporary (302) redirects.
if response.status == 302 or response.status == 301:
return DoesUrlExist(response.getheader('location'))
return response.status == 200
def is_alive(self):
url = urlparse.urlparse(self.conf.monasca_url)
if url.scheme == 'https':
http_connector = httplib.HTTPSConnection
else:
http_connector = httplib.HTTPConnection
try:
connection = http_connector(host=url.netloc)
connection.request('HEAD', url=url.path)
response = connection.getresponse()
except httplib.socket.error:
return False
try:
if getattr(response, 'status') in [200, 401]:
return True
except AttributeError:
pass
return False
def DoesUrlExist(url):
"""Determines whether a resource exists at the given URL.
Args:
url: URL to be verified.
Returns:
True if url exists, otherwise False.
"""
parsed = urlparse.urlparse(url)
try:
conn = httplib.HTTPConnection(parsed.netloc)
conn.request('HEAD', parsed.path)
response = conn.getresponse()
except (socket.gaierror, socket.error):
return False
finally:
conn.close()
# Follow both permanent (301) and temporary (302) redirects.
if response.status == 302 or response.status == 301:
return DoesUrlExist(response.getheader('location'))
return response.status == 200
def _http_request(self, url, timeout=40):
try:
if not url: url = '/'
conn_fuc = httplib.HTTPSConnection if self.schema == 'https' else httplib.HTTPConnection
conn = conn_fuc(self.host, timeout=timeout)
conn.request(method='GET', url=url,
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/38.0.2125.111 Safari/537.36 BBScan/1.0'}
)
resp = conn.getresponse()
resp_headers = dict(resp.getheaders())
status = resp.status
if resp_headers.get('content-type', '').find('text') >= 0 or resp_headers.get('content-type', '').find('html') >= 0 or \
int(resp_headers.get('content-length', '0')) <= 1048576:
html_doc = self._decode_response_text(resp.read())
else:
html_doc = ''
conn.close()
return status, resp_headers, html_doc
except Exception, e:
#logging.error('[Exception in InfoDisScanner._http_request] %s' % e)
return -1, {}, ''
def make_connection(self, host):
#return an existing connection if possible. This allows
#HTTP/1.1 keep-alive.
if self._connection and host == self._connection[0]:
return self._connection[1]
# create a HTTP connection object from a host descriptor
chost, self._extra_headers, x509 = self.get_host_info(host)
#store the host argument along with the connection object
self._connection = host, httplib.HTTPConnection(chost)
return self._connection[1]
##
# Clear any cached connection object.
# Used in the event of socket errors.
#
def http_open(self, req):
return self.do_open(httplib.HTTPConnection, req)
def doit():
params = urllib.urlencode({'field1': read_temp(), 'key'= 'API_KEY_NUMBER'})
headers = {"Content-type": "application/x-www-form-urlencoded","Accept":"text/plain"}
conn = httplib.HTTPConnection("api.thingspeak.com:80")
conn.request("POST", "/update", params, headers)
response = conn.getresponse()
data = response.read()
conn.close()
# Get current mode from DB
def __init__(self, host, port=None, strict=None, timeout=None, proxy_info=None):
httplib.HTTPConnection.__init__(self, host, port, strict)
self.timeout = timeout
self.proxy_info = proxy_info
def __init__(self, host, port=None, key_file=None, cert_file=None,
strict=None, timeout=None, proxy_info=None, ca_certs=None,
disable_ssl_certificate_validation=False):
httplib.HTTPConnection.__init__(self, host, port=port,
strict=strict, timeout=timeout)
def _create_connection(scheme, netloc):
if scheme == 'https':
conn = httplib.HTTPSConnection(netloc)
else:
conn = httplib.HTTPConnection(netloc)
conn.connect()
return conn
def saveFailedTest(data, expect, filename):
"""Upload failed test images to web server to allow CI test debugging.
"""
commit = runSubprocess(['git', 'rev-parse', 'HEAD'])
name = filename.split('/')
name.insert(-1, commit.strip())
filename = '/'.join(name)
host = 'data.pyqtgraph.org'
# concatenate data, expect, and diff into a single image
ds = data.shape
es = expect.shape
shape = (max(ds[0], es[0]) + 4, ds[1] + es[1] + 8 + max(ds[1], es[1]), 4)
img = np.empty(shape, dtype=np.ubyte)
img[..., :3] = 100
img[..., 3] = 255
img[2:2+ds[0], 2:2+ds[1], :ds[2]] = data
img[2:2+es[0], ds[1]+4:ds[1]+4+es[1], :es[2]] = expect
diff = makeDiffImage(data, expect)
img[2:2+diff.shape[0], -diff.shape[1]-2:-2] = diff
png = makePng(img)
conn = httplib.HTTPConnection(host)
req = urllib.urlencode({'name': filename,
'data': base64.b64encode(png)})
conn.request('POST', '/upload.py', req)
response = conn.getresponse().read()
conn.close()
print("\nImage comparison failed. Test result: %s %s Expected result: "
"%s %s" % (data.shape, data.dtype, expect.shape, expect.dtype))
print("Uploaded to: \nhttp://%s/data/%s" % (host, filename))
if not response.startswith(b'OK'):
print("WARNING: Error uploading data to %s" % host)
print(response)
def saveFailedTest(data, expect, filename):
"""Upload failed test images to web server to allow CI test debugging.
"""
commit = runSubprocess(['git', 'rev-parse', 'HEAD'])
name = filename.split('/')
name.insert(-1, commit.strip())
filename = '/'.join(name)
host = 'data.pyqtgraph.org'
# concatenate data, expect, and diff into a single image
ds = data.shape
es = expect.shape
shape = (max(ds[0], es[0]) + 4, ds[1] + es[1] + 8 + max(ds[1], es[1]), 4)
img = np.empty(shape, dtype=np.ubyte)
img[..., :3] = 100
img[..., 3] = 255
img[2:2+ds[0], 2:2+ds[1], :ds[2]] = data
img[2:2+es[0], ds[1]+4:ds[1]+4+es[1], :es[2]] = expect
diff = makeDiffImage(data, expect)
img[2:2+diff.shape[0], -diff.shape[1]-2:-2] = diff
png = makePng(img)
conn = httplib.HTTPConnection(host)
req = urllib.urlencode({'name': filename,
'data': base64.b64encode(png)})
conn.request('POST', '/upload.py', req)
response = conn.getresponse().read()
conn.close()
print("\nImage comparison failed. Test result: %s %s Expected result: "
"%s %s" % (data.shape, data.dtype, expect.shape, expect.dtype))
print("Uploaded to: \nhttp://%s/data/%s" % (host, filename))
if not response.startswith(b'OK'):
print("WARNING: Error uploading data to %s" % host)
print(response)
def _net_worker(self, method, path, data='', headers={}, metadata={}):
connection = httplib.HTTPConnection(self.thehost)
if self.content_md5 != '':
headers['Content-MD5'] = self.content_md5
self.content_md5 = ''
if self.file_secret != '':
headers['Content-Secret'] = self.file_secret
self.file_secret = ''
final_headers = merge_meta(headers, metadata)
if self.upAuth:
self._add_upyun_auth_header(final_headers,method,path)
else :
self._basicAuth(final_headers,self.username,self.password)
connection.request(method, path , data, final_headers)
resp = connection.getresponse()
if self.debug and resp.status != 200 and method != "HEAD" :
raise UpYunException(u'ERROR: Code:%d,Message:%s'%(resp.status,resp.read()))
return resp
#??????
def __init__(self, output_spec):
"""
Uses HTTP chunked transfer-encoding to stream a request body to a
server. Unfortunately requests does not support hooking into this logic
easily, so we use the lower-level httplib module.
"""
super(HttpStreamPushAdapter, self).__init__(output_spec)
self._closed = False
parts = urlparse.urlparse(output_spec['url'])
if parts.scheme == 'https':
ssl_context = ssl.create_default_context()
conn = httplib.HTTPSConnection(parts.netloc, context=ssl_context)
else:
conn = httplib.HTTPConnection(parts.netloc)
try:
conn.putrequest(output_spec.get('method', 'POST').upper(),
parts.path, skip_accept_encoding=True)
for header, value in output_spec.get('headers', {}).items():
conn.putheader(header, value)
conn.putheader('Transfer-Encoding', 'chunked')
conn.endheaders() # This actually flushes the headers to the server
except Exception:
print('HTTP connection to "%s" failed.' % output_spec['url'])
conn.close()
raise
self.conn = conn
def __init__(self, url, headers={}):
self._url = url
"""
Uses HTTP chunked transfer-encoding to stream a request body to a
server. Unfortunately requests does not support hooking into this logic
easily, so we use the lower-level httplib module.
"""
self._closed = False
parts = urlparse.urlparse(self._url)
if parts.scheme == 'https':
ssl_context = ssl.create_default_context()
conn = httplib.HTTPSConnection(parts.netloc, context=ssl_context)
else:
conn = httplib.HTTPConnection(parts.netloc)
try:
url = parts.path
if parts.query is not None:
url = '%s?%s' % (url, parts.query)
conn.putrequest('POST',
url, skip_accept_encoding=True)
for header, value in headers.items():
conn.putheader(header, value)
conn.putheader('Transfer-Encoding', 'chunked')
conn.endheaders() # This actually flushes the headers to the server
except Exception:
sys.stderr.write('HTTP connection to "%s" failed.\n' % self._url)
conn.close()
raise
self.conn = conn
def sendCommand(self, server, url, response, follow_redirects=1,
secure=0, keyfile=None, certfile=None):
data = self.Encode()
if secure:
if keyfile and certfile:
conn = httplib.HTTPSConnection(server, key_file=keyfile,
cert_file=certfile)
else:
conn = httplib.HTTPSConnection(server)
else:
conn = httplib.HTTPConnection(server)
conn.putrequest("POST", url)
conn.putheader("Content-Length", "%d" %len(data))
conn.endheaders()
conn.send(data)
resp = conn.getresponse()
if follow_redirects > 0 and resp.status == 302:
m = URL_RE.match(resp.getheader('Location'))
if m:
protocol, server, url = m.groups()
return self.sendCommand(server, url, response,
follow_redirects=follow_redirects - 1,
secure=(protocol == 'https'),
keyfile=keyfile,
certfile=certfile)
if resp.status != 200:
raise ProtocolBufferReturnError(resp.status)
if response is not None:
response.ParseFromString(resp.read())
return response
def __init__(self, host, port=None, strict=None, timeout=None, proxy_info=None):
httplib.HTTPConnection.__init__(self, host, port, strict)
self.timeout = timeout
self.proxy_info = proxy_info
