JSONResultsReader changes

Author: ashah-splunk

examples/follow.py

@@ -42,7 +42,7 @@ def follow(job, count, items):
             job.refresh()
             continue
         stream = items(offset+1)
-        for event in results.ResultsReader(stream):
+        for event in results.JSONResultsReader(stream):
             pprint(event)
         offset = total
 
@@ -72,10 +72,10 @@ def main():
 
     if job['reportSearch'] is not None: # Is it a transforming search?
         count = lambda: int(job['numPreviews'])
-        items = lambda _: job.preview()
+        items = lambda _: job.preview(output_mode='json')
     else:
         count = lambda: int(job['eventCount'])
-        items = lambda offset: job.events(offset=offset)
+        items = lambda offset: job.events(offset=offset, output_mode='json')
 
     try:
         follow(job, count, items)

examples/oneshot.py

@@ -32,7 +32,7 @@
                     "(e.g., export PYTHONPATH=~/splunk-sdk-python.")
 
 def pretty(response):
-    reader = results.ResultsReader(response)
+    reader = results.JSONResultsReader(response)
     for result in reader:
         if isinstance(result, dict):
             pprint(result)
@@ -46,7 +46,7 @@ def main():
     search = opts.args[0]
     service = connect(**opts.kwargs)
     socket.setdefaulttimeout(None)
-    response = service.jobs.oneshot(search)
+    response = service.jobs.oneshot(search, output_mode='json')
 
     pretty(response)
 

examples/search_modes.py

@@ -24,7 +24,7 @@ def modes(argv):
     while not job.is_ready():
         time.sleep(0.5)
         pass    
-    reader = results.ResultsReader(job.events())
+    reader = results.JSONResultsReader(job.events(output_mode='json'))
     # Events found: 0
     print('Events found with adhoc_search_level="smart": %s' % len([e for e in reader]))
 
@@ -33,7 +33,7 @@ def modes(argv):
     while not job.is_ready():
         time.sleep(0.5)
         pass
-    reader = results.ResultsReader(job.events())
+    reader = results.ResultsReader(job.events(output_mode='json'))
     # Events found: 10
     print('Events found with adhoc_search_level="verbose": %s' % len([e for e in reader]))
 

examples/stail.py

@@ -25,7 +25,7 @@
 from pprint import pprint
 
 from splunklib.client import connect
-from splunklib.results import ResultsReader
+from splunklib.results import ResultsReader, JSONResultsReader
 
 try:
     import utils
@@ -49,9 +49,10 @@ def main():
             search=search,
             earliest_time="rt", 
             latest_time="rt", 
-            search_mode="realtime")
+            search_mode="realtime",
+            output_mode="json")
 
-        for result in ResultsReader(result.body):
+        for result in JSONResultsReader(result.body):
             if result is not None:
                 print(pprint(result))
 

splunklib/results.py

@@ -34,7 +34,7 @@
 
 from __future__ import absolute_import
 
-from io import BytesIO
+from io import BufferedReader, BytesIO
 
 from splunklib import six
 try:
@@ -43,6 +43,7 @@
     import xml.etree.ElementTree as et
 
 from collections import OrderedDict
+from json import loads as json_loads
 
 try:
     from splunklib.six.moves import cStringIO as StringIO
@@ -287,6 +288,68 @@ def __itertext(self):
             else:
                 raise
 
+class JSONResultsReader(object):
+    """This class returns dictionaries and Splunk messages from a JSON results
+    stream.
+    ``JSONResultsReader`` is iterable, and returns a ``dict`` for results, or a
+    :class:`Message` object for Splunk messages. This class has one field,
+    ``is_preview``, which is ``True`` when the results are a preview from a
+    running search, or ``False`` when the results are from a completed search.
+    This function has no network activity other than what is implicit in the
+    stream it operates on.
+    :param `stream`: The stream to read from (any object that supports
+        ``.read()``).
+    **Example**::
+        import results
+        response = ... # the body of an HTTP response
+        reader = results.JSONResultsReader(response)
+        for result in reader:
+            if isinstance(result, dict):
+                print "Result: %s" % result
+            elif isinstance(result, results.Message):
+                print "Message: %s" % result
+        print "is_preview = %s " % reader.is_preview
+    """
+    # Be sure to update the docstrings of client.Jobs.oneshot,
+    # client.Job.results_preview and client.Job.results to match any
+    # changes made to JSONResultsReader.
+    #
+    # This wouldn't be a class, just the _parse_results function below,
+    # except that you cannot get the current generator inside the
+    # function creating that generator. Thus it's all wrapped up for
+    # the sake of one field.
+    def __init__(self, stream):
+        # The search/jobs/exports endpoint, when run with
+        # earliest_time=rt and latest_time=rt, output_mode=json, streams a sequence of
+        # JSON documents, each containing a result, as opposed to one
+        # results element containing lots of results.
+        stream = BufferedReader(stream)
+        self.is_preview = None
+        self._gen = self._parse_results(stream)
 
+    def __iter__(self):
+        return self
 
+    def next(self):
+        return next(self._gen)
 
+    __next__ = next
+
+    def _parse_results(self, stream):
+        """Parse results and messages out of *stream*."""
+        for line in stream.readlines():
+            strip_line = line.strip()
+            if strip_line.__len__() == 0 : continue
+            parsed_line = json_loads(strip_line)
+            if "preview" in parsed_line:
+                self.is_preview = parsed_line["preview"]
+            if "messages" in parsed_line and parsed_line["messages"].__len__() > 0:
+                for message in parsed_line["messages"]:
+                    msg_type = message.get("type", "Unknown Message Type")
+                    text = message.get("text")
+                yield Message(msg_type, text)
+            if "result" in parsed_line:
+                yield parsed_line["result"]
+            if "results" in parsed_line:
+                for result in parsed_line["results"]:
+                    yield result

tests/test_job.py

@@ -54,8 +54,8 @@ def test_oneshot_with_garbage_fails(self):
 
     def test_oneshot(self):
         jobs = self.service.jobs
-        stream = jobs.oneshot("search index=_internal earliest=-1m | head 3")
-        result = results.ResultsReader(stream)
+        stream = jobs.oneshot("search index=_internal earliest=-1m | head 3", output_mode='json')
+        result = results.JSONResultsReader(stream)
         ds = list(result)
         self.assertEqual(result.is_preview, False)
         self.assertTrue(isinstance(ds[0], dict) or \
@@ -69,8 +69,8 @@ def test_export_with_garbage_fails(self):
 
     def test_export(self):
         jobs = self.service.jobs
-        stream = jobs.export("search index=_internal earliest=-1m | head 3")
-        result = results.ResultsReader(stream)
+        stream = jobs.export("search index=_internal earliest=-1m | head 3", output_mode='json')
+        result = results.JSONResultsReader(stream)
         ds = list(result)
         self.assertEqual(result.is_preview, False)
         self.assertTrue(isinstance(ds[0], dict) or \
@@ -82,7 +82,7 @@ def test_export_docstring_sample(self):
         import splunklib.client as client
         import splunklib.results as results
         service = self.service # cheat
-        rr = results.ResultsReader(service.jobs.export("search * | head 5"))
+        rr = results.JSONResultsReader(service.jobs.export("search * | head 5", output_mode='json'))
         for result in rr:
             if isinstance(result, results.Message):
                 # Diagnostic messages may be returned in the results
@@ -98,7 +98,7 @@ def test_results_docstring_sample(self):
         job = service.jobs.create("search * | head 5")
         while not job.is_done():
             sleep(0.2)
-        rr = results.ResultsReader(job.results())
+        rr = results.JSONResultsReader(job.results(output_mode='json'))
         for result in rr:
             if isinstance(result, results.Message):
                 # Diagnostic messages may be returned in the results
@@ -113,7 +113,7 @@ def test_preview_docstring_sample(self):
         import splunklib.results as results
         service = self.service # cheat
         job = service.jobs.create("search * | head 5")
-        rr = results.ResultsReader(job.preview())
+        rr = results.JSONResultsReader(job.preview(output_mode='json'))
         for result in rr:
             if isinstance(result, results.Message):
                 # Diagnostic messages may be returned in the results
@@ -130,7 +130,7 @@ def test_oneshot_docstring_sample(self):
         import splunklib.client as client
         import splunklib.results as results
         service = self.service # cheat
-        rr = results.ResultsReader(service.jobs.oneshot("search * | head 5"))
+        rr = results.JSONResultsReader(service.jobs.oneshot("search * | head 5", output_mode='json'))
         for result in rr:
             if isinstance(result, results.Message):
                 # Diagnostic messages may be returned in the results
@@ -295,12 +295,12 @@ def test_get_preview_and_events(self):
         self.assertEventuallyTrue(self.job.is_done)
         self.assertLessEqual(int(self.job['eventCount']), 3)
 
-        preview_stream = self.job.preview()
-        preview_r = results.ResultsReader(preview_stream)
+        preview_stream = self.job.preview(output_mode='json')
+        preview_r = results.JSONResultsReader(preview_stream)
         self.assertFalse(preview_r.is_preview)
 
-        events_stream = self.job.events()
-        events_r = results.ResultsReader(events_stream)
+        events_stream = self.job.events(output_mode='json')
+        events_r = results.JSONResultsReader(events_stream)
 
         n_events = len([x for x in events_r if isinstance(x, dict)])
         n_preview = len([x for x in preview_r if isinstance(x, dict)])

tests/test_results.py

@@ -30,7 +30,7 @@ def test_read_from_empty_result_set(self):
         job = self.service.jobs.create("search index=_internal_does_not_exist | head 2")
         while not job.is_done():
             sleep(0.5)
-        self.assertEqual(0, len(list(results.ResultsReader(io.BufferedReader(job.results())))))
+        self.assertEqual(0, len(list(results.JSONResultsReader(io.BufferedReader(job.results(output_mode='json'))))))
 
     def test_read_normal_results(self):
         xml_text = """