Add ability to use http/https proxies with iland-sdk

Author: corywright

docs/usage.rst

@@ -27,3 +27,14 @@ cloud::
                              'username': USERNAME}
     >>> api.post('/user/' + USERNAME + '/alert-emails', user_alert_emails)
 
+To use a proxy you can define a ``python-requests`` style proxies_ dictionary
+and set the ``_proxies`` attribute on the api object::
+
+    >>> api = iland.Api(client_id=CLIENT_ID,
+                        client_secret=CLIENT_SECRET,
+                        username=USERNAME,
+                        password=PASSWORD)
+    >>> api._proxies = {'https': 'https://10.10.1.10:3128'}
+    >>> user = api.get('/user/' + USERNAME)
+
+.. _proxies: http://docs.python-requests.org/en/master/user/advanced/#proxies

iland/api.py

@@ -23,6 +23,7 @@ class Api(object):
     _base_url = BASE_URL
     _access_token_url = ACCESS_URL
     _refresh_token_url = REFRESH_URL
+    _proxies = None
 
     _token = None
     _token_expiration_time = None
@@ -128,18 +129,26 @@ def _do_request(self, rpath, verb='GET', form_data=None):
         headers = {
             'Authorization': 'Bearer %s' % self._get_access_token_string(),
             'content-type': 'application/json'}
+
+        request_params = {
+            'headers': headers,
+            'verify':  self._verify_ssl
+        }
+
+        if verb in ('PUT', 'POST'):
+            request_params['data'] = data
+
+        if self._proxies and isinstance(self._proxies, dict):
+            request_params['proxies'] = self._proxies
+
         if verb == 'GET':
-            r = self._session.get(url, headers=headers,
-                                  verify=self._verify_ssl)
+            r = self._session.get(url, **request_params)
         elif verb == 'PUT':
-            r = self._session.put(url, data=data, headers=headers,
-                                  verify=self._verify_ssl)
+            r = self._session.put(url, **request_params)
         elif verb == 'POST':
-            r = self._session.post(url, data=data, headers=headers,
-                                   verify=self._verify_ssl)
+            r = self._session.post(url, **request_params)
         elif verb == 'DELETE':
-            r = self._session.delete(url, headers=headers,
-                                     verify=self._verify_ssl)
+            r = self._session.delete(url, **request_params)
         else:
             raise ApiException({'message': 'Unsupported HTTP verb %s' % verb})
 

tests/apicreds.py.example

@@ -4,3 +4,4 @@ CLIENT_ID = ''
 CLIENT_SECRET = ''
 USERNAME = ''
 PASSWORD = ''
+PROXIES = {}

tests/test_iland_int.py

@@ -20,18 +20,21 @@
     from apicreds import (CLIENT_ID,
                           CLIENT_SECRET,
                           USERNAME,
-                          PASSWORD)
+                          PASSWORD,
+                          PROXIES)
 except ImportError:
     try:
         from .apicreds import (CLIENT_ID,
                                CLIENT_SECRET,
                                USERNAME,
-                               PASSWORD)
+                               PASSWORD,
+                               PROXIES)
     except ImportError:
         CLIENT_ID = None
         CLIENT_SECRET = None
         USERNAME = None
         PASSWORD = None
+        PROXIES = {}
 
 VDC_UUID = \
     'res01.ilandcloud.com:urn:vcloud:vdc:a066325d-6be0-4733-8d9f-7687c36f4536'
@@ -131,3 +134,10 @@ def test_unauthorized_errors(self):
     def test_api_errors(self):
         with self.assertRaises(ApiException):
             self._api.get('/doesnotexist')
+
+    @unittest.skipIf(not PROXIES, "No proxies defined")
+    def test_get_with_proxy(self):
+        self._api._proxies = PROXIES
+        user = self._api.get('/user/' + USERNAME)
+        self.assertEqual(USERNAME, user.get('name'))
+        self.assertTrue(len(user.keys()) > 5)