Version 1.0.27: Example retry logic for real-time connections

Author: dariosky

README.md

@@ -94,17 +94,17 @@ with open('output.csv') as fp:
 
 ### Streaming real-time data
 
-It is possible to subscribe to a real-time stream for a dataset:
+It is possible to subscribe to a real-time stream for a dataset.
 
-```python
-ds = api.get_dataset(dataset_id='us500')
-for record in ds.request_realtime():
-	print(record)
-	print(record.timestamp_utc, record.entity_name,
-              record['event_relevance'])
-```
+Once you create a streaming connection to the real-time feed with your dataset,
+you will receive analytics records as soon as they are published.
+
+
+It is suggested to handle possible disconnection with a retry policy.
+You can find a [real-time streaming example here](ravenpackapi/examples/get_realtime_news.py).
 
-The Result object takes care of converting the various fields to the appropriate type, so `record.timestamp_utc` will be a `datetime`
+The Result object handles the conversion of various fields into the appropriate type, 
+i.e. `record.timestamp_utc` will be converted to `datetime`
 
 ### Entity mapping
 
@@ -157,3 +157,24 @@ for ticker in references.tickers:
     if ticker.is_valid():
         print(ticker)
 ```
+
+### Accessing the low-level requests
+
+RavenPack API wrapper is using the [requests library](https://2.python-requests.org) to do HTTPS requests,
+you can set common requests parameters to all the outbound calls by setting the `common_request_params` attribute.
+
+For example, to disable HTTPS certificate verification and to setup your internal proxy:
+
+```python
+api = RPApi()
+api.common_request_params.update(
+	dict(
+		proxies={'https': 'http://your_internal_proxy:9999'},
+		verify=False,
+	)
+)
+
+# use the api to do requests
+```
+
+PS. For setting your internal proxies, requests will honor the HTTPS_PROXY environment variable.

ravenpackapi/__init__.py

@@ -1,2 +1,3 @@
 from .models.dataset import Dataset
 from .core import RPApi
+from .exceptions import ApiConnectionError

ravenpackapi/core.py

@@ -14,7 +14,7 @@
 from ravenpackapi.utils.constants import JSON_AVAILABLE_FIELDS, ENTITY_TYPES
 
 _VALID_METHODS = ('get', 'post', 'put', 'delete')
-VERSION = '1.0.26'
+VERSION = '1.0.27'
 
 logger = logging.getLogger("ravenpack.core")
 

ravenpackapi/examples/get_realtime_news.py

@@ -1,15 +1,38 @@
 import logging
+import random
+import time
 
-from ravenpackapi import RPApi
+from ravenpackapi import RPApi, ApiConnectionError
 
 logging.basicConfig(level=logging.DEBUG)
+logger = logging.getLogger(__name__)
+
 # initialize the API (here we use the RP_API_KEY in os.environ)
 api = RPApi()
+api.common_request_params['timeout'] = (10, None)  # set a timeout on connection
 
 # query the realtime feed
 ds = api.get_dataset(dataset_id='us500')
 
-for record in ds.request_realtime():
-    print(record)
-    print(record.timestamp_utc, record.entity_name,
-          record['event_relevance'])
+
+def wait_between_attempts():
+    """ Incremental backoff between connection attempts """
+    wait_time = 0.3  # time is in seconds
+    while True:
+        yield wait_time
+        wait_time = min(wait_time * 1.5, 30)
+        wait_time *= (100 + random.randint(0, 50)) / 100
+
+
+wait_time = wait_between_attempts()
+while True:
+    try:
+        for record in ds.request_realtime():
+            print(record)
+            print(record.timestamp_utc, record.entity_name,
+                  record['event_relevance'])
+    except (KeyboardInterrupt, SystemExit):
+        break
+    except ApiConnectionError as e:
+        logger.error("Connection error %s: reconnecting..." % e)
+        time.sleep(next(wait_time))

ravenpackapi/exceptions.py

@@ -1,5 +1,7 @@
 from functools import wraps
 
+import requests
+
 
 class APIException(Exception):
     def __init__(self, *args, **kwargs):
@@ -37,3 +39,7 @@ def decorated_func(instance, *args, **kwargs):
 
 class ValidationError(Exception):
     pass
+
+
+class ApiConnectionError(requests.ConnectionError):
+    pass

ravenpackapi/models/dataset.py

@@ -224,8 +224,8 @@ def request_datafile(self, start_date, end_date,
     @api_method
     def request_realtime(self):
         api = self.api
-        endpoint = "{base}/{dataset_id}".format(base=api._FEED_BASE_URL,
-                                                dataset_id=self.id)
+        endpoint = "{base}/{dataset_id}?keep_alive".format(base=api._FEED_BASE_URL,
+                                                           dataset_id=self.id)
         logger.debug("Connecting with RT feed: %s" % endpoint)
         response = requests.get(endpoint,
                                 headers=api.headers,
@@ -241,4 +241,5 @@ def request_realtime(self):
         response.encoding = 'utf-8'
 
         for line in response.iter_lines(decode_unicode=True, chunk_size=1):
-            yield Result(line)
+            if line:  # skip empty lines to support keep-alive
+                yield Result(line)

setup.py

@@ -1,6 +1,6 @@
 from setuptools import setup, find_packages
 
-VERSION = '1.0.26'
+VERSION = '1.0.27'
 
 with open('README.rst') as readme_file:
     readme = readme_file.read()