parse json columns to python objects

tap_mysql/discover_utils.py

@@ -271,7 +271,7 @@ def schema_for_column(column):  # pylint: disable=too-many-branches
             result.multipleOf = 10 ** (0 - column.numeric_scale)
 
     elif data_type in JSON_TYPES:
-        result.type = ['null', 'object']
+        result.type = ['null', 'object', 'array']
 
     elif data_type in STRING_TYPES:
         result.type = ['null', 'string']
@@ -290,7 +290,7 @@ def schema_for_column(column):  # pylint: disable=too-many-branches
         result.format = 'binary'
 
     elif data_type in SPATIAL_TYPES:
-        result.type = ['null', 'object']
+        result.type = ['null', 'object', 'array']
         result.format = 'spatial'
 
     else:

tap_mysql/sync_strategies/common.py

@@ -4,6 +4,7 @@
 import datetime
 import singer
 import time
+import json
 
 from singer import metadata, utils, metrics
 
@@ -118,6 +119,12 @@ def row_to_singer_record(catalog_entry, version, row, columns, time_extracted):
                 boolean_representation = True
             row_to_persist += (boolean_representation,)
 
+        elif 'object' in property_type or property_type == 'object':
+            if not elem:
+                row_to_persist += (None,)
+            else:
+                row_to_persist += (json.loads(elem),)
+
         else:
             row_to_persist += (elem,)
     rec = dict(zip(columns, row_to_persist))

tests/integration/test_tap_mysql.py

@@ -1272,7 +1272,7 @@ def runTest(self):
 
         record_message = list(filter(lambda m: isinstance(m, singer.RecordMessage), SINGER_MESSAGES))[0]
         self.assertTrue(isinstance(record_message, singer.RecordMessage))
-        self.assertEqual(record_message.record, {'val': '{"a": 10, "b": "c"}'})
+        self.assertEqual(record_message.record, {'val': {'a': 10, 'b': 'c'}})
 
 
 class TestSupportedPK(unittest.TestCase):