We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Error running query: Decimal128('0') is not JSON serializable
ANY mongo query for table with Decimal128 field type fails.
Every mongo query must be aggregated and projected to not return Decimal128 fields.
We are using Decimal128 to work with e.g. transaction amounts so its critical to us.

### Technical details:

* Redash Version: Redash 6.0.0+b8537 (4780bd9c)
* Browser/OS: Safari
* How did you install Redash: Cloud

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `redash/query_runner/mongodb.py`
Content:
```
1 import datetime
2 import logging
3 import re
4 
5 from dateutil.parser import parse
6 
7 from redash.query_runner import *
8 from redash.utils import JSONEncoder, json_dumps, json_loads, parse_human_time
9 
10 logger = logging.getLogger(__name__)
11 
12 try:
13     import pymongo
14     from bson.objectid import ObjectId
15     from bson.timestamp import Timestamp
16     from bson.son import SON
17     from bson.json_util import object_hook as bson_object_hook
18     enabled = True
19 
20 except ImportError:
21     enabled = False
22 
23 
24 TYPES_MAP = {
25     str: TYPE_STRING,
26     unicode: TYPE_STRING,
27     int: TYPE_INTEGER,
28     long: TYPE_INTEGER,
29     float: TYPE_FLOAT,
30     bool: TYPE_BOOLEAN,
31     datetime.datetime: TYPE_DATETIME,
32 }
33 
34 
35 class MongoDBJSONEncoder(JSONEncoder):
36     def default(self, o):
37         if isinstance(o, ObjectId):
38             return str(o)
39         elif isinstance(o, Timestamp):
40             return super(MongoDBJSONEncoder, self).default(o.as_datetime())
41 
42         return super(MongoDBJSONEncoder, self).default(o)
43 
44 
45 date_regex = re.compile("ISODate\(\"(.*)\"\)", re.IGNORECASE)
46 
47 
48 def parse_oids(oids):
49     if not isinstance(oids, list):
50         raise Exception("$oids takes an array as input.")
51 
52     return [bson_object_hook({'$oid': oid}) for oid in oids]
53 
54 
55 def datetime_parser(dct):
56     for k, v in dct.iteritems():
57         if isinstance(v, basestring):
58             m = date_regex.findall(v)
59             if len(m) > 0:
60                 dct[k] = parse(m[0], yearfirst=True)
61 
62     if '$humanTime' in dct:
63         return parse_human_time(dct['$humanTime'])
64 
65     if '$oids' in dct:
66         return parse_oids(dct['$oids'])
67 
68     return bson_object_hook(dct)
69 
70 
71 def parse_query_json(query):
72     query_data = json_loads(query, object_hook=datetime_parser)
73     return query_data
74 
75 
76 def _get_column_by_name(columns, column_name):
77     for c in columns:
78         if "name" in c and c["name"] == column_name:
79             return c
80 
81     return None
82 
83 
84 def parse_results(results):
85     rows = []
86     columns = []
87 
88     for row in results:
89         parsed_row = {}
90 
91         for key in row:
92             if isinstance(row[key], dict):
93                 for inner_key in row[key]:
94                     column_name = u'{}.{}'.format(key, inner_key)
95                     if _get_column_by_name(columns, column_name) is None:
96                         columns.append({
97                             "name": column_name,
98                             "friendly_name": column_name,
99                             "type": TYPES_MAP.get(type(row[key][inner_key]), TYPE_STRING)
100                         })
101 
102                     parsed_row[column_name] = row[key][inner_key]
103 
104             else:
105                 if _get_column_by_name(columns, key) is None:
106                     columns.append({
107                         "name": key,
108                         "friendly_name": key,
109                         "type": TYPES_MAP.get(type(row[key]), TYPE_STRING)
110                     })
111 
112                 parsed_row[key] = row[key]
113 
114         rows.append(parsed_row)
115 
116     return rows, columns
117 
118 
119 class MongoDB(BaseQueryRunner):
120     @classmethod
121     def configuration_schema(cls):
122         return {
123             'type': 'object',
124             'properties': {
125                 'connectionString': {
126                     'type': 'string',
127                     'title': 'Connection String'
128                 },
129                 'dbName': {
130                     'type': 'string',
131                     'title': "Database Name"
132                 },
133                 'replicaSetName': {
134                     'type': 'string',
135                     'title': 'Replica Set Name'
136                 },
137             },
138             'required': ['connectionString', 'dbName']
139         }
140 
141     @classmethod
142     def enabled(cls):
143         return enabled
144 
145     @classmethod
146     def annotate_query(cls):
147         return False
148 
149     def __init__(self, configuration):
150         super(MongoDB, self).__init__(configuration)
151 
152         self.syntax = 'json'
153 
154         self.db_name = self.configuration["dbName"]
155 
156         self.is_replica_set = True if "replicaSetName" in self.configuration and self.configuration["replicaSetName"] else False
157 
158     def _get_db(self):
159         if self.is_replica_set:
160             db_connection = pymongo.MongoClient(self.configuration["connectionString"],
161                                                 replicaSet=self.configuration["replicaSetName"])
162         else:
163             db_connection = pymongo.MongoClient(self.configuration["connectionString"])
164 
165         return db_connection[self.db_name]
166 
167     def test_connection(self):
168         db = self._get_db()
169         if not db.command("connectionStatus")["ok"]:
170             raise Exception("MongoDB connection error")
171 
172     def _merge_property_names(self, columns, document):
173         for property in document:
174               if property not in columns:
175                   columns.append(property)
176 
177     def _is_collection_a_view(self, db, collection_name):
178         if 'viewOn' in db[collection_name].options():
179             return True
180         else:
181             return False
182 
183     def _get_collection_fields(self, db, collection_name):
184         # Since MongoDB is a document based database and each document doesn't have
185         # to have the same fields as another documet in the collection its a bit hard to
186         # show these attributes as fields in the schema.
187         #
188         # For now, the logic is to take the first and last documents (last is determined
189         # by the Natural Order (http://www.mongodb.org/display/DOCS/Sorting+and+Natural+Order)
190         # as we don't know the correct order. In most single server installations it would be
191         # find. In replicaset when reading from non master it might not return the really last
192         # document written.
193         collection_is_a_view = self._is_collection_a_view(db, collection_name)
194         documents_sample = []
195         if collection_is_a_view:
196             for d in db[collection_name].find().limit(2):
197                 documents_sample.append(d)
198         else:
199             for d in db[collection_name].find().sort([("$natural", 1)]).limit(1):
200                 documents_sample.append(d)
201 
202             for d in db[collection_name].find().sort([("$natural", -1)]).limit(1):
203                 documents_sample.append(d)
204         columns = []
205         for d in documents_sample:
206             self._merge_property_names(columns, d)
207         return columns
208 
209     def get_schema(self, get_stats=False):
210         schema = {}
211         db = self._get_db()
212         for collection_name in db.collection_names():
213             if collection_name.startswith('system.'):
214                 continue
215             columns = self._get_collection_fields(db, collection_name)
216             schema[collection_name] = {
217                 "name": collection_name, "columns": sorted(columns)}
218 
219         return schema.values()
220 
221 
222     def run_query(self, query, user):
223         db = self._get_db()
224 
225         logger.debug("mongodb connection string: %s", self.configuration['connectionString'])
226         logger.debug("mongodb got query: %s", query)
227 
228         try:
229             query_data = parse_query_json(query)
230         except ValueError:
231             return None, "Invalid query format. The query is not a valid JSON."
232 
233         if "collection" not in query_data:
234             return None, "'collection' must have a value to run a query"
235         else:
236             collection = query_data["collection"]
237 
238         q = query_data.get("query", None)
239         f = None
240 
241         aggregate = query_data.get("aggregate", None)
242         if aggregate:
243             for step in aggregate:
244                 if "$sort" in step:
245                     sort_list = []
246                     for sort_item in step["$sort"]:
247                         sort_list.append((sort_item["name"], sort_item["direction"]))
248 
249                     step["$sort"] = SON(sort_list)
250 
251         if not aggregate:
252             s = None
253             if "sort" in query_data and query_data["sort"]:
254                 s = []
255                 for field in query_data["sort"]:
256                     s.append((field["name"], field["direction"]))
257 
258         if "fields" in query_data:
259             f = query_data["fields"]
260 
261         s = None
262         if "sort" in query_data and query_data["sort"]:
263             s = []
264             for field_data in query_data["sort"]:
265                 s.append((field_data["name"], field_data["direction"]))
266 
267         columns = []
268         rows = []
269 
270         cursor = None
271         if q or (not q and not aggregate):
272             if s:
273                 cursor = db[collection].find(q, f).sort(s)
274             else:
275                 cursor = db[collection].find(q, f)
276 
277             if "skip" in query_data:
278                 cursor = cursor.skip(query_data["skip"])
279 
280             if "limit" in query_data:
281                 cursor = cursor.limit(query_data["limit"])
282 
283             if "count" in query_data:
284                 cursor = cursor.count()
285 
286         elif aggregate:
287             allow_disk_use = query_data.get('allowDiskUse', False)
288             r = db[collection].aggregate(aggregate, allowDiskUse=allow_disk_use)
289 
290             # Backwards compatibility with older pymongo versions.
291             #
292             # Older pymongo version would return a dictionary from an aggregate command.
293             # The dict would contain a "result" key which would hold the cursor.
294             # Newer ones return pymongo.command_cursor.CommandCursor.
295             if isinstance(r, dict):
296                 cursor = r["result"]
297             else:
298                 cursor = r
299 
300         if "count" in query_data:
301             columns.append({
302                 "name" : "count",
303                 "friendly_name" : "count",
304                 "type" : TYPE_INTEGER
305             })
306 
307             rows.append({ "count" : cursor })
308         else:
309             rows, columns = parse_results(cursor)
310 
311         if f:
312             ordered_columns = []
313             for k in sorted(f, key=f.get):
314                 column = _get_column_by_name(columns, k)
315                 if column:
316                     ordered_columns.append(column)
317 
318             columns = ordered_columns
319 
320         if query_data.get('sortColumns'):
321             reverse = query_data['sortColumns'] == 'desc'
322             columns = sorted(columns, key=lambda col: col['name'], reverse=reverse)
323 
324         data = {
325             "columns": columns,
326             "rows": rows
327         }
328         error = None
329         json_data = json_dumps(data, cls=MongoDBJSONEncoder)
330 
331         return json_data, error
332 
333 register(MongoDB)
334
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/redash/query_runner/mongodb.py b/redash/query_runner/mongodb.py
--- a/redash/query_runner/mongodb.py
+++ b/redash/query_runner/mongodb.py
@@ -13,6 +13,7 @@
     import pymongo
     from bson.objectid import ObjectId
     from bson.timestamp import Timestamp
+    from bson.decimal128 import Decimal128
     from bson.son import SON
     from bson.json_util import object_hook as bson_object_hook
     enabled = True
@@ -38,7 +39,8 @@
             return str(o)
         elif isinstance(o, Timestamp):
             return super(MongoDBJSONEncoder, self).default(o.as_datetime())
-
+        elif isinstance(o, Decimal128):
+            return o.to_decimal()
         return super(MongoDBJSONEncoder, self).default(o)

{"golden_diff": "diff --git a/redash/query_runner/mongodb.py b/redash/query_runner/mongodb.py\n--- a/redash/query_runner/mongodb.py\n+++ b/redash/query_runner/mongodb.py\n@@ -13,6 +13,7 @@\n     import pymongo\n     from bson.objectid import ObjectId\n     from bson.timestamp import Timestamp\n+    from bson.decimal128 import Decimal128\n     from bson.son import SON\n     from bson.json_util import object_hook as bson_object_hook\n     enabled = True\n@@ -38,7 +39,8 @@\n             return str(o)\n         elif isinstance(o, Timestamp):\n             return super(MongoDBJSONEncoder, self).default(o.as_datetime())\n-\n+        elif isinstance(o, Decimal128):\n+            return o.to_decimal()\n         return super(MongoDBJSONEncoder, self).default(o)\n", "issue": "Error running query: Decimal128('0') is not JSON serializable\nANY mongo query for table with Decimal128 field type fails.\r\nEvery mongo query must be aggregated and projected to not return Decimal128 fields.\r\nWe are using Decimal128 to work with e.g. transaction amounts so its critical to us.\r\n\r\n### Technical details:\r\n\r\n* Redash Version: Redash 6.0.0+b8537 (4780bd9c)\r\n* Browser/OS: Safari\r\n* How did you install Redash: Cloud\n", "before_files": [{"content": "import datetime\nimport logging\nimport re\n\nfrom dateutil.parser import parse\n\nfrom redash.query_runner import *\nfrom redash.utils import JSONEncoder, json_dumps, json_loads, parse_human_time\n\nlogger = logging.getLogger(__name__)\n\ntry:\n    import pymongo\n    from bson.objectid import ObjectId\n    from bson.timestamp import Timestamp\n    from bson.son import SON\n    from bson.json_util import object_hook as bson_object_hook\n    enabled = True\n\nexcept ImportError:\n    enabled = False\n\n\nTYPES_MAP = {\n    str: TYPE_STRING,\n    unicode: TYPE_STRING,\n    int: TYPE_INTEGER,\n    long: TYPE_INTEGER,\n    float: TYPE_FLOAT,\n    bool: TYPE_BOOLEAN,\n    datetime.datetime: TYPE_DATETIME,\n}\n\n\nclass MongoDBJSONEncoder(JSONEncoder):\n    def default(self, o):\n        if isinstance(o, ObjectId):\n            return str(o)\n        elif isinstance(o, Timestamp):\n            return super(MongoDBJSONEncoder, self).default(o.as_datetime())\n\n        return super(MongoDBJSONEncoder, self).default(o)\n\n\ndate_regex = re.compile(\"ISODate\\(\\\"(.*)\\\"\\)\", re.IGNORECASE)\n\n\ndef parse_oids(oids):\n    if not isinstance(oids, list):\n        raise Exception(\"$oids takes an array as input.\")\n\n    return [bson_object_hook({'$oid': oid}) for oid in oids]\n\n\ndef datetime_parser(dct):\n    for k, v in dct.iteritems():\n        if isinstance(v, basestring):\n            m = date_regex.findall(v)\n            if len(m) > 0:\n                dct[k] = parse(m[0], yearfirst=True)\n\n    if '$humanTime' in dct:\n        return parse_human_time(dct['$humanTime'])\n\n    if '$oids' in dct:\n        return parse_oids(dct['$oids'])\n\n    return bson_object_hook(dct)\n\n\ndef parse_query_json(query):\n    query_data = json_loads(query, object_hook=datetime_parser)\n    return query_data\n\n\ndef _get_column_by_name(columns, column_name):\n    for c in columns:\n        if \"name\" in c and c[\"name\"] == column_name:\n            return c\n\n    return None\n\n\ndef parse_results(results):\n    rows = []\n    columns = []\n\n    for row in results:\n        parsed_row = {}\n\n        for key in row:\n            if isinstance(row[key], dict):\n                for inner_key in row[key]:\n                    column_name = u'{}.{}'.format(key, inner_key)\n                    if _get_column_by_name(columns, column_name) is None:\n                        columns.append({\n                            \"name\": column_name,\n                            \"friendly_name\": column_name,\n                            \"type\": TYPES_MAP.get(type(row[key][inner_key]), TYPE_STRING)\n                        })\n\n                    parsed_row[column_name] = row[key][inner_key]\n\n            else:\n                if _get_column_by_name(columns, key) is None:\n                    columns.append({\n                        \"name\": key,\n                        \"friendly_name\": key,\n                        \"type\": TYPES_MAP.get(type(row[key]), TYPE_STRING)\n                    })\n\n                parsed_row[key] = row[key]\n\n        rows.append(parsed_row)\n\n    return rows, columns\n\n\nclass MongoDB(BaseQueryRunner):\n    @classmethod\n    def configuration_schema(cls):\n        return {\n            'type': 'object',\n            'properties': {\n                'connectionString': {\n                    'type': 'string',\n                    'title': 'Connection String'\n                },\n                'dbName': {\n                    'type': 'string',\n                    'title': \"Database Name\"\n                },\n                'replicaSetName': {\n                    'type': 'string',\n                    'title': 'Replica Set Name'\n                },\n            },\n            'required': ['connectionString', 'dbName']\n        }\n\n    @classmethod\n    def enabled(cls):\n        return enabled\n\n    @classmethod\n    def annotate_query(cls):\n        return False\n\n    def __init__(self, configuration):\n        super(MongoDB, self).__init__(configuration)\n\n        self.syntax = 'json'\n\n        self.db_name = self.configuration[\"dbName\"]\n\n        self.is_replica_set = True if \"replicaSetName\" in self.configuration and self.configuration[\"replicaSetName\"] else False\n\n    def _get_db(self):\n        if self.is_replica_set:\n            db_connection = pymongo.MongoClient(self.configuration[\"connectionString\"],\n                                                replicaSet=self.configuration[\"replicaSetName\"])\n        else:\n            db_connection = pymongo.MongoClient(self.configuration[\"connectionString\"])\n\n        return db_connection[self.db_name]\n\n    def test_connection(self):\n        db = self._get_db()\n        if not db.command(\"connectionStatus\")[\"ok\"]:\n            raise Exception(\"MongoDB connection error\")\n\n    def _merge_property_names(self, columns, document):\n        for property in document:\n              if property not in columns:\n                  columns.append(property)\n\n    def _is_collection_a_view(self, db, collection_name):\n        if 'viewOn' in db[collection_name].options():\n            return True\n        else:\n            return False\n\n    def _get_collection_fields(self, db, collection_name):\n        # Since MongoDB is a document based database and each document doesn't have\n        # to have the same fields as another documet in the collection its a bit hard to\n        # show these attributes as fields in the schema.\n        #\n        # For now, the logic is to take the first and last documents (last is determined\n        # by the Natural Order (http://www.mongodb.org/display/DOCS/Sorting+and+Natural+Order)\n        # as we don't know the correct order. In most single server installations it would be\n        # find. In replicaset when reading from non master it might not return the really last\n        # document written.\n        collection_is_a_view = self._is_collection_a_view(db, collection_name)\n        documents_sample = []\n        if collection_is_a_view:\n            for d in db[collection_name].find().limit(2):\n                documents_sample.append(d)\n        else:\n            for d in db[collection_name].find().sort([(\"$natural\", 1)]).limit(1):\n                documents_sample.append(d)\n\n            for d in db[collection_name].find().sort([(\"$natural\", -1)]).limit(1):\n                documents_sample.append(d)\n        columns = []\n        for d in documents_sample:\n            self._merge_property_names(columns, d)\n        return columns\n\n    def get_schema(self, get_stats=False):\n        schema = {}\n        db = self._get_db()\n        for collection_name in db.collection_names():\n            if collection_name.startswith('system.'):\n                continue\n            columns = self._get_collection_fields(db, collection_name)\n            schema[collection_name] = {\n                \"name\": collection_name, \"columns\": sorted(columns)}\n\n        return schema.values()\n\n\n    def run_query(self, query, user):\n        db = self._get_db()\n\n        logger.debug(\"mongodb connection string: %s\", self.configuration['connectionString'])\n        logger.debug(\"mongodb got query: %s\", query)\n\n        try:\n            query_data = parse_query_json(query)\n        except ValueError:\n            return None, \"Invalid query format. The query is not a valid JSON.\"\n\n        if \"collection\" not in query_data:\n            return None, \"'collection' must have a value to run a query\"\n        else:\n            collection = query_data[\"collection\"]\n\n        q = query_data.get(\"query\", None)\n        f = None\n\n        aggregate = query_data.get(\"aggregate\", None)\n        if aggregate:\n            for step in aggregate:\n                if \"$sort\" in step:\n                    sort_list = []\n                    for sort_item in step[\"$sort\"]:\n                        sort_list.append((sort_item[\"name\"], sort_item[\"direction\"]))\n\n                    step[\"$sort\"] = SON(sort_list)\n\n        if not aggregate:\n            s = None\n            if \"sort\" in query_data and query_data[\"sort\"]:\n                s = []\n                for field in query_data[\"sort\"]:\n                    s.append((field[\"name\"], field[\"direction\"]))\n\n        if \"fields\" in query_data:\n            f = query_data[\"fields\"]\n\n        s = None\n        if \"sort\" in query_data and query_data[\"sort\"]:\n            s = []\n            for field_data in query_data[\"sort\"]:\n                s.append((field_data[\"name\"], field_data[\"direction\"]))\n\n        columns = []\n        rows = []\n\n        cursor = None\n        if q or (not q and not aggregate):\n            if s:\n                cursor = db[collection].find(q, f).sort(s)\n            else:\n                cursor = db[collection].find(q, f)\n\n            if \"skip\" in query_data:\n                cursor = cursor.skip(query_data[\"skip\"])\n\n            if \"limit\" in query_data:\n                cursor = cursor.limit(query_data[\"limit\"])\n\n            if \"count\" in query_data:\n                cursor = cursor.count()\n\n        elif aggregate:\n            allow_disk_use = query_data.get('allowDiskUse', False)\n            r = db[collection].aggregate(aggregate, allowDiskUse=allow_disk_use)\n\n            # Backwards compatibility with older pymongo versions.\n            #\n            # Older pymongo version would return a dictionary from an aggregate command.\n            # The dict would contain a \"result\" key which would hold the cursor.\n            # Newer ones return pymongo.command_cursor.CommandCursor.\n            if isinstance(r, dict):\n                cursor = r[\"result\"]\n            else:\n                cursor = r\n\n        if \"count\" in query_data:\n            columns.append({\n                \"name\" : \"count\",\n                \"friendly_name\" : \"count\",\n                \"type\" : TYPE_INTEGER\n            })\n\n            rows.append({ \"count\" : cursor })\n        else:\n            rows, columns = parse_results(cursor)\n\n        if f:\n            ordered_columns = []\n            for k in sorted(f, key=f.get):\n                column = _get_column_by_name(columns, k)\n                if column:\n                    ordered_columns.append(column)\n\n            columns = ordered_columns\n\n        if query_data.get('sortColumns'):\n            reverse = query_data['sortColumns'] == 'desc'\n            columns = sorted(columns, key=lambda col: col['name'], reverse=reverse)\n\n        data = {\n            \"columns\": columns,\n            \"rows\": rows\n        }\n        error = None\n        json_data = json_dumps(data, cls=MongoDBJSONEncoder)\n\n        return json_data, error\n\nregister(MongoDB)\n", "path": "redash/query_runner/mongodb.py"}], "after_files": [{"content": "import datetime\nimport logging\nimport re\n\nfrom dateutil.parser import parse\n\nfrom redash.query_runner import *\nfrom redash.utils import JSONEncoder, json_dumps, json_loads, parse_human_time\n\nlogger = logging.getLogger(__name__)\n\ntry:\n    import pymongo\n    from bson.objectid import ObjectId\n    from bson.timestamp import Timestamp\n    from bson.decimal128 import Decimal128\n    from bson.son import SON\n    from bson.json_util import object_hook as bson_object_hook\n    enabled = True\n\nexcept ImportError:\n    enabled = False\n\n\nTYPES_MAP = {\n    str: TYPE_STRING,\n    unicode: TYPE_STRING,\n    int: TYPE_INTEGER,\n    long: TYPE_INTEGER,\n    float: TYPE_FLOAT,\n    bool: TYPE_BOOLEAN,\n    datetime.datetime: TYPE_DATETIME,\n}\n\n\nclass MongoDBJSONEncoder(JSONEncoder):\n    def default(self, o):\n        if isinstance(o, ObjectId):\n            return str(o)\n        elif isinstance(o, Timestamp):\n            return super(MongoDBJSONEncoder, self).default(o.as_datetime())\n        elif isinstance(o, Decimal128):\n            return o.to_decimal()\n        return super(MongoDBJSONEncoder, self).default(o)\n\n\ndate_regex = re.compile(\"ISODate\\(\\\"(.*)\\\"\\)\", re.IGNORECASE)\n\n\ndef parse_oids(oids):\n    if not isinstance(oids, list):\n        raise Exception(\"$oids takes an array as input.\")\n\n    return [bson_object_hook({'$oid': oid}) for oid in oids]\n\n\ndef datetime_parser(dct):\n    for k, v in dct.iteritems():\n        if isinstance(v, basestring):\n            m = date_regex.findall(v)\n            if len(m) > 0:\n                dct[k] = parse(m[0], yearfirst=True)\n\n    if '$humanTime' in dct:\n        return parse_human_time(dct['$humanTime'])\n\n    if '$oids' in dct:\n        return parse_oids(dct['$oids'])\n\n    return bson_object_hook(dct)\n\n\ndef parse_query_json(query):\n    query_data = json_loads(query, object_hook=datetime_parser)\n    return query_data\n\n\ndef _get_column_by_name(columns, column_name):\n    for c in columns:\n        if \"name\" in c and c[\"name\"] == column_name:\n            return c\n\n    return None\n\n\ndef parse_results(results):\n    rows = []\n    columns = []\n\n    for row in results:\n        parsed_row = {}\n\n        for key in row:\n            if isinstance(row[key], dict):\n                for inner_key in row[key]:\n                    column_name = u'{}.{}'.format(key, inner_key)\n                    if _get_column_by_name(columns, column_name) is None:\n                        columns.append({\n                            \"name\": column_name,\n                            \"friendly_name\": column_name,\n                            \"type\": TYPES_MAP.get(type(row[key][inner_key]), TYPE_STRING)\n                        })\n\n                    parsed_row[column_name] = row[key][inner_key]\n\n            else:\n                if _get_column_by_name(columns, key) is None:\n                    columns.append({\n                        \"name\": key,\n                        \"friendly_name\": key,\n                        \"type\": TYPES_MAP.get(type(row[key]), TYPE_STRING)\n                    })\n\n                parsed_row[key] = row[key]\n\n        rows.append(parsed_row)\n\n    return rows, columns\n\n\nclass MongoDB(BaseQueryRunner):\n    @classmethod\n    def configuration_schema(cls):\n        return {\n            'type': 'object',\n            'properties': {\n                'connectionString': {\n                    'type': 'string',\n                    'title': 'Connection String'\n                },\n                'dbName': {\n                    'type': 'string',\n                    'title': \"Database Name\"\n                },\n                'replicaSetName': {\n                    'type': 'string',\n                    'title': 'Replica Set Name'\n                },\n            },\n            'required': ['connectionString', 'dbName']\n        }\n\n    @classmethod\n    def enabled(cls):\n        return enabled\n\n    @classmethod\n    def annotate_query(cls):\n        return False\n\n    def __init__(self, configuration):\n        super(MongoDB, self).__init__(configuration)\n\n        self.syntax = 'json'\n\n        self.db_name = self.configuration[\"dbName\"]\n\n        self.is_replica_set = True if \"replicaSetName\" in self.configuration and self.configuration[\"replicaSetName\"] else False\n\n    def _get_db(self):\n        if self.is_replica_set:\n            db_connection = pymongo.MongoClient(self.configuration[\"connectionString\"],\n                                                replicaSet=self.configuration[\"replicaSetName\"])\n        else:\n            db_connection = pymongo.MongoClient(self.configuration[\"connectionString\"])\n\n        return db_connection[self.db_name]\n\n    def test_connection(self):\n        db = self._get_db()\n        if not db.command(\"connectionStatus\")[\"ok\"]:\n            raise Exception(\"MongoDB connection error\")\n\n    def _merge_property_names(self, columns, document):\n        for property in document:\n              if property not in columns:\n                  columns.append(property)\n\n    def _is_collection_a_view(self, db, collection_name):\n        if 'viewOn' in db[collection_name].options():\n            return True\n        else:\n            return False\n\n    def _get_collection_fields(self, db, collection_name):\n        # Since MongoDB is a document based database and each document doesn't have\n        # to have the same fields as another documet in the collection its a bit hard to\n        # show these attributes as fields in the schema.\n        #\n        # For now, the logic is to take the first and last documents (last is determined\n        # by the Natural Order (http://www.mongodb.org/display/DOCS/Sorting+and+Natural+Order)\n        # as we don't know the correct order. In most single server installations it would be\n        # find. In replicaset when reading from non master it might not return the really last\n        # document written.\n        collection_is_a_view = self._is_collection_a_view(db, collection_name)\n        documents_sample = []\n        if collection_is_a_view:\n            for d in db[collection_name].find().limit(2):\n                documents_sample.append(d)\n        else:\n            for d in db[collection_name].find().sort([(\"$natural\", 1)]).limit(1):\n                documents_sample.append(d)\n\n            for d in db[collection_name].find().sort([(\"$natural\", -1)]).limit(1):\n                documents_sample.append(d)\n        columns = []\n        for d in documents_sample:\n            self._merge_property_names(columns, d)\n        return columns\n\n    def get_schema(self, get_stats=False):\n        schema = {}\n        db = self._get_db()\n        for collection_name in db.collection_names():\n            if collection_name.startswith('system.'):\n                continue\n            columns = self._get_collection_fields(db, collection_name)\n            schema[collection_name] = {\n                \"name\": collection_name, \"columns\": sorted(columns)}\n\n        return schema.values()\n\n\n    def run_query(self, query, user):\n        db = self._get_db()\n\n        logger.debug(\"mongodb connection string: %s\", self.configuration['connectionString'])\n        logger.debug(\"mongodb got query: %s\", query)\n\n        try:\n            query_data = parse_query_json(query)\n        except ValueError:\n            return None, \"Invalid query format. The query is not a valid JSON.\"\n\n        if \"collection\" not in query_data:\n            return None, \"'collection' must have a value to run a query\"\n        else:\n            collection = query_data[\"collection\"]\n\n        q = query_data.get(\"query\", None)\n        f = None\n\n        aggregate = query_data.get(\"aggregate\", None)\n        if aggregate:\n            for step in aggregate:\n                if \"$sort\" in step:\n                    sort_list = []\n                    for sort_item in step[\"$sort\"]:\n                        sort_list.append((sort_item[\"name\"], sort_item[\"direction\"]))\n\n                    step[\"$sort\"] = SON(sort_list)\n\n        if not aggregate:\n            s = None\n            if \"sort\" in query_data and query_data[\"sort\"]:\n                s = []\n                for field in query_data[\"sort\"]:\n                    s.append((field[\"name\"], field[\"direction\"]))\n\n        if \"fields\" in query_data:\n            f = query_data[\"fields\"]\n\n        s = None\n        if \"sort\" in query_data and query_data[\"sort\"]:\n            s = []\n            for field_data in query_data[\"sort\"]:\n                s.append((field_data[\"name\"], field_data[\"direction\"]))\n\n        columns = []\n        rows = []\n\n        cursor = None\n        if q or (not q and not aggregate):\n            if s:\n                cursor = db[collection].find(q, f).sort(s)\n            else:\n                cursor = db[collection].find(q, f)\n\n            if \"skip\" in query_data:\n                cursor = cursor.skip(query_data[\"skip\"])\n\n            if \"limit\" in query_data:\n                cursor = cursor.limit(query_data[\"limit\"])\n\n            if \"count\" in query_data:\n                cursor = cursor.count()\n\n        elif aggregate:\n            allow_disk_use = query_data.get('allowDiskUse', False)\n            r = db[collection].aggregate(aggregate, allowDiskUse=allow_disk_use)\n\n            # Backwards compatibility with older pymongo versions.\n            #\n            # Older pymongo version would return a dictionary from an aggregate command.\n            # The dict would contain a \"result\" key which would hold the cursor.\n            # Newer ones return pymongo.command_cursor.CommandCursor.\n            if isinstance(r, dict):\n                cursor = r[\"result\"]\n            else:\n                cursor = r\n\n        if \"count\" in query_data:\n            columns.append({\n                \"name\" : \"count\",\n                \"friendly_name\" : \"count\",\n                \"type\" : TYPE_INTEGER\n            })\n\n            rows.append({ \"count\" : cursor })\n        else:\n            rows, columns = parse_results(cursor)\n\n        if f:\n            ordered_columns = []\n            for k in sorted(f, key=f.get):\n                column = _get_column_by_name(columns, k)\n                if column:\n                    ordered_columns.append(column)\n\n            columns = ordered_columns\n\n        if query_data.get('sortColumns'):\n            reverse = query_data['sortColumns'] == 'desc'\n            columns = sorted(columns, key=lambda col: col['name'], reverse=reverse)\n\n        data = {\n            \"columns\": columns,\n            \"rows\": rows\n        }\n        error = None\n        json_data = json_dumps(data, cls=MongoDBJSONEncoder)\n\n        return json_data, error\n\nregister(MongoDB)\n", "path": "redash/query_runner/mongodb.py"}]}