本文整理汇总了Python中requests.compat.quote函数的典型用法代码示例。如果您正苦于以下问题:Python quote函数的具体用法?Python quote怎么用?Python quote使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了quote函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_get_auth_from_url
def test_get_auth_from_url(self):
""" Ensures that username and password in well-encoded URI as per RFC 3986 are correclty extracted """
from requests.utils import get_auth_from_url
from requests.compat import quote
percent_encoding_test_chars = "%!*'();:@&=+$,/?#[] "
url_address = "request.com/url.html#test"
url = "http://" + quote(percent_encoding_test_chars, '') + ':' + quote(percent_encoding_test_chars, '') + '@' + url_address
(username, password) = get_auth_from_url(url)
assert username == percent_encoding_test_chars
assert password == percent_encoding_test_chars
开发者ID:zenlambda,项目名称:appengine-python3,代码行数:10,代码来源:test_requests.py
示例2: uri
def uri(self):
"""
"""
uri = self.application_uri
path_info = quote(self.path_info, safe='/=;,')
if not self.env['SCRIPT_NAME']:
uri += path_info[1:]
else:
uri += path_info
if self.query_string:
uri += '?' + quote(self.query_string, safe='&=;,')
return uri
开发者ID:agrausem,项目名称:britney,代码行数:15,代码来源:request.py
示例3: get_entities
def get_entities(self, group_name, expression=None, min_insert_date=None, max_insert_date=None, tags=None,
limit=None):
"""Retrieve a list of entities that are members of the specified entity group and match the specified expression filter.
:param group_name: `str`
:param expression: `str`
:param min_insert_date: `str` | `int` | :class:`datetime`
:param max_insert_date: `str` | `int` | :class:`datetime`
:param tags: `dict`
:param limit: `int`
:return: `list` of :class:`.Entity` objects
"""
_check_name(group_name)
params = {}
if expression is not None:
params["expression"] = expression
if min_insert_date is not None:
params["minInsertDate"] = to_iso(min_insert_date)
if max_insert_date is not None:
params["maxInsertDate"] = to_iso(max_insert_date)
if tags is not None:
params["tags"] = tags
if limit is not None:
params["limit"] = limit
resp = self.conn.get(eg_get_entities_url.format(group=quote(group_name, '')), params)
return _jsonutil.deserialize(resp, Entity)
开发者ID:axibase,项目名称:atsd-api-python,代码行数:26,代码来源:services.py
示例4: series
def series(self, metric, entity=None, tags=None, min_insert_date=None, max_insert_date=None):
"""Retrieve series for the specified metric.
:param metric: `str` | :class:`.Metric`
:param entity: `str` | :class:`.Entity`
:param tags: `dict`
:param min_insert_date: `int` | `str` | None | :class:`datetime`
:param max_insert_date: `int` | `str` | None | :class:`datetime`
:return: :class:`.Series`
"""
metric_name = metric.name if isinstance(metric, Metric) else metric
_check_name(metric_name)
params = {}
if entity is not None:
params['entity'] = entity.name if isinstance(entity, Entity) else entity
if tags is not None and isinstance(tags, dict):
for k, v in tags.items():
params['tags.%s' % k] = v
if min_insert_date is not None:
params['minInsertDate'] = to_iso(min_insert_date)
if max_insert_date is not None:
params['maxInsertDate'] = to_iso(max_insert_date)
try:
response = self.conn.get(metric_series_url.format(metric=quote(metric_name, '')),
params)
except ServerException as e:
if e.status_code == 404:
return []
else:
raise e
return _jsonutil.deserialize(response, Series)
开发者ID:axibase,项目名称:atsd-api-python,代码行数:34,代码来源:services.py
示例5: metrics
def metrics(self, entity, expression=None, min_insert_date=None, max_insert_date=None, use_entity_insert_time=False,
limit=None, tags=None):
"""Retrieve a `list` of metrics matching the specified filters.
:param entity: `str` | :class:`.Entity`
:param expression: `str`
:param min_insert_date: `int` | `str` | None | :class:`datetime`
:param max_insert_date: `int` | `str` | None | :class:`datetime`
:param use_entity_insert_time: `bool` If true, last_insert_date is calculated for the specified entity and metric
:param limit: `int`
:param tags: `str`
:return: :class:`.Metric` objects
"""
entity_name = entity.name if isinstance(entity, Entity) else entity
_check_name(entity_name)
params = {}
if expression is not None:
params['expression'] = expression
if min_insert_date is not None:
params['minInsertDate'] = to_iso(min_insert_date)
if max_insert_date is not None:
params['maxInsertDate'] = to_iso(max_insert_date)
params['useEntityInsertTime'] = use_entity_insert_time
if limit is not None:
params['limit'] = limit
if tags is not None:
params['tags'] = tags
response = self.conn.get(ent_metrics_url.format(entity=quote(entity_name, '')), params)
return _jsonutil.deserialize(response, Metric)
开发者ID:axibase,项目名称:atsd-api-python,代码行数:29,代码来源:services.py
示例6: request
def request(self, method, endpoint, body='', **kwargs):
"""Make a request to the PyCon website, and return the result."""
# The PyCon website runs using HTTPS, but localhost doesn't.
# Determine the right thing.
protocol = 'https'
if 'localhost' in self.host:
protocol = 'http'
# Construct the base URI.
uri = '/2014/pycon_api/%s/' % endpoint
# If keyword arguments are provided, append them to
# the URI.
if kwargs:
uri += '?' + '&'.join(
['%s=%s' % (k, quote(str(v))) for k, v in kwargs.items()],
)
# Construct the full URL.
url = '{protocol}://{host}{uri}'.format(
host=self.host,
protocol=protocol,
uri=uri,
)
# Generate the appropriate request signature to certify
# that this is a valid request.
signature = self._sign_request(uri, method, body)
# Add the appropriate content-type header.
if method == 'POST':
signature['Content-Type'] = 'application/json'
# Make the actual request to the PyCon website.
r = requests.request(method, url, data=body, headers=signature,
verify=False)
# Sanity check: Did we get a bad request of some kind?
if r.status_code >= 400:
# If we got a 500 response, we can't really know what to do
if r.status_code >= 500:
raise InternalServerError(r)
# What exception class shall I use?
exc_class = APIError
if r.status_code == 403:
exc_class = AuthenticationError
if r.status_code == 404:
exc_class = NotFound
# Create and raise the exception
try:
ex = exc_class(r.json()['error'])
except ValueError:
raise InternalServerError(r)
ex.request = r
raise ex
# OK, all is well; return the response.
return r.json()
开发者ID:econchick,项目名称:pc-bot,代码行数:60,代码来源:api.py
示例7: config
def config(self):
url = 'job/%s/config.xml' % quote(self.name)
res = self.server.get(url)
if res.status_code != 200 or res.headers.get('content-type', '') != 'application/xml':
msg = 'fetching configuration for job "%s" did not return a xml document'
raise JenkinsError(msg % self.name)
return res.text
开发者ID:Recras,项目名称:jenkins-webapi,代码行数:7,代码来源:jenkins.py
示例8: create_or_replace
def create_or_replace(self, entity):
"""Create an entity or update an existing entity.
:param entity: :class:`.Entity`
:return: True if success
"""
self.conn.put(ent_create_or_replace_url.format(entity=quote(entity.name, '')), entity)
return True
开发者ID:axibase,项目名称:atsd-api-python,代码行数:8,代码来源:services.py
示例9: update
def update(self, entity):
"""Update the specified entity.
:param entity: :class:`.Entity`
:return: True if success
"""
self.conn.patch(ent_update_url.format(entity=quote(entity.name, '')), entity)
return True
开发者ID:axibase,项目名称:atsd-api-python,代码行数:8,代码来源:services.py
示例10: delete
def delete(self, metric_name):
"""Delete the specified metric.
:param metric_name: :class:`.Metric`
:return: True if success
"""
self.conn.delete(metric_delete_url.format(metric=quote(metric_name, '')))
return True
开发者ID:axibase,项目名称:atsd-api-python,代码行数:8,代码来源:services.py
示例11: type_query
def type_query(self, entity):
"""Returns an array of property types for the entity.
:param entity: :class:`.Entity`
:return: returns `list` of property types for the entity.
"""
entity_name = entity.name if isinstance(entity, Entity) else entity
response = self.conn.get(properties_types_url.format(entity=quote(entity_name, '')))
return response
开发者ID:axibase,项目名称:atsd-api-python,代码行数:9,代码来源:services.py
示例12: delete
def delete(self):
'''Permanently remove job.'''
self._not_exist_raise()
url = 'job/%s/doDelete' % quote(self.name)
self.server.post(url)
if self.exists():
raise JenkinsError('delete of job "%s" failed' % self.name)
开发者ID:Recras,项目名称:jenkins-webapi,代码行数:9,代码来源:jenkins.py
示例13: __init__
def __init__(self, content_type="application/json", sock_path=None):
if sock_path is None:
sock_path = self.secrets_sock_path
self.content_type = content_type
self.session = requests.Session()
self.session.mount("http+unix://", HTTPUnixAdapter())
self.headers = dict({"Content-Type": content_type})
self.url = "http+unix://" + quote(sock_path, safe="") + "/" + self.secrets_container
self._last_response = None
开发者ID:jhrozek,项目名称:sssd,代码行数:10,代码来源:secrets.py
示例14: _sendFreeMobileSMS
def _sendFreeMobileSMS(self, title, msg, cust_id=None, apiKey=None):
"""
Sends a SMS notification
msg: The message to send (unicode)
title: The title of the message
userKey: The pushover user id to send the message to (or to subscribe with)
returns: True if the message succeeded, False otherwise
"""
if cust_id is None:
cust_id = sickbeard.FREEMOBILE_ID
if apiKey is None:
apiKey = sickbeard.FREEMOBILE_APIKEY
logger.log(u"Free Mobile in use with API KEY: " + apiKey, logger.DEBUG)
# build up the URL and parameters
msg = msg.strip()
msg_quoted = quote(title.encode('utf-8') + ": " + msg.encode('utf-8'))
URL = "https://smsapi.free-mobile.fr/sendmsg?user=" + cust_id + "&pass=" + apiKey + "&msg=" + msg_quoted
req = Request(URL)
# send the request to Free Mobile
try:
urlopen(req)
except IOError as e:
if hasattr(e, 'code'):
if e.code == 400:
message = "Missing parameter(s)."
logger.log(message, logger.ERROR)
return False, message
if e.code == 402:
message = "Too much SMS sent in a short time."
logger.log(message, logger.ERROR)
return False, message
if e.code == 403:
message = "API service isn't enabled in your account or ID / API key is incorrect."
logger.log(message, logger.ERROR)
return False, message
if e.code == 500:
message = "Server error. Please retry in few moment."
logger.log(message, logger.ERROR)
return False, message
except Exception as e:
message = u"Error while sending SMS: {0}".format(e)
logger.log(message, logger.ERROR)
return False, message
message = "Free Mobile SMS successful."
logger.log(message, logger.INFO)
return True, message
开发者ID:Thraxis,项目名称:pymedusa,代码行数:53,代码来源:freemobile.py
示例15: __init__
def __init__(self, content_type='application/json', sock_path=None):
if sock_path is None:
sock_path = self.secrets_sock_path
self.content_type = content_type
self.session = requests.Session()
self.session.mount('http+unix://', HTTPUnixAdapter())
self.headers = dict({'Content-Type': content_type})
self.url = 'http+unix://' + \
quote(sock_path, safe='') + \
'/' + \
self.secrets_container
self._last_response = None
开发者ID:SSSD,项目名称:sssd,代码行数:13,代码来源:secrets.py
示例16: get
def get(self, name):
"""Retrieve metric.
:param name: `str` metric name
:return: :class:`.Metric`
"""
_check_name(name)
try:
response = self.conn.get(metric_get_url.format(metric=quote(name, '')))
except ServerException as e:
if e.status_code == 404:
return None
else:
raise e
return _jsonutil.deserialize(response, Metric)
开发者ID:axibase,项目名称:atsd-api-python,代码行数:15,代码来源:services.py
示例17: format
def format(self, record):
"""Strip censored items from string.
:param record: to censor
:type record: logging.LogRecord
:return:
:rtype: str
"""
privacy_level = sickbeard.common.privacy_levels[sickbeard.PRIVACY_LEVEL]
if not privacy_level:
return super(CensoredFormatter, self).format(record)
elif privacy_level == sickbeard.common.privacy_levels['absurd']:
return re.sub(r'[\d\w]', '*', super(CensoredFormatter, self).format(record))
else:
msg = super(CensoredFormatter, self).format(record)
if not isinstance(msg, text_type):
msg = msg.decode(self.encoding, 'replace') # Convert to unicode
# Change the SSL error to a warning with a link to information about how to fix it.
# Check for u'error [SSL: SSLV3_ALERT_HANDSHAKE_FAILURE] sslv3 alert handshake failure (_ssl.c:590)'
for ssl_error in SSL_ERRORS:
if re.findall(ssl_error, msg):
record.levelno = WARNING
record.levelname = logging.getLevelName(record.levelno)
msg = super(CensoredFormatter, self).format(record)
msg = re.sub(ssl_error, SSL_ERROR_HELP_MSG, msg)
# set of censored items
censored = {value for value in itervalues(censored_items) if value}
# set of censored items and urlencoded counterparts
censored = censored | {quote(item) for item in censored}
# convert set items to unicode and typecast to list
censored = list({
item.decode(self.encoding, 'replace')
if not isinstance(item, text_type) else item
for item in censored
})
# sort the list in order of descending length so that entire item is censored
# e.g. password and password_1 both get censored instead of getting ********_1
censored.sort(key=len, reverse=True)
for item in censored:
msg = msg.replace(item, len(item) * '*')
# Needed because Newznab apikey isn't stored as key=value in a section.
msg = re.sub(r'([&?]r|[&?]apikey|[&?]api_key)(?:=|%3D)[^&]*([&\w]?)', r'\1=**********\2', msg, re.I)
return msg
开发者ID:Thraxis,项目名称:pymedusa,代码行数:48,代码来源:logger.py
示例18: build_params
def build_params(self, params):
"""
quotes values and converts keys of params to camelCase from underscore
:param params: dict
:return:
"""
out = {}
for key, value in params.iteritems():
if not isinstance(value, types.StringTypes):
value = str(value)
elif isinstance(value, unicode):
value = value.encode("utf-8")
key = key.split('_')[0] + ''.join(x.capitalize() for x in key.split('_')[1:])
out[key] = quote(value)
return out
开发者ID:pannal,项目名称:Sub-Zero.bundle,代码行数:17,代码来源:drone.py
示例19: _sendFreeMobileSMS
def _sendFreeMobileSMS(self, title, msg, cust_id=None, apiKey=None):
"""
Send a SMS notification
msg: The message to send (unicode)
title: The title of the message
userKey: The pushover user id to send the message to (or to subscribe with)
return: True if the message succeeded, False otherwise
"""
if cust_id is None:
cust_id = app.FREEMOBILE_ID
if apiKey is None:
apiKey = app.FREEMOBILE_APIKEY
log.debug(u'Free Mobile in use with API KEY: {0}', apiKey)
# build up the URL and parameters
msg = msg.strip()
msg_quoted = quote(title.encode('utf-8') + ': ' + msg.encode('utf-8'))
URL = 'https://smsapi.free-mobile.fr/sendmsg?user=' + cust_id + '&pass=' + apiKey + '&msg=' + msg_quoted
req = Request(URL)
# send the request to Free Mobile
try:
urlopen(req)
except IOError as e:
if hasattr(e, 'code'):
error_message = {
400: 'Missing parameter(s).',
402: 'Too much SMS sent in a short time.',
403: 'API service is not enabled in your account or ID / API key is incorrect.',
500: 'Server error. Please retry in few moment.',
}
message = error_message.get(e.code)
if message:
log.error(message)
return False, message
except Exception as e:
message = u'Error while sending SMS: {0}'.format(e)
log.error(message)
return False, message
message = 'Free Mobile SMS successful.'
log.info(message)
return True, message
开发者ID:pymedusa,项目名称:SickRage,代码行数:46,代码来源:freemobile.py
示例20: delete_entities
def delete_entities(self, group_name, entities):
"""Remove specified entities from members of the specified entity group.
To delete all entities, submit an empty list [] using the set_entities method.
Changing members of expression-based groups is not supported.
:param group_name: `str`
:param entities: `list` of :class:`.Entity` objects | `list` of `str` entity names
:return: True if success
"""
_check_name(group_name)
data = []
for e in entities:
data.append(e.name if isinstance(e, Entity) else e)
response = self.conn.post(eg_delete_entities_url.format(group=quote(group_name, '')),
data=data)
return True
开发者ID:axibase,项目名称:atsd-api-python,代码行数:17,代码来源:services.py
注:本文中的requests.compat.quote函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论