def _handle_auth_error(msg):
"""Handle reraising HTTP authentication errors as something clearer.
If an ``HTTPError`` is raised with status 401 (access denied) in
the body of this context manager, reraise it as an
``AuthenticationError`` instead, with *msg* as its message.
This function adds no round trips to the server.
:param msg: The message to be raised in ``AuthenticationError``.
:type msg: ``str``
**Example**::
with _handle_auth_error("Your login failed."):
... # make an HTTP request
"""
try:
yield
except HTTPError as he:
if he.status == 401:
raise AuthenticationError(msg, he)
else:
raise
python类request()的实例源码
def _auth_headers(self):
"""Headers required to authenticate a request.
Assumes your ``Context`` already has a authentication token or
cookie, either provided explicitly or obtained by logging
into the Splunk instance.
:returns: A list of 2-tuples containing key and value
"""
if self.has_cookies():
return [("Cookie", _make_cookie_header(self.get_cookies().items()))]
elif self.basic and (self.username and self.password):
token = 'Basic %s' % b64encode("%s:%s" % (self.username, self.password))
return [("Authorization", token)]
elif self.token is _NoAuthenticationToken:
return []
else:
# Ensure the token is properly formatted
if self.token.startswith('Splunk '):
token = self.token
else:
token = 'Splunk %s' % self.token
return [("Authorization", token)]
def get(self, url, headers=None, **kwargs):
"""Sends a GET request to a URL.
:param url: The URL.
:type url: ``string``
:param headers: A list of pairs specifying the headers for the HTTP
response (for example, ``[('Content-Type': 'text/cthulhu'), ('Token': 'boris')]``).
:type headers: ``list``
:param kwargs: Additional keyword arguments (optional). These arguments
are interpreted as the query part of the URL. The order of keyword
arguments is not preserved in the request, but the keywords and
their arguments will be URL encoded.
:type kwargs: ``dict``
:returns: A dictionary describing the response (see :class:`HttpLib` for
its structure).
:rtype: ``dict``
"""
if headers is None: headers = []
if kwargs:
# url is already a UrlEncoded. We have to manually declare
# the query to be encoded or it will get automatically URL
# encoded by being appended to url.
url = url + UrlEncoded('?' + _encode(**kwargs), skip_encode=True)
return self.request(url, { 'method': "GET", 'headers': headers })
def _handle_auth_error(msg):
"""Handle reraising HTTP authentication errors as something clearer.
If an ``HTTPError`` is raised with status 401 (access denied) in
the body of this context manager, reraise it as an
``AuthenticationError`` instead, with *msg* as its message.
This function adds no round trips to the server.
:param msg: The message to be raised in ``AuthenticationError``.
:type msg: ``str``
**Example**::
with _handle_auth_error("Your login failed."):
... # make an HTTP request
"""
try:
yield
except HTTPError as he:
if he.status == 401:
raise AuthenticationError(msg, he)
else:
raise
def _auth_headers(self):
"""Headers required to authenticate a request.
Assumes your ``Context`` already has a authentication token or
cookie, either provided explicitly or obtained by logging
into the Splunk instance.
:returns: A list of 2-tuples containing key and value
"""
if self.has_cookies():
return [("Cookie", _make_cookie_header(self.get_cookies().items()))]
elif self.token is _NoAuthenticationToken:
return []
else:
# Ensure the token is properly formatted
if self.token.startswith('Splunk '):
token = self.token
else:
token = 'Splunk %s' % self.token
return [("Authorization", token)]
def get(self, url, headers=None, **kwargs):
"""Sends a GET request to a URL.
:param url: The URL.
:type url: ``string``
:param headers: A list of pairs specifying the headers for the HTTP
response (for example, ``[('Content-Type': 'text/cthulhu'), ('Token': 'boris')]``).
:type headers: ``list``
:param kwargs: Additional keyword arguments (optional). These arguments
are interpreted as the query part of the URL. The order of keyword
arguments is not preserved in the request, but the keywords and
their arguments will be URL encoded.
:type kwargs: ``dict``
:returns: A dictionary describing the response (see :class:`HttpLib` for
its structure).
:rtype: ``dict``
"""
if headers is None: headers = []
if kwargs:
# url is already a UrlEncoded. We have to manually declare
# the query to be encoded or it will get automatically URL
# encoded by being appended to url.
url = url + UrlEncoded('?' + _encode(**kwargs), skip_encode=True)
return self.request(url, { 'method': "GET", 'headers': headers })
def _handle_auth_error(msg):
"""Handle reraising HTTP authentication errors as something clearer.
If an ``HTTPError`` is raised with status 401 (access denied) in
the body of this context manager, reraise it as an
``AuthenticationError`` instead, with *msg* as its message.
This function adds no round trips to the server.
:param msg: The message to be raised in ``AuthenticationError``.
:type msg: ``str``
**Example**::
with _handle_auth_error("Your login failed."):
... # make an HTTP request
"""
try:
yield
except HTTPError as he:
if he.status == 401:
raise AuthenticationError(msg, he)
else:
raise
def _auth_headers(self):
"""Headers required to authenticate a request.
Assumes your ``Context`` already has a authentication token or
cookie, either provided explicitly or obtained by logging
into the Splunk instance.
:returns: A list of 2-tuples containing key and value
"""
if self.has_cookies():
return [("Cookie", _make_cookie_header(self.get_cookies().items()))]
elif self.token is _NoAuthenticationToken:
return []
else:
# Ensure the token is properly formatted
if self.token.startswith('Splunk '):
token = self.token
else:
token = 'Splunk %s' % self.token
return [("Authorization", token)]
def get(self, url, headers=None, **kwargs):
"""Sends a GET request to a URL.
:param url: The URL.
:type url: ``string``
:param headers: A list of pairs specifying the headers for the HTTP
response (for example, ``[('Content-Type': 'text/cthulhu'), ('Token': 'boris')]``).
:type headers: ``list``
:param kwargs: Additional keyword arguments (optional). These arguments
are interpreted as the query part of the URL. The order of keyword
arguments is not preserved in the request, but the keywords and
their arguments will be URL encoded.
:type kwargs: ``dict``
:returns: A dictionary describing the response (see :class:`HttpLib` for
its structure).
:rtype: ``dict``
"""
if headers is None: headers = []
if kwargs:
# url is already a UrlEncoded. We have to manually declare
# the query to be encoded or it will get automatically URL
# encoded by being appended to url.
url = url + UrlEncoded('?' + _encode(**kwargs), skip_encode=True)
return self.request(url, { 'method': "GET", 'headers': headers })
def _handle_auth_error(msg):
"""Handle reraising HTTP authentication errors as something clearer.
If an ``HTTPError`` is raised with status 401 (access denied) in
the body of this context manager, reraise it as an
``AuthenticationError`` instead, with *msg* as its message.
This function adds no round trips to the server.
:param msg: The message to be raised in ``AuthenticationError``.
:type msg: ``str``
**Example**::
with _handle_auth_error("Your login failed."):
... # make an HTTP request
"""
try:
yield
except HTTPError as he:
if he.status == 401:
raise AuthenticationError(msg, he)
else:
raise
def _auth_headers(self):
"""Headers required to authenticate a request.
Assumes your ``Context`` already has a authentication token or
cookie, either provided explicitly or obtained by logging
into the Splunk instance.
:returns: A list of 2-tuples containing key and value
"""
if self.has_cookies():
return [("Cookie", _make_cookie_header(self.get_cookies().items()))]
elif self.basic and (self.username and self.password):
token = 'Basic %s' % b64encode("%s:%s" % (self.username, self.password))
return [("Authorization", token)]
elif self.token is _NoAuthenticationToken:
return []
else:
# Ensure the token is properly formatted
if self.token.startswith('Splunk '):
token = self.token
else:
token = 'Splunk %s' % self.token
return [("Authorization", token)]
def get(self, url, headers=None, **kwargs):
"""Sends a GET request to a URL.
:param url: The URL.
:type url: ``string``
:param headers: A list of pairs specifying the headers for the HTTP
response (for example, ``[('Content-Type': 'text/cthulhu'), ('Token': 'boris')]``).
:type headers: ``list``
:param kwargs: Additional keyword arguments (optional). These arguments
are interpreted as the query part of the URL. The order of keyword
arguments is not preserved in the request, but the keywords and
their arguments will be URL encoded.
:type kwargs: ``dict``
:returns: A dictionary describing the response (see :class:`HttpLib` for
its structure).
:rtype: ``dict``
"""
if headers is None: headers = []
if kwargs:
# url is already a UrlEncoded. We have to manually declare
# the query to be encoded or it will get automatically URL
# encoded by being appended to url.
url = url + UrlEncoded('?' + _encode(**kwargs), skip_encode=True)
return self.request(url, { 'method': "GET", 'headers': headers })
def has_cookies(self):
"""Returns true if the ``HttpLib`` member of this instance has at least
one cookie stored.
:return: ``True`` if there is at least one cookie, else ``False``
:rtype: ``bool``
"""
return len(self.get_cookies()) > 0
# Shared per-context request headers
def _spliturl(url):
scheme, opaque = urllib.splittype(url)
netloc, path = urllib.splithost(opaque)
host, port = urllib.splitport(netloc)
# Strip brackets if its an IPv6 address
if host.startswith('[') and host.endswith(']'): host = host[1:-1]
if port is None: port = DEFAULT_PORT
return scheme, host, port, path
# Given an HTTP request handler, this wrapper objects provides a related
# family of convenience methods built using that handler.
def delete(self, url, headers=None, **kwargs):
"""Sends a DELETE request to a URL.
:param url: The URL.
:type url: ``string``
:param headers: A list of pairs specifying the headers for the HTTP
response (for example, ``[('Content-Type': 'text/cthulhu'), ('Token': 'boris')]``).
:type headers: ``list``
:param kwargs: Additional keyword arguments (optional). These arguments
are interpreted as the query part of the URL. The order of keyword
arguments is not preserved in the request, but the keywords and
their arguments will be URL encoded.
:type kwargs: ``dict``
:returns: A dictionary describing the response (see :class:`HttpLib` for
its structure).
:rtype: ``dict``
"""
if headers is None: headers = []
if kwargs:
# url is already a UrlEncoded. We have to manually declare
# the query to be encoded or it will get automatically URL
# encoded by being appended to url.
url = url + UrlEncoded('?' + _encode(**kwargs), skip_encode=True)
message = {
'method': "DELETE",
'headers': headers,
}
return self.request(url, message)
def post(self, url, headers=None, **kwargs):
"""Sends a POST request to a URL.
:param url: The URL.
:type url: ``string``
:param headers: A list of pairs specifying the headers for the HTTP
response (for example, ``[('Content-Type': 'text/cthulhu'), ('Token': 'boris')]``).
:type headers: ``list``
:param kwargs: Additional keyword arguments (optional). If the argument
is ``body``, the value is used as the body for the request, and the
keywords and their arguments will be URL encoded. If there is no
``body`` keyword argument, all the keyword arguments are encoded
into the body of the request in the format ``x-www-form-urlencoded``.
:type kwargs: ``dict``
:returns: A dictionary describing the response (see :class:`HttpLib` for
its structure).
:rtype: ``dict``
"""
if headers is None: headers = []
headers.append(("Content-Type", "application/x-www-form-urlencoded")),
# We handle GET-style arguments and an unstructured body. This is here
# to support the receivers/stream endpoint.
if 'body' in kwargs:
body = kwargs.pop('body')
if len(kwargs) > 0:
url = url + UrlEncoded('?' + _encode(**kwargs), skip_encode=True)
else:
body = _encode(**kwargs)
message = {
'method': "POST",
'headers': headers,
'body': body
}
return self.request(url, message)
def request(self, url, message, **kwargs):
"""Issues an HTTP request to a URL.
:param url: The URL.
:type url: ``string``
:param message: A dictionary with the format as described in
:class:`HttpLib`.
:type message: ``dict``
:param kwargs: Additional keyword arguments (optional). These arguments
are passed unchanged to the handler.
:type kwargs: ``dict``
:returns: A dictionary describing the response (see :class:`HttpLib` for
its structure).
:rtype: ``dict``
"""
response = self.handler(url, message, **kwargs)
response = record(response)
if 400 <= response.status:
raise HTTPError(response)
# Update the cookie with any HTTP request
# Initially, assume list of 2-tuples
key_value_tuples = response.headers
# If response.headers is a dict, get the key-value pairs as 2-tuples
# this is the case when using urllib2
if isinstance(response.headers, dict):
key_value_tuples = response.headers.items()
for key, value in key_value_tuples:
if key.lower() == "set-cookie":
_parse_cookies(value, self._cookies)
return response
# Converts an httplib response into a file-like object.
def has_cookies(self):
"""Returns true if the ``HttpLib`` member of this instance has at least
one cookie stored.
:return: ``True`` if there is at least one cookie, else ``False``
:rtype: ``bool``
"""
return len(self.get_cookies()) > 0
# Shared per-context request headers
def _spliturl(url):
scheme, opaque = urllib.splittype(url)
netloc, path = urllib.splithost(opaque)
host, port = urllib.splitport(netloc)
# Strip brackets if its an IPv6 address
if host.startswith('[') and host.endswith(']'): host = host[1:-1]
if port is None: port = DEFAULT_PORT
return scheme, host, port, path
# Given an HTTP request handler, this wrapper objects provides a related
# family of convenience methods built using that handler.
def delete(self, url, headers=None, **kwargs):
"""Sends a DELETE request to a URL.
:param url: The URL.
:type url: ``string``
:param headers: A list of pairs specifying the headers for the HTTP
response (for example, ``[('Content-Type': 'text/cthulhu'), ('Token': 'boris')]``).
:type headers: ``list``
:param kwargs: Additional keyword arguments (optional). These arguments
are interpreted as the query part of the URL. The order of keyword
arguments is not preserved in the request, but the keywords and
their arguments will be URL encoded.
:type kwargs: ``dict``
:returns: A dictionary describing the response (see :class:`HttpLib` for
its structure).
:rtype: ``dict``
"""
if headers is None: headers = []
if kwargs:
# url is already a UrlEncoded. We have to manually declare
# the query to be encoded or it will get automatically URL
# encoded by being appended to url.
url = url + UrlEncoded('?' + _encode(**kwargs), skip_encode=True)
message = {
'method': "DELETE",
'headers': headers,
}
return self.request(url, message)
