Last active
December 19, 2015 04:19
-
-
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…
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
# 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