Add support for specifying custom JSON parsers (#208)

* Added custom JSON parser

* Added documentation

* Forgot to add contrib doc

* Added contrib to top-level module

* Added community default decoder

* Added pragma:  no-cover

docs/contrib.rst

@@ -0,0 +1,52 @@
+.. _contrib:
+
+===================================
+Community-Contributed Functionality
+===================================
+
+
+When maintaining ``tda-api``, the authors have two goals: make common things 
+easy, and make uncommon things possible. This meets the needs of vast majority 
+of the community, while allowing advanced users or those with very niche 
+requirements to progress using potentially custom approaches. 
+
+However, this philosophy explicitly excludes functionality that is potentially 
+useful to many users, but is either not directly related to the core 
+functionality of the API wrapper. This is where the ``contrib`` module comes 
+into play. 
+
+This module is a collection of high-quality code that was produced by the 
+community and for the community. It includes utility methods that provide 
+additional functionality beyond the core library, fixes for quirks in API 
+behavior, etc. This page lists the available functionality. If you'd like to 
+discuss this or propose/request new additions, please join our `Discord server
+<https://discord.gg/Ddha8cm6dx>`__.
+
+
+.. _custom_json_decoding:
+
+
+--------------------
+Custom JSON Decoding
+--------------------
+
+TDA's API occasionally emits invalid JSON in the stream. This class implements 
+all known workarounds and hacks to get around these quirks:
+
+.. autoclass:: tda.contrib.util::HeuristicJsonDecoder
+  :members:
+  :undoc-members:
+
+You can use it as follows: 
+
+.. code-block:: python
+
+  from tda.contrib.util import HeuristicJsonDecoder
+
+  stream_client = # ... create your stream
+  stream_client.set_json_decoder(HeuristicJsonDecoder())
+  # ... continue as normal
+
+If you encounter invalid stream items that are not fixed by using this decoder, 
+please let us know in our `Discord server <https://discord.gg/Ddha8cm6dx>`__ or 
+follow the guide in :ref:`contributing` to add new functionality.

docs/index.rst

@@ -26,6 +26,7 @@
   util
   example
   help
+  contrib
   contributing
 
 Indices and tables

docs/streaming.rst

@@ -641,3 +641,35 @@ Fixing this is a task for the application developer: if you are writing to a
 database or filesystem as part of your handler, consider profiling it to make 
 the write faster. You may also consider deferring your writes so that slow 
 operations don't happen in the hotpath of the message handler. 
+
+
+---------------
+JSONDecodeError
+---------------
+
+This is an error that is most often raised when TDA sends an invalid JSON 
+string. See :ref:`custom_json_decoding` for details.
+
+For reasons known only to TDAmeritrade's development team, the API occasionally 
+emits invalid stream messages for some endpoints. Because this issue does not 
+affect all endpoints, and because ``tda-api``'s authors are not in the business 
+of handling quirks of an API they don't control, the library simply passes these
+errors up to the user. 
+
+However, some applications cannot handle complete failure. What's more, some 
+users have insight into how to work around these decoder errors. The streaming 
+client supports setting a custom JSON decoder to help with this: 
+
+.. automethod:: tda.streaming.StreamClient.set_json_decoder
+
+Users are free to implement their own JSON decoders by subclassing the following 
+abstract base class: 
+
+.. autoclass:: tda.streaming::StreamJsonDecoder
+  :members:
+  :undoc-members:
+
+Users looking for an out-of-the-box solution can consider using the 
+community-maintained decoder described in :ref:`custom_json_decoding`. Note that 
+while this decoder is constantly improving, it is not guaranteed to solve 
+whatever JSON decoding errors your may be encountering. 

tda/__init__.py

@@ -1,5 +1,6 @@
 from . import auth
 from . import client
+from . import contrib
 from . import debug
 from . import orders
 from . import streaming

tda/contrib/__init__.py

@@ -0,0 +1 @@
+from . import util

tda/contrib/util.py

@@ -0,0 +1,27 @@
+import json
+from tda.streaming import StreamJsonDecoder
+
+
+class HeuristicJsonDecoder(StreamJsonDecoder):
+    def decode_json_string(self, raw):
+        '''
+        Attempts the following, in order:
+        
+        1. Return the JSON decoding of the raw string. 
+        2. Replace all instances of ``\\\\\\\\`` with ``\\\\`` and return the 
+           decoding.
+
+        Note alternative (and potentially expensive) transformations are only 
+        performed when ``JSONDecodeError`` exceptions are raised by earlier 
+        stages.
+        '''
+
+        # Note "no cover" pragmas are added pending addition of real-world test 
+        # cases which trigger this issue.
+
+        try:
+            return json.loads(raw)
+        except json.decoder.JSONDecodeError:  # pragma: no cover
+            raw = raw.replace('\\\\', '\\')
+
+        return json.loads(raw)  # pragma: no cover

tda/streaming.py

@@ -1,3 +1,4 @@
+from abc import ABC, abstractmethod
 from collections import defaultdict, deque
 from enum import Enum
 
@@ -15,6 +16,21 @@
 from .utils import EnumEnforcer
 
 
+class StreamJsonDecoder(ABC):
+    @abstractmethod
+    def decode_json_string(self, raw):
+        '''
+        Parse a JSON-formatted string into a proper object. Raises 
+        ``JSONDecodeError`` on parse failure.
+        '''
+        raise NotImplementedError()
+
+
+class NaiveJsonStreamDecoder(StreamJsonDecoder):
+    def decode_json_string(self, raw):
+        return json.loads(raw)
+
+
 def get_logger():
     return logging.getLogger(__name__)
 
@@ -115,6 +131,25 @@ def __init__(self, client, *, account_id=None,
         self.logger = get_logger()
         self.request_number = 0
 
+        # Initialize the JSON parser to be the naive parser which directly calls 
+        # ``json.loads``
+        self.json_decoder = NaiveJsonStreamDecoder()
+
+
+    def set_json_decoder(self, json_decoder):
+        '''
+        Sets a custom JSON decoder.
+
+        :param json_decoder: Custom JSON decoder to use for to decode all 
+                             incoming JSON strings. See
+                             :class:`StreamJsonDecoder` for details.
+        '''
+        if not isinstance(json_decoder, tda.contrib.util.StreamJsonDecoder):
+            raise ValueError('Custom JSON parser must be a subclass of ' +
+                    'tda.contrib.util.StreamJsonDecoder')
+        self.json_decoder = json_decoder
+
+
     def req_num(self):
         self.request_number += 1
         return self.request_number
@@ -143,7 +178,7 @@ async def _receive(self):
         else:
             raw = await self._socket.recv()
             try:
-                ret = json.loads(raw)
+                ret = self.json_decoder.decode_json_string(raw)
             except json.decoder.JSONDecodeError as e:
                 msg = ('Failed to parse message. This often happens with ' +
                        'unknown symbols or other error conditions. Full ' +

tests/contrib_test.py

@@ -0,0 +1,15 @@
+import unittest
+
+from tda.contrib.util import HeuristicJsonDecoder
+
+
+class HeuristicJsonDecoderTest(unittest.TestCase):
+    def test_raw_string_decodes(self):
+        self.assertEqual(HeuristicJsonDecoder().decode_json_string(
+            r'{"\\\\\\\\": "test"}'),
+            {r'\\\\': 'test'})
+
+
+    def test_replace_backslashes(self):
+        # TODO: Actually collect some failing use cases...
+        pass

tests/streaming_test.py

@@ -98,6 +98,49 @@ async def login_and_get_socket(self, ws_connect):
         socket.reset_mock()
         return socket
 
+    ##########################################################################
+    # Custom JSON Decoder
+
+
+    @asynctest.patch('tda.streaming.websockets.client.connect', new_callable=asynctest.CoroutineMock)
+    async def test_default_parser_invalid_message(self, ws_connect):
+        socket = await self.login_and_get_socket(ws_connect)
+
+        socket.recv.side_effect = ['invalid json']
+
+        # No custom parser
+        msg = ('Failed to parse message. This often happens with ' +
+               'unknown symbols or other error conditions. Full ' +
+               'message text:')
+        with self.assertRaisesRegex(tda.streaming.UnparsableMessage, msg):
+            await self.client.level_one_equity_subs(['GOOG', 'MSFT'])
+
+
+    @asynctest.patch('tda.streaming.websockets.client.connect', new_callable=asynctest.CoroutineMock)
+    async def test_custom_parser_invalid_message(self, ws_connect):
+        socket = await self.login_and_get_socket(ws_connect)
+
+        socket.recv.side_effect = ['invalid json']
+
+        class CustomJsonDecoder(tda.contrib.util.StreamJsonDecoder):
+            def decode_json_string(_, raw):
+                self.assertEqual(raw, 'invalid json')
+                return self.success_response(1, 'QUOTE', 'SUBS')
+
+        self.client.set_json_decoder(CustomJsonDecoder())
+        await self.client.level_one_equity_subs(['GOOG', 'MSFT'])
+
+
+    @asynctest.patch('tda.streaming.websockets.client.connect', new_callable=asynctest.CoroutineMock)
+    async def test_custom_parser_wrong_type(self, ws_connect):
+        socket = await self.login_and_get_socket(ws_connect)
+
+        socket.recv.side_effect = ['invalid json']
+
+        with self.assertRaises(ValueError):
+            self.client.set_json_decoder('')
+
+
     ##########################################################################
     # Login