我们从Python开源项目中,提取了以下24个代码示例,用于说明如何使用requests.utils.quote()。
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.
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 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