Additional datafile examples and iterate_results method

Author: Dario Varotto

ravenpackapi/core.py

@@ -13,17 +13,20 @@
 from ravenpackapi.utils.constants import JSON_AVAILABLE_FIELDS
 
 _VALID_METHODS = ('get', 'post', 'put', 'delete')
+VERSION = '1.0.9'
 
 logger = logging.getLogger("ravenpack.core")
 
 
 class RPApi(object):
-    _BASE_URL = os.environ.get('RP_API_ENDPOINT',
-                               'https://api.ravenpack.com/1.0')
-    _FEED_BASE_URL = os.environ.get('RP_FEED_ENDPOINT',
-                                    'https://feed.ravenpack.com/data')
-
     def __init__(self, api_key=None):
+        self._BASE_URL = os.environ.get(
+            'RP_API_ENDPOINT',
+            'https://api.ravenpack.com/1.0')
+        self._FEED_BASE_URL = os.environ.get(
+            'RP_FEED_ENDPOINT',
+            'https://feed.ravenpack.com/1.0/json'
+        )
         api_key = api_key or os.environ.get('RP_API_KEY')
         if api_key is None:
             raise ValueError(
@@ -32,6 +35,12 @@ def __init__(self, api_key=None):
             )
         self.api_key = api_key
 
+    @property
+    def headers(self):
+        return {"API_KEY": self.api_key,
+                'User-Agent': 'RavenPack Python v%s' % VERSION,
+                }
+
     def request(self, endpoint, data=None, params=None, method='get'):
         assert method in _VALID_METHODS, 'Method {used} not accepted. Please use {valid_methods}'
         logger.debug("Request to %s" % endpoint)
@@ -40,7 +49,7 @@ def request(self, endpoint, data=None, params=None, method='get'):
                                                              endpoint=endpoint))
         response = requests_call(
             url=self._BASE_URL + endpoint,
-            headers=dict(API_KEY=self.api_key),
+            headers=self.headers,
             data=json.dumps(data) if data else None,
             params=params,
         )

ravenpackapi/examples/query_datafile_rollout.py

@@ -0,0 +1,38 @@
+import logging
+
+from ravenpackapi import RPApi, Dataset
+from ravenpackapi.models.job import Job
+
+logging.basicConfig(level=logging.INFO)
+# initialize the API (here we use the RP_API_KEY in os.environ)
+api = RPApi()
+
+# query the realtime feed
+eu_600 = api.get_dataset(dataset_id='eu600')  # get the EU600 filters
+
+dataset_id = None  # put here a dataset_id if you have it already
+
+if dataset_id is None:
+    dataset = Dataset(
+        api=api,
+        filters=eu_600.filters,
+        name='EU600 average sentiment',
+        frequency='daily',
+        fields=[{'average_ess': {'avg': {'field': 'EVENT_SENTIMENT_SCORE'}}}]
+    )
+    dataset_id = dataset.save()
+else:
+    dataset = api.get_dataset(dataset_id)
+
+# job = Job(api=api, token='xxx') # if you already have a job you can use this
+# ... or request a new one
+job = dataset.request_datafile(
+    start_date='2018-01-01 00:00:00',
+    end_date='2018-01-02 00:00:00',
+)
+
+# write only the ROLLOUT rows
+for line in job.iterate_results():
+    timestamp, entity_id, entity_name, avg_sentiment = line.split(',')
+    if entity_name == 'ROLLOUT':
+        print(line)

ravenpackapi/models/dataset.py

@@ -92,6 +92,18 @@ def delete(self):
             method='delete',
         )
 
+    @api_method
+    def save(self):
+        response = self.api.request(
+            endpoint="/datasets",
+            data=self.as_dict(),
+            method='post'
+        )
+
+        dataset_id = response.json()['dataset_uuid']
+        logger.info("Created dataset %s" % dataset_id)
+        self._data['uuid'] = dataset_id
+
     @api_method
     def json(self,
              start_date,
@@ -161,7 +173,7 @@ def request_realtime(self):
                                                 dataset_id=self.id)
         logger.debug("Connecting with RT feed: %s" % endpoint)
         r = requests.get(endpoint,
-                         headers=dict(API_KEY=api.api_key),
+                         headers=api.headers,
                          stream=True,
                          )
 

ravenpackapi/models/job.py

@@ -36,6 +36,10 @@ def is_ready(self):
     def is_processing(self):
         return self.status in {'enqueued', 'processing'}
 
+    @property
+    def is_undefined(self):
+        return self.status in {'unknown'}
+
     def __str__(self):
         return "Job {status}: {token}".format(status=self.status,
                                               token=self.token)
@@ -57,9 +61,14 @@ def wait_for_completion(self, timeout_seconds=None):
         max_end_date = datetime.datetime.utcnow() + datetime.timedelta(
             seconds=timeout_seconds
         ) if timeout_seconds else None
+
+        if self.is_undefined:
+            self.get_status()
+
         while True:
             if self.is_ready:
                 break
+            sleep(self._FILE_AVAILABILIY_SECONDS_DELAY)
             try:
                 self.get_status()
             except APIException:  # keep waiting if API raises exceptions
@@ -72,7 +81,6 @@ def wait_for_completion(self, timeout_seconds=None):
             if not printed_once:
                 logger.info("Waiting for the job to be ready...")
                 printed_once = True
-            sleep(self._FILE_AVAILABILIY_SECONDS_DELAY)
 
     @api_method
     def save_to_file(self, filename):
@@ -85,10 +93,27 @@ def save_to_file(self, filename):
             # this is a different request than the normal API
             # streaming the file in chunks
             r = requests.get(job.url,
-                             headers=dict(API_KEY=api.api_key),
+                             headers=api.headers,
                              stream=True,
                              )
 
             for chunk in r.iter_content(chunk_size=self._CHUNK_SIZE):
                 if chunk:
                     output.write(chunk)
+
+    @api_method
+    def iterate_results(self):
+        api = self.api
+        job = self  # just to be clear
+        job.wait_for_completion()
+
+        with requests.Session() as s:
+            r = s.get(job.url,
+                      headers=api.headers,
+                      stream=True,
+                      )
+            iterator = r.iter_lines(chunk_size=self._CHUNK_SIZE)
+            header = next(iterator)  # discard the header
+
+            for line in iterator:
+                yield line.decode('utf-8')

setup.py

@@ -1,11 +1,13 @@
 from setuptools import setup, find_packages
 
+from ravenpackapi.core import VERSION
+
 with open('README.rst') as readme_file:
     readme = readme_file.read()
 
 setup(
     name='ravenpackapi',
-    version='1.0.8',
+    version=VERSION,
     packages=find_packages(include=['ravenpackapi']),
     include_package_data=True,