Add proxy support for SSE transport

requirements-test.txt

@@ -0,0 +1,3 @@
+-r requirements.txt
+
+httmock

requirements.txt

@@ -1,5 +1,4 @@
 certifi
-httmock
 protobuf>=3.0.0
 pyformance>=0.3.1
 requests>=2.7.0

setup.py

@@ -26,8 +26,8 @@
     long_description=long_description,
     zip_safe=True,
     packages=find_packages(),
-    test_requires=test_requirements,
     install_requires=requirements,
+    tests_require=test_requirements,
     classifiers=[
         'Operating System :: OS Independent',
         'Programming Language :: Python',

signalfx/signalflow/__init__.py

@@ -16,8 +16,9 @@ class SignalFlowClient(object):
     def __init__(self, token, endpoint=constants.DEFAULT_STREAM_ENDPOINT,
                  timeout=constants.DEFAULT_TIMEOUT,
                  transport=ws.WebSocketTransport,
-                 compress=True):
-        self._transport = transport(token, endpoint, timeout, compress)
+                 compress=True, proxy_url=None):
+        self._transport = transport(token, endpoint, timeout, compress,
+                                    proxy_url)
         self._computations = set([])
 
     def __enter__(self):

signalfx/signalflow/errors.py

@@ -5,9 +5,10 @@ class SignalFlowException(Exception):
     """A generic error encountered when interacting with the SignalFx
     SignalFlow API."""
 
-    def __init__(self, code, message=None):
+    def __init__(self, code, message=None, error_type=None):
         self._code = code
         self._message = message
+        self._error_type = error_type
 
     @property
     def code(self):
@@ -19,10 +20,19 @@ def message(self):
         """Returns an optional error message attached to this error."""
         return self._message
 
+    @property
+    def error_type(self):
+        """Returns an optional error type attached to this error."""
+        return self._error_type
+
     def __str__(self):
+        err = self._code
+        if self._error_type:
+            err = '{0} ({1})'.format(self._code, self._error_type)
+
         if self._message:
-            return '{0}: {1}'.format(self._code, self._message)
-        return 'Error {0}'.format(self._code)
+            return '{0}: {1}'.format(err, self._message)
+        return 'Error {0}'.format(err)
 
 
 class ComputationAborted(Exception):

signalfx/signalflow/sse.py

@@ -24,7 +24,8 @@ class SSETransport(transport._SignalFlowTransport):
     _SIGNALFLOW_ENDPOINT = 'v2/signalflow'
 
     def __init__(self, token, endpoint=constants.DEFAULT_STREAM_ENDPOINT,
-                 timeout=constants.DEFAULT_TIMEOUT, compress=True):
+                 timeout=constants.DEFAULT_TIMEOUT, compress=True,
+                 proxy_url=None):
         super(SSETransport, self).__init__(token, endpoint, timeout)
         pool_args = {
             'url': self._endpoint,
@@ -43,7 +44,12 @@ def __init__(self, token, endpoint=constants.DEFAULT_STREAM_ENDPOINT,
                 'ca_certs': certifi.where()    # Path to the Certifi bundle.
             })
 
-        self._http = urllib3.connectionpool.connection_from_url(**pool_args)
+        if proxy_url:
+            proxy_manager = urllib3.poolmanager.proxy_from_url(proxy_url)
+            endpoint = pool_args.pop('url')
+            self._http = proxy_manager.connection_from_url(endpoint, pool_kwargs=pool_args)
+        else:
+            self._http = urllib3.connectionpool.connection_from_url(**pool_args)
 
     def __str__(self):
         return 'sse+{0}'.format(self._endpoint)
@@ -58,7 +64,8 @@ def _post(self, url, fields=None, body=None):
         if r.status != 200:
             try:
                 if r.headers['Content-Type'] == 'application/json':
-                    raise errors.SignalFlowException(**json.loads(r.read()))
+                    rbody = json.loads(r.read())
+                    raise errors.SignalFlowException(r.status, rbody.get('message'), rbody.get('errorType'))
                 raise errors.SignalFlowException(r.status)
             finally:
                 r.close()

signalfx/signalflow/ws.py

@@ -29,7 +29,11 @@ class WebSocketTransport(transport._SignalFlowTransport, WebSocketClient):
     _SIGNALFLOW_WEBSOCKET_ENDPOINT = 'v2/signalflow/connect'
 
     def __init__(self, token, endpoint=constants.DEFAULT_STREAM_ENDPOINT,
-                 timeout=constants.DEFAULT_TIMEOUT, compress=True):
+                 timeout=constants.DEFAULT_TIMEOUT, compress=True,
+                 proxy_url=None):
+        if proxy_url:
+            raise NotImplementedError('Websocket transport cannot be proxied!')
+
         ws_endpoint = '{0}/{1}'.format(
             endpoint.replace('http', 'ws', 1),
             WebSocketTransport._SIGNALFLOW_WEBSOCKET_ENDPOINT)

tox.ini

@@ -8,7 +8,7 @@ envlist = py27,py35,flake8
 [testenv]
 usedevelop = True
 passenv = ORG_NAME_FOR_SIGNALFX_API_TOKEN_IS_SFx_Test SIGNALFX_API_TOKEN
-deps = -r{toxinidir}/requirements.txt
+deps = -r{toxinidir}/requirements-test.txt
 ignore_errors = True
 commands = python {toxinidir}/tests/test_data_reporting.py
            python {toxinidir}/tests/live_tests.py --metric_name 'MET' --tag_name 'TG' --key 'K' --value 'V'