3

I'm hitting an API in python through requests' Session class. I'm making GET & POST method call using requests.Session().

On every call(GET/POST) failure, I want to notify another process. I can do this by creating a utility method as follows:

s = request.Session()
def post():
    try:
        s.post(URL,data,headers)
    except:
        notify_another_process()

And call this method instead of requests.Session().post directly.

But, I want to monkeypatch this code to requests.Session().post and want the additional functionality of notifying the other process in the requests.Session().post method call itself. How can I achieve this?

EDIT 1 :

requests.Session()'s post method has the following signature:

def post(self, url, data=None, json=None, **kwargs):
    return self.request('POST', url, data=data, json=json, **kwargs)

If I somehow try to make my a custom post like the following:

def post_new(self, url, data=None, json=None, **kwargs):
    try:
        s.post(url,data, json,kwargs)
    except:
        notify_another_process()

and do a patch as follows:

requests.post = post_new

This isn't really a good monkeypatching because I'm not using self but session's object inside session.post.

Pankaj Singhal
  • 15,283
  • 9
  • 47
  • 86
  • Possible duplicate of [How does one monkey patch a function in python?](http://stackoverflow.com/questions/2375403/how-does-one-monkey-patch-a-function-in-python) – Ilja Everilä Oct 04 '16 at 10:50
  • [Requests `post` method signature is different than one you've posted](https://github.com/kennethreitz/requests/blob/master/requests/api.py#L99) – Łukasz Rogalski Oct 04 '16 at 11:12
  • @ŁukaszRogalski I'm trying to use [sessions post method](https://github.com/kennethreitz/requests/blob/master/requests/sessions.py#L513). – Pankaj Singhal Oct 04 '16 at 11:39
  • @PankajSinghal do why your code snippets does not mention creating `Session()` at all? Learn more on creating [MCVE]. – Łukasz Rogalski Oct 04 '16 at 11:41
  • @ŁukaszRogalski Sorry for creating confusion. I've updated the question. Thanks for correcting – Pankaj Singhal Oct 04 '16 at 12:01

3 Answers3

2

This should resolve this. You basically save the old function with different name and give your function as the default post.

setattr(requests, 'old_post', requests.post)

def post(url, data=None, json=None, **kwargs):
    try:
        requests.old_post(url, data, json, kwargs)
    except:
        notify_another_process()

setattr(requests, 'post', post)
Siddharth Gupta
  • 1,573
  • 9
  • 24
  • I've updated the question. I missed mentioning about Session. Also, though your answer would be correct in case of requests but I'm going to post the answer which worked for me for the sake of clarity of the question & it's answer for the future audience(if it helps anybody in future). Thanks for answering. :) – Pankaj Singhal Oct 04 '16 at 12:21
2

You're almost there, but you should use the self argument

def post_new(self, url, data=None, json=None, **kwargs):
    try:
        return self.request('POST', url, data=data, json=json, **kwargs)
    except:
        notify_another_process()

Then set the post function to post_new

requests.post = post_new
lafferc
  • 2,741
  • 3
  • 24
  • 37
  • I've updated the question. I missed mentioning about Session. Also, though your answer would be correct in case of requests but I'm going to post the answer which worked for me for the sake of clarity of the question & it's answer for the future audience(if it helps anybody in future). – Pankaj Singhal Oct 04 '16 at 12:17
  • Thanks for answering. :) – Pankaj Singhal Oct 04 '16 at 12:21
1

This is the answer which worked for me. It is inspired by the answers mentioned by Siddharth & lafferc both. This is on top of what both of them mentioned.

>>> import requests
>>> def post(self, url, data=None, json=None, **kwargs):
...     try:
...         raise Exception()
...     except:
...         print "notifying another process"
... 
>>> setattr(requests.Session, 'post_old', requests.Session.post)
>>> setattr(requests.Session, 'post', post)
>>> s = requests.Session()
>>> s.post("url")
notifying another process
Community
  • 1
  • 1
Pankaj Singhal
  • 15,283
  • 9
  • 47
  • 86