Source code for ahoyhoy.http.proxy

import logging
from functools import wraps

import requests

logger = logging.getLogger(__name__)


[docs]class SessionProxy(object): """ Proxy adapter so a subclass can proxy methods of a Requests Session, but can alter behavior via pre, post, and exception callbacks. Can be used by itself, but is intended to be subclassed with overwritten callbacks. >>> from ahoyhoy.http import SessionProxy >>> sp = SessionProxy() >>> sp.get('http://google.com') <Response [200]> In order to be a bit more transparent, excpetion_callback will raise the thrown exception. >>> from ahoyhoy.http import SessionProxy >>> import requests >>> sp = SessionProxy(requests.Session()) >>> try: ... sp.get('http://wwwwww.google.com') ... except requests.exceptions.ConnectionError as e: ... print("Error raised") ... Error raised You may override the pre|post|exception callbacks either by subclassing them or by runtime configuration. >>> from ahoyhoy.http import SessionProxy >>> import requests >>> >>> def pre_callback(url): ... print('pre') ... return url >>> >>> def post_callback(res): ... print('post') >>> >>> def exception_callback(e): ... print('Exception!!') >>> >>> sp = SessionProxy(requests.Session(), pre_callback=pre_callback, post_callback=post_callback, exception_callback=exception_callback) >>> sp.get('http://google.com') pre post >>> sp.get('http://google1.com') pre Exception!! post Test proxy for other methods and attributes: >>> sp.headers {...'User-Agent': ...} """ HTTP_CALLS = ( 'get', 'options', 'post', 'put', 'head', 'patch', 'delete', ) def __init__(self, session=None, pre_callback=None, post_callback=None, exception_callback=None): """ :param session: custom session :param pre_callback: executed before the proxied method :param post_callback: executed after the proxied method :param exception_callback: the proxied method is wrapped in a try/except block, and when an exception occurs, the exception is passed to this callback """ if session is not None: self._session = session else: self._session = requests.Session() self._pre = pre_callback self._post = post_callback self._exc = exception_callback def __getattr__(self, name): """ Proxy Requests methods to self._session, but first calculate the correct load balanced url to use via `pre_callback`. """ realfunc = getattr(self._session, name) # if the attribute is HTTP call - use callbacks if name in self.HTTP_CALLS: @wraps(realfunc) def func(*args, **kwargs): result = None # first positional param is always url url = self.pre_callback(args[0]) try: result = realfunc(url, **kwargs) # matches Requests API except Exception as e: self.exception_callback(e) return self.post_callback(result) return func # if the attribute is NOT HTTP call - return w/ arguments elif callable(realfunc): @wraps(realfunc) def func(*args, **kwargs): return realfunc(*args, **kwargs) return func # if the attribute is NOT callable - return w/o arguments else: return realfunc
[docs] def pre_callback(self, urlpath): """ Executed before the proxied Requests method. Intended to be overwritten or set by a derived class. :param urlpath: the arg[0] of the called proxied method. Requests Session methods all take 'url' as the first positional parameter. By default, returns what's passed in. """ logger.debug("Entered pre-callback") if self._pre is None: return urlpath return self._pre(urlpath)
[docs] def post_callback(self, result): """ Executed after the proxied Requests method. Intended to be overwritten or set by a derived class. :param result: the return value of the proxied Requests method. By default, returns what's passed in. """ logger.debug("Entered post-callback") if self._post is None: return result return self._post(result)
[docs] def exception_callback(self, exc): """ Executed when an exception is thrown by the proxied Requests Session method. Intended to be overwritten or set by a derived class. :param exc: the exception raised by the proxied Requests Session method. By default, returns what's passed in. """ logger.debug("Entered exception callback") if self._exc is None: raise exc return self._exc(exc)
def __dir__(self): return list(self.__dict__.keys()) + dir(self.__class__) + requests.Session().__dir__