def main():
if len(sys.argv) != 3:
print("Usage:\n\n\tpython preprocess_spec.py kuberneres_branch " \
"output_spec_path")
return 1
spec_url = 'https://raw.githubusercontent.com/kubernetes/kubernetes/' \
'%s/api/openapi-spec/swagger.json' % sys.argv[1]
output_path = sys.argv[2]
pool = urllib3.PoolManager()
with pool.request('GET', spec_url, preload_content=False) as response:
if response.status != 200:
print("Error downloading spec file. Reason: %s" % response.reason)
return 1
in_spec = json.load(response, object_pairs_hook=OrderedDict)
out_spec = process_swagger(in_spec)
with open(output_path, 'w') as out:
json.dump(out_spec, out, sort_keys=False, indent=2,
separators=(',', ': '), ensure_ascii=True)
return 0
python类PoolManager()的实例源码
def openshift_main():
import sys
import json
import codecs
import urllib3
from collections import OrderedDict
pool = urllib3.PoolManager()
reader = codecs.getreader('utf-8')
spec_url = 'https://raw.githubusercontent.com/openshift/origin/' \
'%s/api/swagger-spec/openshift-openapi-spec.json' % sys.argv[1]
output_path = sys.argv[2]
print("writing to {}".format(output_path))
with pool.request('GET', spec_url, preload_content=False) as response:
if response.status != 200:
print("Error downloading spec file. Reason: %s" % response.reason)
return 1
in_spec = json.load(reader(response), object_pairs_hook=OrderedDict)
out_spec = process_swagger(process_openshift_swagger(in_spec, output_path))
update_codegen_ignore(out_spec, output_path)
with open(output_path, 'w') as out:
json.dump(out_spec, out, sort_keys=True, indent=2,
separators=(',', ': '), ensure_ascii=True)
return 0
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def mercatox():
http = urllib3.PoolManager(cert_reqs='CERT_REQUIRED',ca_certs=certifi.where())
url = 'https://mercatox.com/public/json24'
#response = http.request('GET', url, headers=header, timeout=20.0)
response = http.request('GET', url, timeout=20.0)
json_mercatox = json.loads(response.data)
json_array = json_mercatox['pairs']['XRB_BTC']
try:
last_price = int(float(json_array['last']) * (10 ** 8))
except KeyError:
last_price = 0
high_price = int(float(json_array['high24hr']) * (10 ** 8))
low_price = int(float(json_array['low24hr']) * (10 ** 8))
ask_price = int(float(json_array['lowestAsk']) * (10 ** 8))
bid_price = int(float(json_array['highestBid']) * (10 ** 8))
volume = int(float(json_array['baseVolume']))
btc_volume = int(float(json_array['quoteVolume']) * (10 ** 8))
mysql_set_price(1, last_price, high_price, low_price, ask_price, bid_price, volume, btc_volume)
def monitoring_block_count():
# set bot
bot = Bot(api_key)
count = int(rpc({"action": "block_count"}, 'count'))
reference_count = int(reference_block_count())
http = urllib3.PoolManager(cert_reqs='CERT_REQUIRED',ca_certs=certifi.where())
response = http.request('GET', summary_url, headers=header, timeout=20.0)
try:
json_data = json.loads(response.data)
community_count = int(json_data['blocks'])
except ValueError as e:
community_count = reference_count
difference = int(math.fabs(community_count - count))
response = http.request('GET', block_count_url, headers=header, timeout=20.0)
raiwallet_count = int(response.data)
if (difference > block_count_difference_threshold):
# Warning to admins
for user_id in admin_list:
push(bot, user_id, 'Block count: {0}\nCommunity: {1}\nDifference: *{2}*\nReference: {3}\nraiwallet.info: {4}'.format(count, community_count, difference, reference_count, raiwallet_count))
# trying to fix
bootstrap_multi()
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __connect(self):
num_pools = float(self.pool_size_total) / self.pool_size_per_route
headers = {
'Content-Type': 'application/x-protobuf',
'Accept': 'application/x-protobuf',
'User-Agent': 'python-pilosa/' + VERSION,
}
timeout = urllib3.Timeout(connect=self.connect_timeout, read=self.socket_timeout)
client_options = {
"num_pools": num_pools,
"maxsize": self.pool_size_per_route,
"block": True,
"headers": headers,
"timeout": timeout,
"retries": self.retry_count,
}
if not self.tls_skip_verify:
client_options["cert_reqs"] = "CERT_REQUIRED"
client_options["ca_certs"] = self.tls_ca_certificate_path
client = urllib3.PoolManager(**client_options)
self.__client = client
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, lat, lon):
"""
Populates and parses location data in Location object
"""
url = REQUEST_URL + "?lat=" + lat + "&lng=" + lon + "&date=" + DATE
http = urllib3.PoolManager()
r = http.request(REQUEST_TYPE, url)
self.__RAW_SUN_DATA__ = sun_json_parse(r.data)
self.__SUN_DATA__ = self.__RAW_SUN_DATA__.split(DATA_SPLIT)
self.SUNRISE = self.__SUN_DATA__[0]
self.SUNSET = self.__SUN_DATA__[1]
self.SOLAR_NOON = self.__SUN_DATA__[2]
self.DAY_LENGTH = self.__SUN_DATA__[3]
self.CIVIL_TWILIGHT_BEGIN = self.__SUN_DATA__[4]
self.CIVIL_TWILIGHT_END = self.__SUN_DATA__[5]
self.NAUTICAL_TWILIGHT_BEGIN = self.__SUN_DATA__[6]
self.NAUTICAL_TWILIGHT_END = self.__SUN_DATA__[7]
self.ASTROLOGICAL_TWILIGHT_BEGIN = self.__SUN_DATA__[8]
self.ASTROLOGICAL_TWILIGHT_END = self.__SUN_DATA__[9]
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def urllib3_test():
retry = urllib3.util.Retry(total=MAX_RETRIES, connect=MAX_RETRIES, read=MAX_RETRIES, backoff_factor=BACKOFF_FACTOR)
http = urllib3.PoolManager(retries=retry, timeout=TIMEOUT)
def attempt(url, http=http, retry=retry):
r = None
try:
r = http.request('GET', url, retries=retry)
except Exception as e:
retry = r.retries if r else retry
retry = retry.increment('GET', url, error=e)
retry.sleep()
logging.warning("Retrying (%r) after connection broken by '%r': '%s'", retry, e, url)
return attempt(url, retry=retry)
return r
return json.loads(attempt(URL, http).data.decode('utf-8'))
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
def __init__(self, headers=None, retries=None, validate_certificate=True,
urlfetch_retries=True):
if not urlfetch:
raise AppEnginePlatformError(
"URLFetch is not available in this environment.")
if is_prod_appengine_mvms():
raise AppEnginePlatformError(
"Use normal urllib3.PoolManager instead of AppEngineManager"
"on Managed VMs, as using URLFetch is not necessary in "
"this environment.")
warnings.warn(
"urllib3 is using URLFetch on Google App Engine sandbox instead "
"of sockets. To use sockets directly instead of URLFetch see "
"https://urllib3.readthedocs.io/en/latest/reference/urllib3.contrib.html.",
AppEnginePlatformWarning)
RequestMethods.__init__(self, headers)
self.validate_certificate = validate_certificate
self.urlfetch_retries = urlfetch_retries
self.retries = retries or Retry.DEFAULT
