Get Even More Visitors To Your Blog, Upgrade To A Business Listing >>

# -*- coding: utf-8 -*-


This module provides a Session object to manage and persist settings across
requests (cookies, auth, proxies).
import os
import platform
import time
from collections import Mapping
from datetime import timedelta

from .auth import _basic_auth_str
from .compat import cookielib, is_py3, OrderedDict, urljoin, urlparse
from .cookies import (
cookiejar_from_dict, extract_cookies_to_jar, RequestsCookieJar, merge_cookies)
from .models import Request, PreparedRequest, DEFAULT_REDIRECT_LIMIT
from .hooks import default_hooks, dispatch_hook
from ._internal_utils import to_native_string
from .utils import to_key_val_list, default_headers
from .exceptions import (
TooManyRedirects, InvalidSchema, ChunkedEncodingError, ContentDecodingError)

from .structures import CaseInsensitiveDict
from .adapters import HTTPAdapter

from .utils import (
requote_uri, get_environ_proxies, get_netrc_auth, should_bypass_proxies,
get_auth_from_url, rewind_body

from .status_codes import codes

# formerly defined here, reexposed here for backward compatibility
from .models import REDIRECT_STATI

# Preferred clock, based on which one is more accurate on a given system.
if platform.system() == 'Windows':
try: # Python 3.3+
preferred_clock = time.perf_counter
except AttributeError: # Earlier than Python 3.
preferred_clock = time.clock
preferred_clock = time.time

def merge_setting(request_setting, session_setting, dict_class=OrderedDict):
"""Determines appropriate setting for a given request, taking into account
the explicit setting on that request, and the setting in the session. If a
setting is a dictionary, they will be merged together using `dict_class`

if session_setting is None:
return request_setting

if request_setting is None:
return session_setting

# Bypass if not a dictionary (e.g. verify)
if not (
isinstance(session_setting, Mapping) and
isinstance(request_setting, Mapping)
return request_setting

merged_setting = dict_class(to_key_val_list(session_setting))

# Remove keys that are set to None. Extract keys first to avoid altering
# the dictionary during iteration.
none_keys = [k for (k, v) in merged_setting.items() if v is None]
for key in none_keys:
del merged_setting[key]

return merged_setting

def merge_hooks(request_hooks, session_hooks, dict_class=OrderedDict):
"""Properly merges both requests and session hooks.

This is necessary because when request_hooks == {'response': []}, the
merge breaks Session hooks entirely.
if session_hooks is None or session_hooks.get('response') == []:
return request_hooks

if request_hooks is None or request_hooks.get('response') == []:
return session_hooks

return merge_setting(request_hooks, session_hooks, dict_class)

class SessionRedirectMixin(object):

def get_redirect_target(self, resp):
"""Receives a Response. Returns a redirect URI or ``None``"""
# Due to the nature of how requests processes redirects this method will
# be called at least once upon the original response and at least twice
# on each subsequent redirect response (if any).
# If a custom mixin is used to handle this logic, it may be advantageous
# to cache the redirect location onto the response object as a private
# attribute.
if resp.is_redirect:
location = resp.headers['location']
# Currently the underlying http module on py3 decode headers
# in latin1, but empirical evidence suggests that latin1 is very
# rarely used with non-ASCII characters in HTTP headers.
# It is more likely to get UTF8 header rather than latin1.
# This causes incorrect handling of UTF8 encoded location headers.
# To solve this, we re-encode the location in latin1.
if is_py3:
location = location.encode('latin1')
return to_native_string(location, 'utf8')
return None

def resolve_redirects(self, resp, req, stream=False, timeout=None,
verify=True, cert=None, proxies=None, yield_requests=False, **adapter_kwargs):
"""Receives a Response. Returns a generator of Responses or Requests."""

hist = [] # keep track of history

url = self.get_redirect_target(resp)
while url:
prepared_request = req.copy()

# Update history and keep track of redirects.
# resp.history must ignore the original request in this loop
resp.history = hist[1:]

resp.content # Consume socket so it can be released
except (ChunkedEncodingError, ContentDecodingError, RuntimeError):

if len(resp.history) >= self.max_redirects:
raise TooManyRedirects('Exceeded %s redirects.' % self.max_redirects, response=resp)

# Release the connection back into the pool.

# Handle redirection without scheme (see: RFC 1808 Section 4)
if url.startswith('//'):
parsed_rurl = urlparse(resp.url)
url = '%s:%s' % (to_native_string(parsed_rurl.scheme), url)

# The scheme should be lower case...
parsed = urlparse(url)
url = parsed.geturl()

# Facilitate relative 'location' headers, as allowed by RFC 7231.
# (e.g. '/path/to/resource' instead of 'http://domain.tld/path/to/resource')
# Compliant with RFC3986, we percent encode the url.
if not parsed.netloc:
url = urljoin(resp.url, requote_uri(url))
url = requote_uri(url)

prepared_request.url = to_native_string(url)

self.rebuild_method(prepared_request, resp)

if resp.status_code not in (codes.temporary_redirect, codes.permanent_redirect):
purged_headers = ('Content-Length', 'Content-Type', 'Transfer-Encoding')
for header in purged_headers:
prepared_request.headers.pop(header, None)
prepared_request.body = None

headers = prepared_request.headers
del headers['Cookie']
except KeyError:

# Extract any cookies sent on the response to the cookiejar
# in the new request. Because we've mutated our copied prepared
# request, use the old one that we haven't yet touched.
extract_cookies_to_jar(prepared_request._cookies, req, resp.raw)
merge_cookies(prepared_request._cookies, self.cookies)

# Rebuild auth and proxy information.
proxies = self.rebuild_proxies(prepared_request, proxies)
self.rebuild_auth(prepared_request, resp)

# A failed tell() sets `_body_position` to `object()`. This non-None
# value ensures `rewindable` will be True, allowing us to raise an
# UnrewindableBodyError, instead of hanging the connection.
rewindable = (
prepared_request._body_position is not None and
('Content-Length' in headers or 'Transfer-Encoding' in headers)

# Attempt to rewind consumed file-like object.
if rewindable:

# Override the original request.
req = prepared_request

if yield_requests:
yield req

resp = self.send(

extract_cookies_to_jar(self.cookies, prepared_request, resp.raw)

# extract redirect url, if any, for the next loop
url = self.get_redirect_target(resp)
yield resp

def rebuild_auth(self, prepared_request, response):
"""When being redirected we may want to strip authentication from the
request to avoid leaking credentials. This method intelligently removes
and reapplies authentication where possible to avoid credential loss.
headers = prepared_request.headers
url = prepared_request.url

if 'Authorization' in headers:
# If we get redirected to a new host, we should strip out any
# authentication headers.
original_parsed = urlparse(response.request.url)
redirect_parsed = urlparse(url)

if (original_parsed.hostname != redirect_parsed.hostname):
del headers['Authorization']

# .netrc might have more auth for us on our new host.
new_auth = get_netrc_auth(url) if self.trust_env else None
if new_auth is not None:


def rebuild_proxies(self, prepared_request, proxies):
"""This method re-evaluates the proxy configuration by considering the
environment variables. If we are redirected to a URL covered by
NO_PROXY, we strip the proxy configuration. Otherwise, we set missing
proxy keys for this URL (in case they were stripped by a previous

This method also replaces the Proxy-Authorization header where

:rtype: dict
proxies = proxies if proxies is not None else {}
headers = prepared_request.headers
url = prepared_request.url
scheme = urlparse(url).scheme
new_proxies = proxies.copy()
no_proxy = proxies.get('no_proxy')

bypass_proxy = should_bypass_proxies(url, no_proxy=no_proxy)
if self.trust_env and not bypass_proxy:
environ_proxies = get_environ_proxies(url, no_proxy=no_proxy)

proxy = environ_proxies.get(scheme, environ_proxies.get('all'))

if proxy:
new_proxies.setdefault(scheme, proxy)

if 'Proxy-Authorization' in headers:
del headers['Proxy-Authorization']

username, password = get_auth_from_url(new_proxies[scheme])
except KeyError:
username, password = None, None

if username and password:
headers['Proxy-Authorization'] = _basic_auth_str(username, password)

return new_proxies

def rebuild_method(self, prepared_request, response):
"""When being redirected we may want to change the method of the request
based on certain specs or browser behavior.
method = prepared_request.method

if response.status_code == codes.see_other and method != 'HEAD':
method = 'GET'

# Do what the browsers do, despite standards...
# First, turn 302s into GETs.
if response.status_code == codes.found and method != 'HEAD':
method = 'GET'

# Second, if a POST is responded to with a 301, turn it into a GET.
# This bizarre behaviour is explained in Issue 1704.
if response.status_code == codes.moved and method == 'POST':
method = 'GET'

prepared_request.method = method

class Session(SessionRedirectMixin):
"""A Requests session.

Provides cookie persistence, connection-pooling, and configuration.

Basic Usage::

>>> import requests
>>> s = requests.Session()
>>> s.get('')

Or as a context manager::

>>> with requests.Session() as s:
>>> s.get('')


__attrs__ = [
'headers', 'cookies', 'auth', 'proxies', 'hooks', 'params', 'verify',
'cert', 'prefetch', 'adapters', 'stream', 'trust_env',

def __init__(self):

#: A case-insensitive dictionary of headers to be sent on each
#: :class:`Request ` sent from this
#: :class:`Session `.
self.headers = default_headers()

#: Default Authentication tuple or object to attach to
#: :class:`Request `.
self.auth = None

#: Dictionary mapping protocol or protocol and host to the URL of the proxy
#: (e.g. {'http': '', '': ''}) to
#: be used on each :class:`Request `.
self.proxies = {}

#: Event-handling hooks.
self.hooks = default_hooks()

#: Dictionary of querystring data to attach to each
#: :class:`Request `. The dictionary values may be lists for
#: representing multivalued query parameters.
self.params = {}

#: Stream response content default. = False

#: SSL Verification default.
self.verify = True

#: SSL client certificate default, if String, path to ssl client
#: cert file (.pem). If Tuple, ('cert', 'key') pair.
self.cert = None

#: Maximum number of redirects allowed. If the request exceeds this
#: limit, a :class:`TooManyRedirects` exception is raised.
#: This defaults to requests.models.DEFAULT_REDIRECT_LIMIT, which is
#: 30.
self.max_redirects = DEFAULT_REDIRECT_LIMIT

#: Trust environment settings for proxy configuration, default
#: authentication and similar.
self.trust_env = True

#: A CookieJar containing all currently outstanding cookies set on this
#: session. By default it is a
#: :class:`RequestsCookieJar `, but
#: may be any other ``cookielib.CookieJar`` compatible object.
self.cookies = cookiejar_from_dict({})

# Default connection adapters.
self.adapters = OrderedDict()
self.mount('https://', HTTPAdapter())
self.mount('http://', HTTPAdapter())

def __enter__(self):
return self

def __exit__(self, *args):

def prepare_request(self, request):
"""Constructs a :class:`PreparedRequest ` for
transmission and returns it. The :class:`PreparedRequest` has settings
merged from the :class:`Request ` instance and those of the

:param request: :class:`Request` instance to prepare with this
session's settings.
:rtype: requests.PreparedRequest
cookies = request.cookies or {}

# Bootstrap CookieJar.
if not isinstance(cookies, cookielib.CookieJar):
cookies = cookiejar_from_dict(cookies)

# Merge with session cookies
merged_cookies = merge_cookies(
merge_cookies(RequestsCookieJar(), self.cookies), cookies)

# Set environment's basic authentication if not explicitly set.
auth = request.auth
if self.trust_env and not auth and not self.auth:
auth = get_netrc_auth(request.url)

p = PreparedRequest()
headers=merge_setting(request.headers, self.headers, dict_class=CaseInsensitiveDict),
params=merge_setting(request.params, self.params),
auth=merge_setting(auth, self.auth),
hooks=merge_hooks(request.hooks, self.hooks),
return p

def request(self, method, url,
params=None, data=None, headers=None, cookies=None, files=None,
auth=None, timeout=None, allow_redirects=True, proxies=None,
hooks=None, stream=None, verify=None, cert=None, json=None):
"""Constructs a :class:`Request `, prepares it and sends it.
Returns :class:`Response ` object.

:param method: method for the new :class:`Request` object.
:param url: URL for the new :class:`Request` object.
:param params: (optional) Dictionary or bytes to be sent in the query
string for the :class:`Request`.
:param data: (optional) Dictionary, bytes, or file-like object to send
in the body of the :class:`Request`.
:param json: (optional) json to send in the body of the
:param headers: (optional) Dictionary of HTTP Headers to send with the
:param cookies: (optional) Dict or CookieJar object to send with the
:param files: (optional) Dictionary of ``'filename': file-like-objects``
for multipart encoding upload.
:param auth: (optional) Auth tuple or callable to enable
Basic/Digest/Custom HTTP Auth.
:param timeout: (optional) How long to wait for the server to send
data before giving up, as a float, or a :ref:`(connect timeout,
read timeout) ` tuple.
:type timeout: float or tuple
:param allow_redirects: (optional) Set to True by default.
:type allow_redirects: bool
:param proxies: (optional) Dictionary mapping protocol or protocol and
hostname to the URL of the proxy.
:param stream: (optional) whether to immediately download the response
content. Defaults to ``False``.
:param verify: (optional) Either a boolean, in which case it controls whether we verify
the server's TLS certificate, or a string, in which case it must be a path
to a CA bundle to use. Defaults to ``True``.
:param cert: (optional) if String, path to ssl client cert file (.pem).
If Tuple, ('cert', 'key') pair.
:rtype: requests.Response
# Create the Request.
req = Request(
data=data or {},
params=params or {},
prep = self.prepare_request(req)

proxies = proxies or {}

settings = self.merge_environment_settings(
prep.url, proxies, stream, verify, cert

# Send the request.
send_kwargs = {
'timeout': timeout,
'allow_redirects': allow_redirects,
resp = self.send(prep, **send_kwargs)

return resp

def get(self, url, **kwargs):
r"""Sends a GET request. Returns :class:`Response` object.

:param url: URL for the new :class:`Request` object.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response

kwargs.setdefault('allow_redirects', True)
return self.request('GET', url, **kwargs)

def options(self, url, **kwargs):
r"""Sends a OPTIONS request. Returns :class:`Response` object.

:param url: URL for the new :class:`Request` object.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response

kwargs.setdefault('allow_redirects', True)
return self.request('OPTIONS', url, **kwargs)

def head(self, url, **kwargs):
r"""Sends a HEAD request. Returns :class:`Response` object.

:param url: URL for the new :class:`Request` object.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response

kwargs.setdefault('allow_redirects', False)
return self.request('HEAD', url, **kwargs)

def post(self, url, data=None, json=None, **kwargs):
r"""Sends a POST request. Returns :class:`Response` object.

:param url: URL for the new :class:`Request` object.
:param data: (optional) Dictionary, bytes, or file-like object to send in the body of the :class:`Request`.
:param json: (optional) json to send in the body of the :class:`Request`.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response

return self.request('POST', url, data=data, json=json, **kwargs)

def put(self, url, data=None, **kwargs):
r"""Sends a PUT request. Returns :class:`Response` object.

:param url: URL for the new :class:`Request` object.
:param data: (optional) Dictionary, bytes, or file-like object to send in the body of the :class:`Request`.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response

return self.request('PUT', url, data=data, **kwargs)

def patch(self, url, data=None, **kwargs):
r"""Sends a PATCH request. Returns :class:`Response` object.

:param url: URL for the new :class:`Request` object.
:param data: (optional) Dictionary, bytes, or file-like object to send in the body of the :class:`Request`.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response

return self.request('PATCH', url, data=data, **kwargs)

def delete(self, url, **kwargs):
r"""Sends a DELETE request. Returns :class:`Response` object.

:param url: URL for the new :class:`Request` object.
:param \*\*kwargs: Optional arguments that ``request`` takes.
:rtype: requests.Response

return self.request('DELETE', url, **kwargs)

def send(self, request, **kwargs):
"""Send a given PreparedRequest.

:rtype: requests.Response
# Set defaults that the hooks can utilize to ensure they always have
# the correct parameters to reproduce the previous request.
kwargs.setdefault('verify', self.verify)
kwargs.setdefault('cert', self.cert)
kwargs.setdefault('proxies', self.proxies)

# It's possible that users might accidentally send a Request object.
# Guard against that specific failure case.
if isinstance(request, Request):
raise ValueError('You can only send PreparedRequests.')

# Set up variables needed for resolve_redirects and dispatching of hooks
allow_redirects = kwargs.pop('allow_redirects', True)
stream = kwargs.get('stream')
hooks = request.hooks

# Get the appropriate adapter to use
adapter = self.get_adapter(url=request.url)

# Start time (approximately) of the request
start = preferred_clock()

# Send the request
r = adapter.send(request, **kwargs)

# Total elapsed time of the request (approximately)
elapsed = preferred_clock() - start
r.elapsed = timedelta(seconds=elapsed)

# Response manipulation hooks
r = dispatch_hook('response', hooks, r, **kwargs)

# Persist cookies
if r.history:

# If the hooks create history then we want those cookies too
for resp in r.history:
extract_cookies_to_jar(self.cookies, resp.request, resp.raw)

extract_cookies_to_jar(self.cookies, request, r.raw)

# Redirect resolving generator.
gen = self.resolve_redirects(r, request, **kwargs)

# Resolve redirects if allowed.
history = [resp for resp in gen] if allow_redirects else []

# Shuffle things around if there's history.
if history:
# Insert the first (original) request at the start
history.insert(0, r)
# Get the last request made
r = history.pop()
r.history = history

# If redirects aren't being followed, store the response on the Request for
if not allow_redirects:
r._next = next(self.resolve_redirects(r, request, yield_requests=True, **kwargs))
except StopIteration:

if not stream:

return r

def merge_environment_settings(self, url, proxies, stream, verify, cert):
Check the environment and merge it with some settings.

:rtype: dict
# Gather clues from the surrounding environment.
if self.trust_env:
# Set environment's proxies.
no_proxy = proxies.get('no_proxy') if proxies is not None else None
env_proxies = get_environ_proxies(url, no_proxy=no_proxy)
for (k, v) in env_proxies.items():
proxies.setdefault(k, v)

# Look for requests environment configuration and be compatible
# with cURL.
if verify is True or verify is None:
verify = (os.environ.get('REQUESTS_CA_BUNDLE') or

# Merge all the kwargs.
proxies = merge_setting(proxies, self.proxies)
stream = merge_setting(stream,
verify = merge_setting(verify, self.verify)
cert = merge_setting(cert, self.cert)

return {'verify': verify, 'proxies': proxies, 'stream': stream,
'cert': cert}

def get_adapter(self, url):
Returns the appropriate connection adapter for the given URL.

:rtype: requests.adapters.BaseAdapter
for (prefix, adapter) in self.adapters.items():

if url.lower().startswith(prefix):
return adapter

# Nothing matches :-/
raise InvalidSchema("No connection adapters were found for '%s'" % url)

def close(self):
"""Closes all adapters and as such the session"""
for v in self.adapters.values():

def mount(self, prefix, adapter):
"""Registers a connection adapter to a prefix.

Adapters are sorted in descending order by prefix length.
self.adapters[prefix] = adapter
keys_to_move = [k for k in self.adapters if len(k)
for key in keys_to_move:
self.adapters[key] = self.adapters.pop(key)

def __getstate__(self):
state = dict((attr, getattr(self, attr, None)) for attr in self.__attrs__)
return state

def __setstate__(self, state):
for attr, value in state.items():
setattr(self, attr, value)

def session():
Returns a :class:`Session` for context-management.

:rtype: Session

return Session()

This post first appeared on Ricky's Python Notes, please read the originial post: here

Share the post


Subscribe to Ricky's Python Notes

Get updates delivered right to your inbox!

Thank you for your subscription