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):