We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Error to load MongoDB collections
### Issue Summary

When you create a Mongodb data source using a mongodb user which has access to a databse but doesn't have privileges to find records in a specific collection under the database, redash can't refresh the schema as it tried to get a data sample even though the user doesn't have access to the collection, that probably happens because the command list_collections returns a list of all collections regardless if the user has access to its data or not.

### Steps to Reproduce

1. Create a role in mongodb and give access only to certain collections.
2. Create a user in mongodb and assign the previous role to it.
3. Create a data source in redash.
4. Try selecting the newly created data source in the query page.

### Technical details:

* Redash Version: 10.1.0
* Browser/OS: Any
* How did you install Redash: Tried with Helm chart and aws market place


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

{"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@@ -221,15 +221,21 @@\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+        try:\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+        except Exception as ex:\n+            template = \"An exception of type {0} occurred. Arguments:\\n{1!r}\"\n+            message = template.format(type(ex).__name__, ex.args)\n+            logger.error(message)\n+            return []\n         columns = []\n         for d in documents_sample:\n             self._merge_property_names(columns, d)\n@@ -242,10 +248,11 @@\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,\n-                \"columns\": sorted(columns),\n-            }\n+            if columns:\n+                schema[collection_name] = {\n+                    \"name\": collection_name,\n+                    \"columns\": sorted(columns),\n+                }\n \n         return list(schema.values())\n", "issue": "Error to load MongoDB collections\n### Issue Summary\r\n\r\nWhen you create a Mongodb data source using a mongodb user which has access to a databse but doesn't have privileges to find records in a specific collection under the database, redash can't refresh the schema as it tried to get a data sample even though the user doesn't have access to the collection, that probably happens because the command list_collections returns a list of all collections regardless if the user has access to its data or not.\r\n\r\n### Steps to Reproduce\r\n\r\n1. Create a role in mongodb and give access only to certain collections.\r\n2. Create a user in mongodb and assign the previous role to it.\r\n3. Create a data source in redash.\r\n4. Try selecting the newly created data source in the query page.\r\n\r\n### Technical details:\r\n\r\n* Redash Version: 10.1.0\r\n* Browser/OS: Any\r\n* How did you install Redash: Tried with Helm chart and aws market place\r\n\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.decimal128 import Decimal128\n    from bson.son import SON\n    from bson.json_util import object_hook as bson_object_hook\n\n    enabled = True\n\nexcept ImportError:\n    enabled = False\n\n\nTYPES_MAP = {\n    str: TYPE_STRING,\n    bytes: TYPE_STRING,\n    int: 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.items():\n        if isinstance(v, str):\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 = \"{}.{}\".format(key, inner_key)\n                    if _get_column_by_name(columns, column_name) is None:\n                        columns.append(\n                            {\n                                \"name\": column_name,\n                                \"friendly_name\": column_name,\n                                \"type\": TYPES_MAP.get(\n                                    type(row[key][inner_key]), TYPE_STRING\n                                ),\n                            }\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                        {\n                            \"name\": key,\n                            \"friendly_name\": key,\n                            \"type\": TYPES_MAP.get(type(row[key]), TYPE_STRING),\n                        }\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    should_annotate_query = False\n\n    @classmethod\n    def configuration_schema(cls):\n        return {\n            \"type\": \"object\",\n            \"properties\": {\n                \"connectionString\": {\"type\": \"string\", \"title\": \"Connection String\"},\n                \"username\": {\"type\": \"string\"},\n                \"password\": {\"type\": \"string\"},\n                \"dbName\": {\"type\": \"string\", \"title\": \"Database Name\"},\n                \"replicaSetName\": {\"type\": \"string\", \"title\": \"Replica Set Name\"},\n                \"readPreference\": {\n                    \"type\": \"string\",\n                    \"extendedEnum\": [\n                        {\"value\": \"primaryPreferred\", \"name\": \"Primary Preferred\"},\n                        {\"value\": \"primary\", \"name\": \"Primary\"},\n                        {\"value\": \"secondary\", \"name\": \"Secondary\"},\n                        {\"value\": \"secondaryPreferred\", \"name\": \"Secondary Preferred\"},\n                        {\"value\": \"nearest\", \"name\": \"Nearest\"},\n                    ],\n                    \"title\": \"Replica Set Read Preference\",\n                },\n            },\n            \"secret\": [\"password\"],\n            \"required\": [\"connectionString\", \"dbName\"],\n        }\n\n    @classmethod\n    def enabled(cls):\n        return enabled\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 = (\n            True\n            if \"replicaSetName\" in self.configuration\n            and self.configuration[\"replicaSetName\"]\n            else False\n        )\n\n    def _get_db(self):\n        kwargs = {}\n        if self.is_replica_set:\n            kwargs[\"replicaSet\"] = self.configuration[\"replicaSetName\"]\n            readPreference = self.configuration.get(\"readPreference\")\n            if readPreference:\n                kwargs[\"readPreference\"] = readPreference\n\n        if \"username\" in self.configuration:\n            kwargs[\"username\"] = self.configuration[\"username\"]\n\n        if \"password\" in self.configuration:\n            kwargs[\"password\"] = self.configuration[\"password\"]\n\n        db_connection = pymongo.MongoClient(\n            self.configuration[\"connectionString\"], **kwargs\n        )\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        return db\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        # fine. 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,\n                \"columns\": sorted(columns),\n            }\n\n        return list(schema.values())\n\n    def run_query(self, query, user):\n        db = self._get_db()\n\n        logger.debug(\n            \"mongodb connection string: %s\", self.configuration[\"connectionString\"]\n        )\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 \"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\", \"friendly_name\": \"count\", \"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 = {\"columns\": columns, \"rows\": rows}\n        error = None\n        json_data = json_dumps(data, cls=MongoDBJSONEncoder)\n\n        return json_data, error\n\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\n    enabled = True\n\nexcept ImportError:\n    enabled = False\n\n\nTYPES_MAP = {\n    str: TYPE_STRING,\n    bytes: TYPE_STRING,\n    int: 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.items():\n        if isinstance(v, str):\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 = \"{}.{}\".format(key, inner_key)\n                    if _get_column_by_name(columns, column_name) is None:\n                        columns.append(\n                            {\n                                \"name\": column_name,\n                                \"friendly_name\": column_name,\n                                \"type\": TYPES_MAP.get(\n                                    type(row[key][inner_key]), TYPE_STRING\n                                ),\n                            }\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                        {\n                            \"name\": key,\n                            \"friendly_name\": key,\n                            \"type\": TYPES_MAP.get(type(row[key]), TYPE_STRING),\n                        }\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    should_annotate_query = False\n\n    @classmethod\n    def configuration_schema(cls):\n        return {\n            \"type\": \"object\",\n            \"properties\": {\n                \"connectionString\": {\"type\": \"string\", \"title\": \"Connection String\"},\n                \"username\": {\"type\": \"string\"},\n                \"password\": {\"type\": \"string\"},\n                \"dbName\": {\"type\": \"string\", \"title\": \"Database Name\"},\n                \"replicaSetName\": {\"type\": \"string\", \"title\": \"Replica Set Name\"},\n                \"readPreference\": {\n                    \"type\": \"string\",\n                    \"extendedEnum\": [\n                        {\"value\": \"primaryPreferred\", \"name\": \"Primary Preferred\"},\n                        {\"value\": \"primary\", \"name\": \"Primary\"},\n                        {\"value\": \"secondary\", \"name\": \"Secondary\"},\n                        {\"value\": \"secondaryPreferred\", \"name\": \"Secondary Preferred\"},\n                        {\"value\": \"nearest\", \"name\": \"Nearest\"},\n                    ],\n                    \"title\": \"Replica Set Read Preference\",\n                },\n            },\n            \"secret\": [\"password\"],\n            \"required\": [\"connectionString\", \"dbName\"],\n        }\n\n    @classmethod\n    def enabled(cls):\n        return enabled\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 = (\n            True\n            if \"replicaSetName\" in self.configuration\n            and self.configuration[\"replicaSetName\"]\n            else False\n        )\n\n    def _get_db(self):\n        kwargs = {}\n        if self.is_replica_set:\n            kwargs[\"replicaSet\"] = self.configuration[\"replicaSetName\"]\n            readPreference = self.configuration.get(\"readPreference\")\n            if readPreference:\n                kwargs[\"readPreference\"] = readPreference\n\n        if \"username\" in self.configuration:\n            kwargs[\"username\"] = self.configuration[\"username\"]\n\n        if \"password\" in self.configuration:\n            kwargs[\"password\"] = self.configuration[\"password\"]\n\n        db_connection = pymongo.MongoClient(\n            self.configuration[\"connectionString\"], **kwargs\n        )\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        return db\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        # fine. 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        try:\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        except Exception as ex:\n            template = \"An exception of type {0} occurred. Arguments:\\n{1!r}\"\n            message = template.format(type(ex).__name__, ex.args)\n            logger.error(message)\n            return []\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            if columns:\n                schema[collection_name] = {\n                    \"name\": collection_name,\n                    \"columns\": sorted(columns),\n                }\n\n        return list(schema.values())\n\n    def run_query(self, query, user):\n        db = self._get_db()\n\n        logger.debug(\n            \"mongodb connection string: %s\", self.configuration[\"connectionString\"]\n        )\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 \"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\", \"friendly_name\": \"count\", \"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 = {\"columns\": columns, \"rows\": rows}\n        error = None\n        json_data = json_dumps(data, cls=MongoDBJSONEncoder)\n\n        return json_data, error\n\n\nregister(MongoDB)\n", "path": "redash/query_runner/mongodb.py"}]}