PYTHON-1783 Decode user-facing documents but not internal driver-server

communications.

Author: prashantmital

bson/__init__.py

@@ -215,7 +215,7 @@ def _get_array(data, position, obj_end, opts, element_name):
     end -= 1
     result = []
 
-    # Avoid doing global and attibute lookups in the loop.
+    # Avoid doing global and attribute lookups in the loop.
     append = result.append
     index = data.index
     getter = _ELEMENT_GETTER
@@ -940,6 +940,59 @@ def decode_all(data, codec_options=DEFAULT_CODEC_OPTIONS):
     decode_all = _cbson.decode_all
 
 
+def _decode_selective(rawdoc, fields, codec_options):
+    doc = codec_options.document_class()
+    for key, value in iteritems(rawdoc):
+        if key in fields:
+            if fields[key] == list:
+                doc[key] = [_bson_to_dict(r.raw, codec_options) for r in value]
+            elif fields[key] == dict:
+                doc[key] = _bson_to_dict(value.raw, codec_options)
+            else:
+                doc[key] = _decode_selective(value, fields[key], codec_options)
+            continue
+        doc[key] = value
+    return doc
+
+
+def _decode_all_selective(data, codec_options, fields):
+    """Decode BSON data to a single document while using user-provided
+    custom decoding logic.
+
+    `data` must be a string representing a valid, BSON-encoded document.
+
+    :Parameters:
+      - `data`: BSON data
+      - `codec_options`: An instance of
+        :class:`~bson.codec_options.CodecOptions` with user-specified type
+        decoders. If no decoders are found, this method is the same as
+        ``decode_all``.
+      - `fields`: Map of document namespaces where data that needs
+        to be custom decoded lives or None. For example, to custom decode a
+        list of objects in 'field1.subfield1', the specified value should be
+        ``{'field1': {'subfield1': list}}``. Use ``dict`` instead of ``list``
+        if the field contains a single object to custom decode. If ``fields``
+        is an empty map or None, this method is the same as ``decode_all``.
+
+    :Returns:
+      - `document_list`: Single-member list containing the decoded document.
+
+    .. versionadded:: 3.8
+    """
+    if not codec_options.type_registry._decoder_map:
+        return decode_all(data, codec_options)
+
+    if not fields:
+        return decode_all(data, codec_options.with_options(type_registry=None))
+
+    # Decode documents for internal use.
+    from bson.raw_bson import RawBSONDocument
+    internal_codec_options = codec_options.with_options(
+        document_class=RawBSONDocument, type_registry=None)
+    _doc = _bson_to_dict(data, internal_codec_options)
+    return [_decode_selective(_doc, fields, codec_options,)]
+
+
 def decode_iter(data, codec_options=DEFAULT_CODEC_OPTIONS):
     """Decode BSON data to multiple documents as a generator.
 

gridfs/__init__.py

@@ -25,7 +25,8 @@
 from gridfs.grid_file import (GridIn,
                               GridOut,
                               GridOutCursor,
-                              DEFAULT_CHUNK_SIZE)
+                              DEFAULT_CHUNK_SIZE,
+                              _clear_entity_type_registry)
 from pymongo import (ASCENDING,
                      DESCENDING)
 from pymongo.common import UNAUTHORIZED_CODES, validate_string
@@ -61,6 +62,8 @@ def __init__(self, database, collection="fs", disable_md5=False):
         if not isinstance(database, Database):
             raise TypeError("database must be an instance of Database")
 
+        database = _clear_entity_type_registry(database)
+
         if not database.write_concern.acknowledged:
             raise ConfigurationError('database must use '
                                      'acknowledged write_concern')
@@ -443,6 +446,8 @@ def __init__(self, db, bucket_name="fs",
         if not isinstance(db, Database):
             raise TypeError("database must be an instance of Database")
 
+        db = _clear_entity_type_registry(db)
+
         wtc = write_concern if write_concern is not None else db.write_concern
         if not wtc.acknowledged:
             raise ConfigurationError('write concern must be acknowledged')

gridfs/grid_file.py

@@ -98,6 +98,12 @@ def getter(self):
     return property(getter, doc=docstring)
 
 
+def _clear_entity_type_registry(entity, **kwargs):
+    """Clear the given database/collection object's type registry."""
+    codecopts = entity.codec_options.with_options(type_registry=None)
+    return entity.with_options(codec_options=codecopts, **kwargs)
+
+
 class GridIn(object):
     """Class to write data to GridFS.
     """
@@ -168,8 +174,8 @@ def __init__(
         if "chunk_size" in kwargs:
             kwargs["chunkSize"] = kwargs.pop("chunk_size")
 
-        coll = root_collection.with_options(
-            read_preference=ReadPreference.PRIMARY)
+        coll = _clear_entity_type_registry(
+            root_collection, read_preference=ReadPreference.PRIMARY)
 
         if not disable_md5:
             kwargs["md5"] = hashlib.md5()
@@ -449,6 +455,8 @@ def __init__(self, root_collection, file_id=None, file_document=None,
             raise TypeError("root_collection must be an "
                             "instance of Collection")
 
+        root_collection = _clear_entity_type_registry(root_collection)
+
         self.__chunks = root_collection.chunks
         self.__files = root_collection.files
         self.__file_id = file_id
@@ -800,6 +808,8 @@ def __init__(self, collection, filter=None, skip=0, limit=0,
 
         .. mongodoc:: cursors
         """
+        collection = _clear_entity_type_registry(collection)
+
         # Hold on to the base "fs" collection to create GridOut objects later.
         self.__root_collection = collection
 

pymongo/collection.py

@@ -53,6 +53,7 @@
 
 _NO_OBJ_ERROR = "No matching object found"
 _UJOIN = u"%s.%s"
+_FIND_AND_MODIFY_DOC_FIELDS = {'value': dict}
 
 
 class ReturnDocument(object):
@@ -203,7 +204,8 @@ def _command(self, sock_info, command, slave_ok=False,
                  write_concern=None,
                  collation=None,
                  session=None,
-                 retryable_write=False):
+                 retryable_write=False,
+                 user_fields=None):
         """Internal command helper.
 
         :Parameters:
@@ -242,7 +244,8 @@ def _command(self, sock_info, command, slave_ok=False,
                 collation=collation,
                 session=s,
                 client=self.__database.client,
-                retryable_write=retryable_write)
+                retryable_write=retryable_write,
+                user_fields=user_fields)
 
     def __create(self, options, collation, session):
         """Sends a create command with the given options.
@@ -315,9 +318,8 @@ def database(self):
         """
         return self.__database
 
-    def with_options(
-            self, codec_options=None, read_preference=None,
-            write_concern=None, read_concern=None):
+    def with_options(self, codec_options=None, read_preference=None,
+                     write_concern=None, read_concern=None):
         """Get a clone of this collection changing the specified settings.
 
           >>> coll1.read_preference
@@ -2310,7 +2312,8 @@ def _aggregate(self, pipeline, cursor_class, first_batch_size, session,
                 write_concern=write_concern,
                 collation=collation,
                 session=session,
-                client=self.__database.client)
+                client=self.__database.client,
+                user_fields={'cursor': {'firstBatch': list}})
 
             if "cursor" in result:
                 cursor = result["cursor"]
@@ -2571,7 +2574,8 @@ def group(self, key, condition, initial, reduce, finalize=None, **kwargs):
 
         with self._socket_for_reads(session=None) as (sock_info, slave_ok):
             return self._command(sock_info, cmd, slave_ok,
-                                 collation=collation)["retval"]
+                                 collation=collation,
+                                 user_fields={'retval': list})["retval"]
 
     def rename(self, new_name, session=None, **kwargs):
         """Rename this collection.
@@ -2675,7 +2679,8 @@ def distinct(self, key, filter=None, session=None, **kwargs):
         with self._socket_for_reads(session) as (sock_info, slave_ok):
             return self._command(sock_info, cmd, slave_ok,
                                  read_concern=self.read_concern,
-                                 collation=collation, session=session)["values"]
+                                 collation=collation,
+                                 session=session)["values"]
 
     def map_reduce(self, map, reduce, out, full_response=False, session=None,
                    **kwargs):
@@ -2755,12 +2760,17 @@ def map_reduce(self, map, reduce, out, full_response=False, session=None,
                 write_concern = self._write_concern_for(session)
             else:
                 write_concern = None
+            if inline:
+                user_fields = {'results': list}
+            else:
+                user_fields = None
 
             response = self._command(
                 sock_info, cmd, slave_ok, read_pref,
                 read_concern=read_concern,
                 write_concern=write_concern,
-                collation=collation, session=session)
+                collation=collation, session=session,
+                user_fields=user_fields)
 
         if full_response or not response.get('result'):
             return response
@@ -2810,16 +2820,19 @@ def inline_map_reduce(self, map, reduce, full_response=False, session=None,
                    ("map", map),
                    ("reduce", reduce),
                    ("out", {"inline": 1})])
+        user_fields = {'results': list}
         collation = validate_collation_or_none(kwargs.pop('collation', None))
         cmd.update(kwargs)
         with self._socket_for_reads(session) as (sock_info, slave_ok):
             if sock_info.max_wire_version >= 4 and 'readConcern' not in cmd:
                 res = self._command(sock_info, cmd, slave_ok,
                                     read_concern=self.read_concern,
-                                    collation=collation, session=session)
+                                    collation=collation, session=session,
+                                    user_fields=user_fields)
             else:
                 res = self._command(sock_info, cmd, slave_ok,
-                                    collation=collation, session=session)
+                                    collation=collation, session=session,
+                                    user_fields=user_fields)
 
         if full_response:
             return res
@@ -2837,6 +2850,7 @@ def __find_and_modify(self, filter, projection, sort, upsert=None,
                           return_document=ReturnDocument.BEFORE,
                           array_filters=None, session=None, **kwargs):
         """Internal findAndModify helper."""
+
         common.validate_is_mapping("filter", filter)
         if not isinstance(return_document, bool):
             raise ValueError("return_document must be "
@@ -2876,8 +2890,10 @@ def _find_and_modify(session, sock_info, retryable_write):
                                 write_concern=write_concern,
                                 allowable_errors=[_NO_OBJ_ERROR],
                                 collation=collation, session=session,
-                                retryable_write=retryable_write)
+                                retryable_write=retryable_write,
+                                user_fields=_FIND_AND_MODIFY_DOC_FIELDS)
             _check_write_command_response(out)
+
             return out.get("value")
 
         return self.__database.client._retryable_write(
@@ -3293,7 +3309,8 @@ def _find_and_modify(session, sock_info, retryable_write):
             result = self._command(
                 sock_info, cmd, read_preference=ReadPreference.PRIMARY,
                 allowable_errors=[_NO_OBJ_ERROR], collation=collation,
-                session=session, retryable_write=retryable_write)
+                session=session, retryable_write=retryable_write,
+                user_fields=_FIND_AND_MODIFY_DOC_FIELDS)
 
             _check_write_command_response(result)
             return result

pymongo/command_cursor.py

@@ -150,14 +150,18 @@ def duration(): return datetime.datetime.now() - start
 
         try:
             with client._reset_on_error(self.__address, self.__session):
-                docs = self._unpack_response(reply,
-                                             self.__id,
-                                             self.__collection.codec_options)
+                user_fields = None
+                legacy_response = True
+                if from_command:
+                    user_fields = {'cursor': {'nextBatch': list}}
+                    legacy_response = False
+                docs = self._unpack_response(
+                    reply, self.__id, self.__collection.codec_options,
+                    legacy_response=legacy_response, user_fields=user_fields)
                 if from_command:
                     first = docs[0]
                     client._process_response(first, self.__session)
                     helpers._check_command_response(first)
-
         except OperationFailure as exc:
             kill()
 
@@ -208,8 +212,10 @@ def duration(): return datetime.datetime.now() - start
             kill()
         self.__data = deque(documents)
 
-    def _unpack_response(self, response, cursor_id, codec_options):
-        return response.unpack_response(cursor_id, codec_options)
+    def _unpack_response(self, response, cursor_id, codec_options,
+                         user_fields=None, legacy_response=False):
+        return response.unpack_response(cursor_id, codec_options, user_fields,
+                                        legacy_response)
 
     def _refresh(self):
         """Refreshes the cursor with more data from the server.
@@ -330,7 +336,8 @@ def __init__(self, collection, cursor_info, address, retrieved=0,
             collection, cursor_info, address, retrieved, batch_size,
             max_await_time_ms, session, explicit_session)
 
-    def _unpack_response(self, response, cursor_id, codec_options):
+    def _unpack_response(self, response, cursor_id, codec_options,
+                         user_fields=None, legacy_response=False):
         return response.raw_response(cursor_id)
 
     def __getitem__(self, index):

pymongo/cursor.py

@@ -40,7 +40,7 @@
                              _RawBatchGetMore,
                              _Query,
                              _RawBatchQuery)
-from pymongo.read_preferences import ReadPreference
+
 
 _QUERY_OPTIONS = {
     "tailable_cursor": 2,
@@ -50,6 +50,7 @@
     "await_data": 32,
     "exhaust": 64,
     "partial": 128}
+_CURSOR_DOC_FIELDS = {'cursor': {'firstBatch': list, 'nextBatch': list}}
 
 
 class CursorType(object):
@@ -996,9 +997,14 @@ def duration(): return datetime.datetime.now() - start
 
         try:
             with client._reset_on_error(self.__address, self.__session):
-                docs = self._unpack_response(reply,
-                                             self.__id,
-                                             self.__collection.codec_options)
+                user_fields = None
+                legacy_response = True
+                if from_command:
+                    user_fields = _CURSOR_DOC_FIELDS
+                    legacy_response = False
+                docs = self._unpack_response(
+                    reply, self.__id, self.__collection.codec_options,
+                    legacy_response=legacy_response, user_fields=user_fields)
                 if from_command:
                     first = docs[0]
                     client._process_response(first, self.__session)
@@ -1085,8 +1091,10 @@ def duration(): return datetime.datetime.now() - start
         if self.__limit and self.__id and self.__limit <= self.__retrieved:
             self.__die()
 
-    def _unpack_response(self, response, cursor_id, codec_options):
-        return response.unpack_response(cursor_id, codec_options)
+    def _unpack_response(self, response, cursor_id, codec_options,
+                         user_fields=None, legacy_response=False):
+        return response.unpack_response(cursor_id, codec_options, user_fields,
+                                        legacy_response)
 
     def _read_preference(self):
         if self.__read_preference is None:
@@ -1303,7 +1311,8 @@ def __init__(self, *args, **kwargs):
             raise InvalidOperation(
                 "Cannot use RawBatchCursor with manipulate=True")
 
-    def _unpack_response(self, response, cursor_id, codec_options):
+    def _unpack_response(self, response, cursor_id, codec_options,
+                         user_fields=None, legacy_response=False):
         return response.raw_response(cursor_id)
 
     def explain(self):