PYTHON-981 - Implement ReadConcern.

Author: Luke Lovett

pymongo/client_options.py

@@ -21,6 +21,7 @@
 from pymongo.errors import ConfigurationError
 from pymongo.monitoring import _EventListeners
 from pymongo.pool import PoolOptions
+from pymongo.read_concern import ReadConcern
 from pymongo.read_preferences import make_read_preference
 from pymongo.ssl_support import get_ssl_context
 from pymongo.write_concern import WriteConcern
@@ -55,6 +56,12 @@ def _parse_write_concern(options):
  return WriteConcern(concern, wtimeout, j, fsync)
 
 
+def _parse_read_concern(options):
+ """Parse read concern options."""
+ concern = options.get('readconcernlevel')
+ return ReadConcern(concern)
+
+
 def _parse_ssl_options(options):
  """Parse ssl options."""
  use_ssl = options.get('ssl')
@@ -122,6 +129,7 @@ def __init__(self, username, password, database, options):
  self.__read_preference = _parse_read_preference(options)
  self.__replica_set_name = options.get('replicaset')
  self.__write_concern = _parse_write_concern(options)
+ self.__read_concern = _parse_read_concern(options)
  self.__connect = options.get('connect')
 
  @property
@@ -173,3 +181,8 @@ def replica_set_name(self):
  def write_concern(self):
  """A :class:`~pymongo.write_concern.WriteConcern` instance."""
  return self.__write_concern
+
+ @property
+ def read_concern(self):
+ """A :class:`~pymongo.read_concern.ReadConcern` instance."""
+ return self.__read_concern

pymongo/collection.py

@@ -35,6 +35,7 @@
 from pymongo.errors import ConfigurationError, InvalidName, OperationFailure
 from pymongo.helpers import _check_write_command_response
 from pymongo.operations import _WriteOp, IndexModel
+from pymongo.read_concern import DEFAULT_READ_CONCERN
 from pymongo.read_preferences import ReadPreference
 from pymongo.results import (BulkWriteResult,
  DeleteResult,
@@ -71,7 +72,8 @@ class Collection(common.BaseObject):
  """
 
  def __init__(self, database, name, create=False, codec_options=None,
- read_preference=None, write_concern=None, **kwargs):
+ read_preference=None, write_concern=None, read_concern=None,
+ **kwargs):
  """Get / create a Mongo collection.
 
  Raises :class:`TypeError` if `name` is not an instance of
@@ -100,9 +102,15 @@ def __init__(self, database, name, create=False, codec_options=None,
  - `write_concern` (optional): An instance of
  :class:`~pymongo.write_concern.WriteConcern`. If ``None`` (the
  default) database.write_concern is used.
+ - `read_concern` (optional): An instance of
+ :class:`~pymongo.read_concern.ReadConcern`. If ``None`` (the
+ default) database.read_concern is used.
  - `**kwargs` (optional): additional keyword arguments will
  be passed as options for the create collection command
 
+ .. versionchanged:: 3.2
+ Added the read_concern option.
+
  .. versionchanged:: 3.0
  Added the codec_options, read_preference, and write_concern options.
  Removed the uuid_subtype attribute.
@@ -128,7 +136,8 @@ def __init__(self, database, name, create=False, codec_options=None,
  super(Collection, self).__init__(
  codec_options or database.codec_options,
  read_preference or database.read_preference,
- write_concern or database.write_concern)
+ write_concern or database.write_concern,
+ read_concern or database.read_concern)
 
  if not isinstance(name, string_type):
  raise TypeError("name must be an instance "
@@ -164,7 +173,8 @@ def _socket_for_writes(self):
 
  def _command(self, sock_info, command, slave_ok=False,
  read_preference=None,
- codec_options=None, check=True, allowable_errors=None):
+ codec_options=None, check=True, allowable_errors=None,
+ read_concern=DEFAULT_READ_CONCERN):
  """Internal command helper.
 
  :Parameters:
@@ -175,6 +185,8 @@ def _command(self, sock_info, command, slave_ok=False,
  :class:`~bson.codec_options.CodecOptions`.
  - `check`: raise OperationFailure if there are errors
  - `allowable_errors`: errors to ignore if `check` is True
+ - `read_concern` (optional) - An instance of
+ :class:`~pymongo.read_concern.ReadConcern`.
 
  :Returns:
 
@@ -188,7 +200,8 @@ def _command(self, sock_info, command, slave_ok=False,
  read_preference or self.read_preference,
  codec_options or self.codec_options,
  check,
- allowable_errors)
+ allowable_errors,
+ read_concern=read_concern)
 
  def __create(self, options):
  """Sends a create command with the given options.
@@ -254,7 +267,8 @@ def database(self):
  return self.__database
 
  def with_options(
- self, codec_options=None, read_preference=None, write_concern=None):
+ 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
@@ -279,13 +293,18 @@ def with_options(
  :class:`~pymongo.write_concern.WriteConcern`. If ``None`` (the
  default) the :attr:`write_concern` of this :class:`Collection`
  is used.
+ - `read_concern` (optional): An instance of
+ :class:`~pymongo.read_concern.ReadConcern`. If ``None`` (the
+ default) the :attr:`read_concern` of this :class:`Collection`
+ is used.
  """
  return Collection(self.__database,
  self.__name,
  False,
  codec_options or self.codec_options,
  read_preference or self.read_preference,
- write_concern or self.write_concern)
+ write_concern or self.write_concern,
+ read_concern or self.read_concern)
 
  def initialize_unordered_bulk_op(self):
  """Initialize an unordered batch of write operations.
@@ -1059,7 +1078,8 @@ def parallel_scan(self, num_cursors):
  ('numCursors', num_cursors)])
 
  with self._socket_for_reads() as (sock_info, slave_ok):
- result = self._command(sock_info, cmd, slave_ok)
+ result = self._command(sock_info, cmd, slave_ok,
+ read_concern=self.read_concern)
 
  return [CommandCursor(self, cursor['cursor'], sock_info.address)
  for cursor in result['cursors']]
@@ -1068,7 +1088,8 @@ def _count(self, cmd):
  """Internal count helper."""
  with self._socket_for_reads() as (sock_info, slave_ok):
  res = self._command(sock_info, cmd, slave_ok,
- allowable_errors=["ns missing"])
+ allowable_errors=["ns missing"],
+ read_concern=self.read_concern)
  if res.get("errmsg", "") == "ns missing":
  return 0
  return int(res["n"])
@@ -1522,7 +1543,8 @@ def aggregate(self, pipeline, **kwargs):
 
  cmd.update(kwargs)
 
- result = self._command(sock_info, cmd, slave_ok)
+ result = self._command(sock_info, cmd, slave_ok,
+ read_concern=self.read_concern)
 
  if "cursor" in result:
  cursor = result["cursor"]
@@ -1653,7 +1675,8 @@ def distinct(self, key, filter=None, **kwargs):
  kwargs["query"] = filter
  cmd.update(kwargs)
  with self._socket_for_reads() as (sock_info, slave_ok):
- return self._command(sock_info, cmd, slave_ok)["values"]
+ return self._command(sock_info, cmd, slave_ok,
+ read_concern=self.read_concern)["values"]
 
  def map_reduce(self, map, reduce, out, full_response=False, **kwargs):
  """Perform a map/reduce operation on this collection.

pymongo/common.py

@@ -25,6 +25,7 @@
 from pymongo.auth import MECHANISMS
 from pymongo.errors import ConfigurationError
 from pymongo.monitoring import _validate_event_listeners
+from pymongo.read_concern import ReadConcern
 from pymongo.read_preferences import (read_pref_mode_from_name,
  _ServerMode)
 from pymongo.ssl_support import validate_cert_reqs
@@ -423,6 +424,7 @@ def validate_ok_for_update(update):
  'ssl_cert_reqs': validate_cert_reqs,
  'ssl_ca_certs': validate_readable,
  'ssl_match_hostname': validate_boolean_or_string,
+ 'readconcernlevel': validate_string_or_none,
  'read_preference': validate_read_preference,
  'readpreference': validate_read_preference_mode,
  'readpreferencetags': validate_read_preference_tags,
@@ -495,7 +497,8 @@ class BaseObject(object):
  SHOULD NOT BE USED BY DEVELOPERS EXTERNAL TO MONGODB.
  """
 
- def __init__(self, codec_options, read_preference, write_concern):
+ def __init__(self, codec_options, read_preference, write_concern,
+ read_concern):
 
  if not isinstance(codec_options, CodecOptions):
  raise TypeError("codec_options must be an instance of "
@@ -513,6 +516,11 @@ def __init__(self, codec_options, read_preference, write_concern):
  "pymongo.write_concern.WriteConcern")
  self.__write_concern = write_concern
 
+ if not isinstance(read_concern, ReadConcern):
+ raise TypeError("read_concern must be an instance of "
+ "pymongo.read_concern.ReadConcern")
+ self.__read_concern = read_concern
+
  @property
  def codec_options(self):
  """Read only access to the :class:`~bson.codec_options.CodecOptions`
@@ -538,3 +546,11 @@ def read_preference(self):
  The :attr:`read_preference` attribute is now read only.
  """
  return self.__read_preference
+
+ @property
+ def read_concern(self):
+ """Read only access to the read concern of this instance.
+
+ .. versionadded:: 3.2
+ """
+ return self.__read_concern

pymongo/cursor.py

@@ -184,6 +184,7 @@ def __init__(self, collection, filter=None, projection=None, skip=0,
 
  self.__codec_options = collection.codec_options
  self.__read_preference = collection.read_preference
+ self.__read_concern = collection.read_concern
 
  self.__query_flags = cursor_type
  if self.__read_preference != ReadPreference.PRIMARY:
@@ -985,7 +986,8 @@ def _refresh(self):
  self.__codec_options,
  self.__read_preference,
  self.__limit,
- self.__batch_size))
+ self.__batch_size,
+ self.__read_concern))
  if not self.__id:
  self.__killed = True
  elif self.__id: # Get More

pymongo/database.py

@@ -17,7 +17,7 @@
 import warnings
 
 from bson.code import Code
-from bson.codec_options import CodecOptions
+from bson.codec_options import CodecOptions, DEFAULT_CODEC_OPTIONS
 from bson.dbref import DBRef
 from bson.objectid import ObjectId
 from bson.py3compat import iteritems, string_type, _unicode
@@ -51,8 +51,8 @@ class Database(common.BaseObject):
  """A Mongo database.
  """
 
- def __init__(self, client, name, codec_options=None,
- read_preference=None, write_concern=None):
+ def __init__(self, client, name, codec_options=None, read_preference=None,
+ write_concern=None, read_concern=None):
  """Get a database by client and name.
 
  Raises :class:`TypeError` if `name` is not an instance of
@@ -71,9 +71,15 @@ def __init__(self, client, name, codec_options=None,
  - `write_concern` (optional): An instance of
  :class:`~pymongo.write_concern.WriteConcern`. If ``None`` (the
  default) client.write_concern is used.
+ - `read_concern` (optional): An instance of
+ :class:`~pymongo.read_concern.ReadConcern`. If ``None`` (the
+ default) client.read_concern is used.
 
  .. mongodoc:: databases
 
+ .. versionchanged:: 3.2
+ Added the read_concern option.
+
  .. versionchanged:: 3.0
  Added the codec_options, read_preference, and write_concern options.
  :class:`~pymongo.database.Database` no longer returns an instance
@@ -89,7 +95,8 @@ def __init__(self, client, name, codec_options=None,
  super(Database, self).__init__(
  codec_options or client.codec_options,
  read_preference or client.read_preference,
- write_concern or client.write_concern)
+ write_concern or client.write_concern,
+ read_concern or client.read_concern)
 
  if not isinstance(name, string_type):
  raise TypeError("name must be an instance "
@@ -225,8 +232,8 @@ def __getitem__(self, name):
  """
  return Collection(self, name)
 
- def get_collection(self, name, codec_options=None,
- read_preference=None, write_concern=None):
+ def get_collection(self, name, codec_options=None, read_preference=None,
+ write_concern=None, read_concern=None):
  """Get a :class:`~pymongo.collection.Collection` with the given name
  and options.
 
@@ -259,12 +266,18 @@ def get_collection(self, name, codec_options=None,
  :class:`~pymongo.write_concern.WriteConcern`. If ``None`` (the
  default) the :attr:`write_concern` of this :class:`Database` is
  used.
+ - `read_concern` (optional): An instance of
+ :class:`~pymongo.read_concern.ReadConcern`. If ``None`` (the
+ default) the :attr:`read_concern` of this :class:`Database` is
+ used.
  """
  return Collection(
- self, name, False, codec_options, read_preference, write_concern)
+ self, name, False, codec_options, read_preference,
+ write_concern, read_concern)
 
  def create_collection(self, name, codec_options=None,
- read_preference=None, write_concern=None, **kwargs):
+ read_preference=None, write_concern=None,
+ read_concern=None, **kwargs):
  """Create a new :class:`~pymongo.collection.Collection` in this
  database.
 
@@ -298,6 +311,10 @@ def create_collection(self, name, codec_options=None,
  :class:`~pymongo.write_concern.WriteConcern`. If ``None`` (the
  default) the :attr:`write_concern` of this :class:`Database` is
  used.
+ - `read_concern` (optional): An instance of
+ :class:`~pymongo.read_concern.ReadConcern`. If ``None`` (the
+ default) the :attr:`read_concern` of this :class:`Database` is
+ used.
  - `**kwargs` (optional): additional keyword arguments will
  be passed as options for the create collection command
 
@@ -311,7 +328,8 @@ def create_collection(self, name, codec_options=None,
  raise CollectionInvalid("collection %s already exists" % name)
 
  return Collection(self, name, True, codec_options,
- read_preference, write_concern, **kwargs)
+ read_preference, write_concern,
+ read_concern, **kwargs)
 
  def _apply_incoming_manipulators(self, son, collection):
  """Apply incoming manipulators to `son`."""
@@ -351,7 +369,7 @@ def _fix_outgoing(self, son, collection):
 
  def _command(self, sock_info, command, slave_ok=False, value=1, check=True,
  allowable_errors=None, read_preference=ReadPreference.PRIMARY,
- codec_options=CodecOptions(), **kwargs):
+ codec_options=DEFAULT_CODEC_OPTIONS, **kwargs):
  """Internal command helper."""
  if isinstance(command, string_type):
  command = SON([(command, value)])
@@ -367,7 +385,7 @@ def _command(self, sock_info, command, slave_ok=False, value=1, check=True,
 
  def command(self, command, value=1, check=True,
  allowable_errors=None, read_preference=ReadPreference.PRIMARY,
- codec_options=CodecOptions(), **kwargs):
+ codec_options=DEFAULT_CODEC_OPTIONS, **kwargs):
  """Issue a MongoDB command.
 
  Send command `command` to the database and return the

pymongo/helpers.py

@@ -32,6 +32,7 @@
  WriteConcernError,
  WTimeoutError)
 from pymongo.message import _Query, _convert_exception
+from pymongo.read_concern import DEFAULT_READ_CONCERN
 
 
 _UUNDER = u("_")
@@ -240,7 +241,7 @@ def _first_batch(sock_info, db, coll, query, ntoreturn,
  """Simple query helper for retrieving a first (and possibly only) batch."""
  query = _Query(
  0, db, coll, 0, ntoreturn, query, None,
- codec_options, read_preference, 0, ntoreturn)
+ codec_options, read_preference, 0, 0, DEFAULT_READ_CONCERN)
 
  name = next(iter(cmd))
  duration = None