PYTHON-3232 Improved change stream event visibility for C2C Replication (#1062)

Author: juliusgeo

doc/changelog.rst

@@ -22,6 +22,10 @@ PyMongo 4.3 brings a number of improvements including:
   deadlocks are still possible because libraries that PyMongo depends like
   OpenSSL cannot be made fork() safe in multithreaded applications.
   (`PYTHON-2484`_). For more info see :ref:`pymongo-fork-safe`.
+- When used with MongoDB 6.0+, :class:`~pymongo.change_stream.ChangeStream` s
+  now allow for new types of events (such as DDL and C2C replication events)
+  to be recorded with the new parameter ``show_expanded_events``
+  that can be passed to methods such as :meth:`~pymongo.collection.Collection.watch`.
 
 Bug fixes
 .........

pymongo/aggregation.py

@@ -40,6 +40,7 @@ def __init__(
         user_fields=None,
         result_processor=None,
         comment=None,
+        show_expanded_events=None,
     ):
         if "explain" in options:
             raise ConfigurationError(
@@ -60,6 +61,7 @@ def __init__(
             options["let"] = let
         if comment is not None:
             options["comment"] = comment
+
         self._options = options
 
         # This is the batchSize that will be used for setting the initial

pymongo/change_stream.py

@@ -109,6 +109,7 @@ def __init__(
         start_after: Optional[Mapping[str, Any]],
         comment: Optional[Any] = None,
         full_document_before_change: Optional[str] = None,
+        show_expanded_events: Optional[bool] = None,
     ) -> None:
         if pipeline is None:
             pipeline = []
@@ -143,6 +144,7 @@ def __init__(
         self._comment = comment
         self._closed = False
         self._timeout = self._target._timeout
+        self._show_expanded_events = show_expanded_events
         # Initialize cursor.
         self._cursor = self._create_cursor()
 
@@ -175,6 +177,10 @@ def _change_stream_options(self):
 
         if self._start_at_operation_time is not None:
             options["startAtOperationTime"] = self._start_at_operation_time
+
+        if self._show_expanded_events:
+            options["showExpandedEvents"] = self._show_expanded_events
+
         return options
 
     def _command_options(self):
@@ -230,6 +236,7 @@ def _run_aggregation_cmd(self, session, explicit_session):
             explicit_session,
             result_processor=self._process_result,
             comment=self._comment,
+            show_expanded_events=self._show_expanded_events,
         )
         return self._client._retryable_read(
             cmd.get_cursor, self._target._read_preference_for(session), session

pymongo/collection.py

@@ -2495,6 +2495,7 @@ def watch(
         start_after: Optional[Mapping[str, Any]] = None,
         comment: Optional[Any] = None,
         full_document_before_change: Optional[str] = None,
+        show_expanded_events: Optional[bool] = None,
     ) -> CollectionChangeStream[_DocumentType]:
         """Watch changes on this collection.
 
@@ -2579,12 +2580,16 @@ def watch(
             This option and `resume_after` are mutually exclusive.
           - `comment` (optional): A user-provided comment to attach to this
             command.
+          - `show_expanded_events` (optional): Include expanded events such as DDL events like `dropIndexes`.
 
         :Returns:
           A :class:`~pymongo.change_stream.CollectionChangeStream` cursor.
 
+        .. versionchanged:: 4.3
+           Added `show_expanded_events` parameter.
+
         .. versionchanged:: 4.2
-            Added ``full_document_before_change`` parameter.
+           Added ``full_document_before_change`` parameter.
 
         .. versionchanged:: 4.1
            Added ``comment`` parameter.
@@ -2615,6 +2620,7 @@ def watch(
             start_after,
             comment,
             full_document_before_change,
+            show_expanded_events,
         )
 
     @_csot.apply

pymongo/database.py

@@ -547,6 +547,7 @@ def watch(
         start_after: Optional[Mapping[str, Any]] = None,
         comment: Optional[Any] = None,
         full_document_before_change: Optional[str] = None,
+        show_expanded_events: Optional[bool] = None,
     ) -> DatabaseChangeStream[_DocumentType]:
         """Watch changes on this database.
 
@@ -624,10 +625,14 @@ def watch(
             This option and `resume_after` are mutually exclusive.
           - `comment` (optional): A user-provided comment to attach to this
             command.
+          - `show_expanded_events` (optional): Include expanded events such as DDL events like `dropIndexes`.
 
         :Returns:
           A :class:`~pymongo.change_stream.DatabaseChangeStream` cursor.
 
+        .. versionchanged:: 4.3
+           Added `show_expanded_events` parameter.
+
         .. versionchanged:: 4.2
             Added ``full_document_before_change`` parameter.
 
@@ -657,6 +662,7 @@ def watch(
             start_after,
             comment,
             full_document_before_change,
+            show_expanded_events=show_expanded_events,
         )
 
     def _command(

pymongo/mongo_client.py

@@ -895,6 +895,7 @@ def watch(
         start_after: Optional[Mapping[str, Any]] = None,
         comment: Optional[Any] = None,
         full_document_before_change: Optional[str] = None,
+        show_expanded_events: Optional[bool] = None,
     ) -> ChangeStream[_DocumentType]:
         """Watch changes on this cluster.
 
@@ -972,10 +973,14 @@ def watch(
             This option and `resume_after` are mutually exclusive.
           - `comment` (optional): A user-provided comment to attach to this
             command.
+          - `show_expanded_events` (optional): Include expanded events such as DDL events like `dropIndexes`.
 
         :Returns:
           A :class:`~pymongo.change_stream.ClusterChangeStream` cursor.
 
+        .. versionchanged:: 4.3
+           Added `show_expanded_events` parameter.
+
         .. versionchanged:: 4.2
             Added ``full_document_before_change`` parameter.
 
@@ -1005,6 +1010,7 @@ def watch(
             start_after,
             comment,
             full_document_before_change,
+            show_expanded_events=show_expanded_events,
         )
 
     @property