Bind request sessions to an IP address.

create_bound_session.py

import requests
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.connection import HTTPConnection, HTTPSConnection
from requests.packages.urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool
from requests.packages.urllib3.poolmanager import PoolManager


def create_bound_session(source_address):
    '''
    Creates a requests session bound to a given source address. Proving once
    again that class-based composition sucks.
    '''
    class BoundHTTPConnection(HTTPConnection, object):
        def __init__(self, *args, **kwargs):
            super(BoundHTTPConnection, self).__init__(*args, **kwargs)
            self.source_address = (source_address, 0)

    class BoundHTTPSConnection(HTTPSConnection, object):
        def __init__(self, *args, **kwargs):
            super(BoundHTTPSConnection, self).__init__(*args, **kwargs)
            self.source_address = (source_address, 0)

    class BoundHTTPConnectionPool(HTTPConnectionPool):
        ConnectionCls = BoundHTTPConnection

    class BoundHTTPSConnectionPool(HTTPSConnectionPool):
        ConnectionCls = BoundHTTPSConnection

    pool_classes_by_scheme = {
        'http': BoundHTTPConnectionPool,
        'https': BoundHTTPSConnectionPool,
    }

    SSL_KEYWORDS = ('key_file', 'cert_file', 'cert_reqs', 'ca_certs',
                    'ssl_version')

    class BoundManager(PoolManager):
        def _new_pool(self, scheme, host, port):
            pool_cls = pool_classes_by_scheme[scheme]
            kwargs = self.connection_pool_kw
            if scheme == 'http':
                kwargs = self.connection_pool_kw.copy()
                for kw in SSL_KEYWORDS:
                    kwargs.pop(kw, None)

            return pool_cls(host, port, **kwargs)

    class BoundAdapter(HTTPAdapter):
        def init_poolmanager(self, connections, maxsize, block=True):
            self._pool_connections = connections
            self._pool_maxsize = maxsize
            self._pool_block = block
            self.poolmanager = BoundManager(num_pools=connections, maxsize=maxsize, block=block)

    s = requests.Session()
    s.mount('http://', BoundAdapter())
    return s