PYTHON-1398 - Raw batch methods do not support sessions

Author: behackett

doc/api/pymongo/collection.rst

@@ -45,8 +45,10 @@
  .. automethod:: delete_one
  .. automethod:: delete_many
  .. automethod:: aggregate
+ .. automethod:: aggregate_raw_batches
  .. automethod:: watch
- .. automethod:: find(filter=None, projection=None, skip=0, limit=0, no_cursor_timeout=False, cursor_type=CursorType.NON_TAILABLE, sort=None, allow_partial_results=False, oplog_replay=False, modifiers=None, manipulate=True)
+ .. automethod:: find(filter=None, projection=None, skip=0, limit=0, no_cursor_timeout=False, cursor_type=CursorType.NON_TAILABLE, sort=None, allow_partial_results=False, oplog_replay=False, modifiers=None, batch_size=0, manipulate=True, collation=None, hint=None, max_scan=None, max_time_ms=None, max=None, min=None, return_key=False, show_record_id=False, snapshot=False, comment=None, session=None)
+ .. automethod:: find_raw_batches(filter=None, projection=None, skip=0, limit=0, no_cursor_timeout=False, cursor_type=CursorType.NON_TAILABLE, sort=None, allow_partial_results=False, oplog_replay=False, modifiers=None, batch_size=0, manipulate=True, collation=None, hint=None, max_scan=None, max_time_ms=None, max=None, min=None, return_key=False, show_record_id=False, snapshot=False, comment=None)
  .. automethod:: find_one(filter=None, *args, **kwargs)
  .. automethod:: find_one_and_delete
  .. automethod:: find_one_and_replace(filter, replacement, projection=None, sort=None, return_document=ReturnDocument.BEFORE, **kwargs)

pymongo/collection.py

@@ -1403,7 +1403,7 @@ def find(self, *args, **kwargs):
  def find_raw_batches(self, *args, **kwargs):
  """Query the database and retrieve batches of raw BSON.
 
- Takes the same parameters as :meth:`find` but returns a
+ Similar to the :meth:`find` method but returns a
  :class:`~pymongo.cursor.RawBatchCursor`.
 
  This example demonstrates how to work with raw batches, but in practice
@@ -1416,11 +1416,15 @@ def find_raw_batches(self, *args, **kwargs):
  >>> for batch in cursor:
  ... print(bson.decode_all(batch))
 
- Unlike most PyMongo methods, this method sends no session id to the
- server.
+ .. note:: find_raw_batches does not support sessions.
 
  .. versionadded:: 3.6
  """
+ # OP_MSG with document stream returns is required to support
+ # sessions.
+ if "session" in kwargs:
+ raise ConfigurationError(
+ "find_raw_batches does not support sessions")
  return RawBatchCursor(self, *args, **kwargs)
 
  def parallel_scan(self, num_cursors, session=None, **kwargs):
@@ -2181,7 +2185,7 @@ def aggregate(self, pipeline, session=None, **kwargs):
  def aggregate_raw_batches(self, pipeline, **kwargs):
  """Perform an aggregation and retrieve batches of raw BSON.
 
- Takes the same parameters as :meth:`aggregate` but returns a
+ Similar to the :meth:`aggregate` method but returns a
  :class:`~pymongo.cursor.RawBatchCursor`.
 
  This example demonstrates how to work with raw batches, but in practice
@@ -2195,11 +2199,15 @@ def aggregate_raw_batches(self, pipeline, **kwargs):
  >>> for batch in cursor:
  ... print(bson.decode_all(batch))
 
- Unlike most PyMongo methods, this method sends no session id to the
- server.
+ .. note:: aggregate_raw_batches does not support sessions.
 
  .. versionadded:: 3.6
  """
+ # OP_MSG with document stream returns is required to support
+ # sessions.
+ if "session" in kwargs:
+ raise ConfigurationError(
+ "aggregate_raw_batches does not support sessions")
  return self._aggregate(pipeline, RawBatchCommandCursor, 0,
  None, False, **kwargs)
 

test/test_session.py

@@ -337,8 +337,6 @@ def test_cursor(self):
  # Test all cursor methods.
  ops = [
  ('find', lambda session: list(coll.find(session=session))),
- ('find_raw_batches',
- lambda session: list(coll.find_raw_batches(session=session))),
  ('getitem', lambda session: coll.find(session=session)[0]),
  ('count', lambda session: coll.find(session=session).count()),
  ('distinct',
@@ -696,16 +694,8 @@ def test_reads(self):
  'ismaster', session=session))
  self._test_reads(
  lambda coll, session: list(coll.aggregate([], session=session)))
- # PYTHON-1398
- #self._test_reads(
- # lambda coll, session: list(
- # coll.aggregate_raw_batches([], session=session)))
  self._test_reads(
  lambda coll, session: list(coll.find({}, session=session)))
- # PYTHON-1398
- #self._test_reads(
- # lambda coll, session: list(
- # coll.find_raw_batches({}, session=session)))
  self._test_reads(
  lambda coll, session: coll.find_one({}, session=session))
  self._test_reads(
@@ -729,6 +719,17 @@ def test_reads(self):
  lambda coll, session: list(
  coll.parallel_scan(1, session=session)))
 
+ self.assertRaises(
+ ConfigurationError,
+ self._test_reads,
+ lambda coll, session: list(
+ coll.aggregate_raw_batches([], session=session)))
+ self.assertRaises(
+ ConfigurationError,
+ self._test_reads,
+ lambda coll, session: list(
+ coll.find_raw_batches({}, session=session)))
+
  def _test_writes(self, op):
  coll = self.client.pymongo_test.test
  with self.client.start_session() as sess: