Skip to content

Instantly share code, notes, and snippets.

@joepie91
Last active December 19, 2015 04:19
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save joepie91/5896273 to your computer and use it in GitHub Desktop.
Save joepie91/5896273 to your computer and use it in GitHub Desktop.
READ ALL OF THIS, IT'S IMPORTANT: This is how you monkeypatch support for binding to a specific IP into Python Requests. You use the new bound_ip keyword argument when creating a new session, to assign a certain IP to that Requests session. All subsequent HTTP requests made from that session via the standard methods, will be bound to that IP. No…
# I needed this for my nzbspider project (https://github.com/joepie91/nzbspider). Decided to throw up a stand-alone version.
import requests, socket
# Very nasty monkeypatching ahead!
socket.real_create_connection = socket.create_connection
class ModifiedSession(requests.Session):
def __init__(self, *args, **kwargs):
try:
self.bound_ip = kwargs['bound_ip']
del kwargs['bound_ip']
except KeyError, e:
self.bound_ip = ""
requests.Session.__init__(self, *args, **kwargs)
def patch_socket(self):
socket.create_connection = get_patched_func(self.bound_ip)
def unpatch_socket(self):
socket.create_connection = socket.real_create_connection
def get(self, *args, **kwargs):
self.patch_socket()
response = requests.Session.get(self, *args, **kwargs)
self.unpatch_socket()
return response
def post(self, *args, **kwargs):
self.patch_socket()
response = requests.Session.post(self, *args, **kwargs)
self.unpatch_socket()
return response
def head(self, *args, **kwargs):
self.patch_socket()
response = requests.Session.head(self, *args, **kwargs)
self.unpatch_socket()
return response
def put(self, *args, **kwargs):
self.patch_socket()
response = requests.Session.put(self, *args, **kwargs)
self.unpatch_socket()
return response
def patch(self, *args, **kwargs):
self.patch_socket()
response = requests.Session.patch(self, *args, **kwargs)
self.unpatch_socket()
return response
def delete(self, *args, **kwargs):
self.patch_socket()
response = requests.Session.delete(self, *args, **kwargs)
self.unpatch_socket()
return response
def get_patched_func(bind_addr):
def set_src_addr(*args):
address, timeout = args[0], args[1]
source_address = (bind_addr, 0)
return socket.real_create_connection(address, timeout, source_address)
return set_src_addr
# You're looking at duct tape and tie-wraps. It's like your local Home
# Depot, except in Python.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment