def _request(self, method, uri, path_params=None, flatten_params=True, **kwargs):
if path_params:
# Ensure path param is encoded.
path_params = {key: quote(str(value), safe=u'') for key, value in path_params.items()}
uri %= path_params
# Custom nested object flattening
if flatten_params and 'params' in kwargs:
kwargs['params'] = self._flatten_param(kwargs['params'])
full_uri = self._endpoint + uri
response = self._session.request(method, full_uri, **kwargs)
log_message = format_request(response)
logging.info(log_message)
if not 200 <= response.status_code <= 299:
logging.error(log_message)
return response
# Delayed qualifying decorator as staticmethod. This is a workaround to error raised from using a decorator
# decorated by @staticmethod.
python类quote()的实例源码
def query(location, cty_codes, query_method, fuzzy):
results = []
try:
base_url = get_geonames_base_url()
username = get_geonames_user_name()
query_string = base_url + 'username={user}&{query_method}={name}&' \
'style=FULL&orderby={order}&startRow=0&maxRows=5&fuzzy={fuzzy}' \
.format(user=username, query_method=query_method, name=quote(location), order='relevance', fuzzy=fuzzy)
if cty_codes and len(cty_codes) > 0:
query_string = query_string + '&' + '&'.join([('country={}'.format(c)) for c in cty_codes])
json_decode = json.JSONDecoder() # used to parse json response
response = urlopen(query_string)
response_string = response.read().decode('utf-8')
parsed_response = json_decode.decode(response_string)
if parsed_response.get('geonames') and len(parsed_response.get('geonames')) > 0:
for item in parsed_response['geonames']:
results.append(parse(item))
except URLError as e:
logger.info("Oops! something didn't go well")
logger.info(e)
return results
def require_login_frontend(only_if=True):
"""
Same logic as the API require_login, but this decorator is intended for use for frontend interfaces.
It returns a redirect to the login page, along with a post-login redirect_url as a GET parameter.
:param only_if: Optionally specify a boolean condition that needs to be true for the frontend login to be required.
This is semantically equivalent to "require login for this view endpoint only if <condition>,
otherwise, no login is required"
"""
def decorator(func):
@wraps(func)
def decorated_view(*args, **kwargs):
if not current_user.is_authenticated and only_if:
return redirect(UserLoginInterfaceURI.uri(redirect_url=quote(request.url, safe='')))
return func(*args, **kwargs)
return decorated_view
return decorator
def fetch_project(base_url, api, project, auth=None):
"""
:return: project
:rtype: dict
"""
if not project:
return None
project = quote(str(project), '')
endpoint = ('/api/{api}/projects/{project}'
.format(api=api, project=project))
response = do_request('get', base_url, endpoint, auth=auth)
if response.status_code != 200:
return None
response = json.loads(response.text)
return response
def _build_url(self, vhost, queues=False):
"""Build the management api url for the requested entities"""
url = self.base_url
if queues:
url += '/queues'
if vhost:
url += '/{}'.format(quote(vhost, safe=''))
else:
url += '/vhosts'
samples = []
for metric in ['lengths', 'data_rates', 'msg_rates', 'node_stats']:
samples.append('{}_age={}'.format(metric, self.length))
samples.append('{}_incr={}'.format(metric, self.length))
url += '?{}'.format('&'.join(samples))
return url
def url_safe_version(self):
return quote_url(self.version, safe='')
def get_params_for_url(params):
return '?' + '&'.join([quote(name) + '=' + quote(str(value)) if not isinstance(value, list) else '&'.join([quote(name) + '=' + quote(str(intValue)) for intValue in value]) for name, value in sorted(params.items(), key=lambda p: p[0]) if value is not None]) if params else ''
def build_dlna_play_container(udn, server_type, path):
s = "dlna-playcontainer://" + quote(udn)
s += "?"
s += 'sid=' + quote(server_type)
s += '&cid=' + quote(path)
s += '&md=0'
return s
def build_dlna_play_single(udn, server_type, path):
s = "dlna-playsingle://" + quote(udn)
s += "?"
s += 'sid=' + quote(server_type)
s += '&iid=' + quote(path)
return s
def _append_query_parms(self, query_parms, prop_name, prop_match):
if isinstance(prop_match, (list, tuple)):
for pm in prop_match:
self._append_query_parms(query_parms, prop_name, pm)
else:
# Just in case, we also escape the property name
parm_name = quote(prop_name, safe='')
parm_value = quote(str(prop_match), safe='')
qp = '{}={}'.format(parm_name, parm_value)
query_parms.append(qp)
def append_url_with_template_parameters(url,
parameters):
"""Replaces template parameters in the given url.
Args:
url (str): The query url string to replace the template parameters.
parameters (dict): The parameters to replace in the url.
Returns:
str: URL with replaced parameters.
"""
# Parameter validation
if url is None:
raise ValueError("URL is None.")
if parameters is None:
return url
# Iterate and replace parameters
for key in parameters:
element = parameters[key]
replace_value = ""
# Load parameter value
if element is None:
replace_value = ""
elif isinstance(element, list):
replace_value = "/".join(quote(str(x), safe='') for x in element)
else:
replace_value = quote(str(element), safe='')
url = url.replace('{{{0}}}'.format(key), str(replace_value))
return url
def last_on_branch(cls, project_id, branch, api):
info = api.call(GET(
'/projects/{project_id}/repository/branches/{branch}'.format(
project_id=project_id,
branch=quote(branch, safe=''),
),
))['commit']
return cls(api, info)
def fetch_plugin_license(base_url, plugin_key, auth=None):
if not plugin_key:
return None
plugin_key = quote(str(plugin_key), '')
endpoint = ('/rest/plugins/1.0/{plugin_key}/license'
.format(plugin_key=plugin_key))
response = do_request('get', base_url, endpoint, auth=auth)
if not response.ok:
return
return response.json()
def get_fetch_plugin_license_request(base_url, plugin_key, auth=None):
if not plugin_key:
return None
plugin_key = quote(str(plugin_key), '')
endpoint = ('/rest/plugins/1.0/{plugin_key}/license'
.format(plugin_key=plugin_key))
return grequests.get(base_url + endpoint, auth=auth)
def fetch_plugin_versions(base_url, plugin_key, params={}):
if not plugin_key:
return
plugin_key = quote(str(plugin_key), '')
endpoint = ('/rest/2/addons/{plugin_key}/versions'
.format(plugin_key=plugin_key))
response = do_request('get', base_url, endpoint, params)
if not response.ok:
return
return response.json()['_embedded']['versions']
def get_fetch_plugin_versions_request(base_url, plugin_key, params={}):
if not plugin_key:
return
plugin_key = quote(str(plugin_key), '')
endpoint = ('/rest/2/addons/{plugin_key}/versions'
.format(plugin_key=plugin_key))
return grequests.get(base_url + endpoint, params=params)
def get_branch_request(base_url, api, project, branch, auth=None):
if not project or not branch:
return None
project = quote(str(project), '')
branch = quote(str(branch), '')
endpoint = ('/api/{api}/projects/{project}/repository/branches/{branch}'
.format(api=api, project=project, branch=branch))
return get_request('get', base_url, endpoint, auth=auth)
def url(self, *fragments):
base = '{scheme}://{host}:{port}{prefix}'.format(
scheme='https' if self._https else 'http',
host=self.host,
port=self.port,
prefix=self._prefix,
)
if fragments:
base = "{base}/{path}".format(base=base, path='/'.join(quote(fragment) for fragment in fragments))
return base
def append_url_with_template_parameters(url,
parameters):
"""Replaces template parameters in the given url.
Args:
url (str): The query url string to replace the template parameters.
parameters (dict): The parameters to replace in the url.
Returns:
str: Url with replaced parameters.
"""
# Parameter validation
if url is None:
raise ValueError("URL is None.")
if parameters is None:
return url
# Iterate and replace parameters
for key in parameters:
element = parameters[key]
replace_value = ""
# Load parameter value
if element is None:
replace_value = ""
elif isinstance(element, list):
replace_value = "/".join(quote(str(x), safe='') for x in element)
else:
replace_value = quote(str(element), safe='')
url = url.replace('{{{0}}}'.format(key),str(replace_value))
return url
def _sort_parameters(self, query):
# split on ampersand
params = query.split('&')
# split each param into two-tuples of (key,value) and quote tuple entries
params = map(lambda p: map(quote, p.split('=', 1)), params)
# sort based on key portion
params = sorted(params, key=lambda p: p[0])
# join back together on ampersand
return '&'.join(map(lambda p: '='.join(p), params))
# ICE's current automatic limit on results returned in the absence of a specific requested
# page size
def append_url_with_template_parameters(url,
parameters):
"""Replaces template parameters in the given url.
Args:
url (str): The query url string to replace the template parameters.
parameters (dict): The parameters to replace in the url.
Returns:
str: URL with replaced parameters.
"""
# Parameter validation
if url is None:
raise ValueError("URL is None.")
if parameters is None:
return url
# Iterate and replace parameters
for key in parameters:
element = parameters[key]
replace_value = ""
# Load parameter value
if element is None:
replace_value = ""
elif isinstance(element, list):
replace_value = "/".join(quote(str(x), safe='') for x in element)
else:
replace_value = quote(str(element), safe='')
url = url.replace('{{{0}}}'.format(key), str(replace_value))
return url
def mount_iso_image(self, image, image_name, ins_file_name):
"""
Upload an ISO image and associate it to this Partition
using the HMC operation 'Mount ISO Image'.
When the partition already has an ISO image associated,
the newly uploaded image replaces the current one.
Authorization requirements:
* Object-access permission to this Partition.
* Task permission to the "Partition Details" task.
Parameters:
image (:term:`byte string` or file-like object):
The content of the ISO image.
Images larger than 2GB cannot be specified as a Byte string; they
must be specified as a file-like object.
File-like objects must have opened the file in binary mode.
image_name (:term:`string`): The displayable name of the image.
This value must be a valid Linux file name without directories,
must not contain blanks, and must end with '.iso' in lower case.
This value will be shown in the 'boot-iso-image-name' property of
this partition.
ins_file_name (:term:`string`): The path name of the INS file within
the file system of the ISO image.
This value will be shown in the 'boot-iso-ins-file' property of
this partition.
Raises:
:exc:`~zhmcclient.HTTPError`
:exc:`~zhmcclient.ParseError`
:exc:`~zhmcclient.AuthError`
:exc:`~zhmcclient.ConnectionError`
"""
query_parms_str = '?image-name={}&ins-file-name={}'. \
format(quote(image_name, safe=''), quote(ins_file_name, safe=''))
self.manager.session.post(
self.uri + '/operations/mount-iso-image' + query_parms_str,
body=image)
def append_url_with_query_parameters(url,
parameters,
array_serialization="indexed"):
"""Adds query parameters to a URL.
Args:
url (str): The URL string.
parameters (dict): The query parameters to add to the URL.
array_serialization (str): The format of array parameter serialization.
Returns:
str: URL with added query parameters.
"""
# Parameter validation
if url is None:
raise ValueError("URL is None.")
if parameters is None:
return url
for key, value in parameters.items():
seperator = '&' if '?' in url else '?'
if not value is None:
if isinstance(value, list):
value = [element for element in value if element]
if array_serialization is "csv":
url += "{0}{1}={2}".format(seperator, key,
",".join(quote(str(x), safe='') for x in value))
elif array_serialization is "psv":
url += "{0}{1}={2}".format(seperator, key,
"|".join(quote(str(x), safe='') for x in value))
elif array_serialization is "tsv":
url += "{0}{1}={2}".format(seperator, key,
"\t".join(quote(str(x), safe='') for x in value))
else:
url += "{0}{1}".format(seperator,
"&".join(("{0}={1}".format(k, quote(str(v), safe='')))
for k, v in APIHelper.serialize_array(key, value, array_serialization)))
else:
url += "{0}{1}={2}".format(seperator, key, quote(str(value), safe=''))
return url
def append_url_with_query_parameters(url,
parameters,
array_serialization="indexed"):
"""Adds query parameters to a URL.
Args:
url (str): The URL string.
parameters (dict): The query parameters to add to the URL.
array_serialization (str): The format of array parameter serialization.
Returns:
str: URL with added query parameters.
"""
# Parameter validation
if url is None:
raise ValueError("URL is None.")
if parameters is None:
return url
for key, value in parameters.items():
seperator = '&' if '?' in url else '?'
if value is not None:
if isinstance(value, list):
value = [element for element in value if element]
if array_serialization is "csv":
url += "{0}{1}={2}".format(seperator, key,
",".join(quote(str(x), safe='') for x in value))
elif array_serialization is "psv":
url += "{0}{1}={2}".format(seperator, key,
"|".join(quote(str(x), safe='') for x in value))
elif array_serialization is "tsv":
url += "{0}{1}={2}".format(seperator, key,
"\t".join(quote(str(x), safe='') for x in value))
else:
url += "{0}{1}".format(seperator,
"&".join(("{0}={1}".format(k, quote(str(v), safe='')))
for k, v in APIHelper.serialize_array(key, value, array_serialization)))
else:
url += "{0}{1}={2}".format(seperator, key, quote(str(value), safe=''))
return url
