def __call__(self, r):
raise NotImplementedError('Auth hooks must be callable.')
+
class HTTPBasicAuth(AuthBase):
"""Attaches HTTP Basic Authentication to the given Request object."""
def __init__(self, username, password):
# XXX should the partial digests be encoded too?
base = 'username="%s", realm="%s", nonce="%s", uri="%s", ' \
- 'response="%s"' % (self.username, realm, nonce, path, respdig)
+ 'response="%s"' % (self.username, realm, nonce, path, respdig)
if opaque:
base += ', opaque="%s"' % opaque
if algorithm:
def handle_401(self, r):
"""Takes the given response and tries digest-auth, if needed."""
- num_401_calls = getattr(self, 'num_401_calls', 1)
+ num_401_calls = getattr(self, 'num_401_calls', 1)
s_auth = r.headers.get('www-authenticate', '')
if 'digest' in s_auth.lower() and num_401_calls < 2:
# to allow our new request to reuse the same one.
r.content
r.raw.release_conn()
-
+
r.request.headers['Authorization'] = self.build_digest_header(r.request.method, r.request.url)
_r = r.connection.send(r.request)
_r.history.append(r)
return _r
- setattr(self, 'num_401_calls', 1)
+ setattr(self, 'num_401_calls', 1)
return r
def __call__(self, r):
import os.path
+
def where():
"""Return the preferred certificate bundle."""
# vendored bundle inside Requests
numeric_types = (int, long, float)
-
elif is_py3:
from urllib.parse import urlparse, urlunparse, urljoin, urlsplit, urlencode, quote, unquote, quote_plus, unquote_plus, urldefrag
from urllib.request import parse_http_list
HOOKS = ['response']
+
def default_hooks():
hooks = {}
for event in HOOKS:
# TODO: response is the only one
+
def dispatch_hook(key, hooks, hook_data):
"""Dispatches a hook dictionary on a given piece of data."""
cookies=None,
hooks=None):
-
# Default empty dicts for dict params.
data = [] if data is None else data
files = [] if files is None else files
pending = None
- for chunk in self.iter_content(
- chunk_size=chunk_size,
- decode_unicode=decode_unicode):
+ for chunk in self.iter_content(chunk_size=chunk_size,
+ decode_unicode=decode_unicode):
if pending is not None:
chunk = pending + chunk
pass
resp = self.request(
- url=url,
- method=method,
- headers=headers,
- auth=req.auth,
- cookies=req.cookies,
- allow_redirects=False,
- stream=stream,
- timeout=timeout,
- verify=verify,
- cert=cert,
- proxies=proxies,
- hooks=req.hooks,
+ url=url,
+ method=method,
+ headers=headers,
+ auth=req.auth,
+ cookies=req.cookies,
+ allow_redirects=False,
+ stream=stream,
+ timeout=timeout,
+ verify=verify,
+ cert=cert,
+ proxies=proxies,
+ hooks=req.hooks,
)
i += 1
if not verify and verify is not False:
verify = os.environ.get('CURL_CA_BUNDLE')
-
# Merge all the kwargs.
params = merge_kwargs(params, self.params)
headers = merge_kwargs(headers, self.headers)
verify = merge_kwargs(verify, self.verify)
cert = merge_kwargs(cert, self.cert)
-
# Create the Request.
req = Request()
req.method = method.upper()
import os
from itertools import islice
+
class IteratorProxy(object):
"""docstring for IteratorProxy"""
def __init__(self, i):
def read(self, n):
return "".join(islice(self.i, None, n))
+
class CaseInsensitiveDict(dict):
"""Case-insensitive Dictionary
DEFAULT_CA_BUNDLE_PATH = certs.where()
+
def dict_to_sequence(d):
"""Returns an internal sequence dictionary update."""
return d
+
def super_len(o):
if hasattr(o, '__len__'):
return len(o)
if hasattr(o, 'fileno'):
return os.fstat(o.fileno()).st_size
+
def get_netrc_auth(url):
"""Returns the Requests tuple auth for a given url from netrc."""
if _implementation == 'CPython':
_implementation_version = platform.python_version()
elif _implementation == 'PyPy':
- _implementation_version = '%s.%s.%s' % (
- sys.pypy_version_info.major,
+ _implementation_version = '%s.%s.%s' % (sys.pypy_version_info.major,
sys.pypy_version_info.minor,
- sys.pypy_version_info.micro
- )
+ sys.pypy_version_info.micro)
if sys.pypy_version_info.releaselevel != 'final':
_implementation_version = ''.join([_implementation_version, sys.pypy_version_info.releaselevel])
elif _implementation == 'Jython':
p_system = 'Unknown'
p_release = 'Unknown'
- return " ".join([
- 'python-requests/%s' % __version__,
- '%s/%s' % (_implementation, _implementation_version),
- '%s/%s' % (p_system, p_release),
- ])
+ return " ".join(['python-requests/%s' % __version__,
+ '%s/%s' % (_implementation, _implementation_version),
+ '%s/%s' % (p_system, p_release)])
+
def default_headers():
return {
for param in params.split(";"):
try:
- key,value = param.split("=")
+ key, value = param.split("=")
except ValueError:
break
HTTPBIN = os.environ.get('HTTPBIN_URL', 'http://httpbin.org/')
+
def httpbin(*suffix):
"""Returns url for HTTPBIN resource."""
return HTTPBIN + '/'.join(suffix)
+
class RequestsTestCase(unittest.TestCase):
_multiprocess_can_split_ = True
assert pr.url == req.url
assert pr.body == 'life=42'
-
def test_no_content_length(self):
get_req = requests.Request('GET', httpbin('get')).prepare()
self.assertTrue('Content-Length' not in get_req.headers)
head_req = requests.Request('HEAD', httpbin('head')).prepare()
self.assertTrue('Content-Length' not in head_req.headers)
-
def test_path_is_not_double_encoded(self):
request = requests.Request('GET', "http://0.0.0.0/get/test case").prepare()
def test_user_agent_transfers(self):
heads = {
- 'User-agent':
- 'Mozilla/5.0 (github.com/kennethreitz/requests)'
+ 'User-agent': 'Mozilla/5.0 (github.com/kennethreitz/requests)'
}
r = requests.get(httpbin('user-agent'), headers=heads)
self.assertTrue(heads['User-agent'] in r.text)
heads = {
- 'user-agent':
- 'Mozilla/5.0 (github.com/kennethreitz/requests)'
+ 'user-agent': 'Mozilla/5.0 (github.com/kennethreitz/requests)'
}
r = requests.get(httpbin('user-agent'), headers=heads)