PYTHON-3093 Change streams support for user-facing PIT pre- and post-images (#972)

Author: bcwarner

pymongo/change_stream.py

@@ -96,6 +96,7 @@ def __init__(
         session: Optional["ClientSession"],
         start_after: Optional[Mapping[str, Any]],
         comment: Optional[Any] = None,
+        full_document_before_change: Optional[str] = None,
     ) -> None:
         if pipeline is None:
             pipeline = []
@@ -118,6 +119,7 @@ def __init__(
 
         self._pipeline = copy.deepcopy(pipeline)
         self._full_document = full_document
+        self._full_document_before_change = full_document_before_change
         self._uses_start_after = start_after is not None
         self._uses_resume_after = resume_after is not None
         self._resume_token = copy.deepcopy(start_after or resume_after)
@@ -147,6 +149,9 @@ def _change_stream_options(self):
         if self._full_document is not None:
             options["fullDocument"] = self._full_document
 
+        if self._full_document_before_change is not None:
+            options["fullDocumentBeforeChange"] = self._full_document_before_change
+
         resume_token = self.resume_token
         if resume_token is not None:
             if self._uses_start_after:

pymongo/collection.py

@@ -2501,6 +2501,7 @@ def watch(
         session: Optional["ClientSession"] = None,
         start_after: Optional[Mapping[str, Any]] = None,
         comment: Optional[Any] = None,
+        full_document_before_change: Optional[str] = None,
     ) -> CollectionChangeStream[_DocumentType]:
         """Watch changes on this collection.
 
@@ -2559,6 +2560,8 @@ def watch(
             updates will include both a delta describing the changes to the
             document, as well as a copy of the entire document that was
             changed from some time after the change occurred.
+          - `full_document_before_change`: Allowed values: `whenAvailable` and `required`. Change events
+            may now result in a `fullDocumentBeforeChange` response field.
           - `resume_after` (optional): A resume token. If provided, the
             change stream will start returning changes that occur directly
             after the operation specified in the resume token. A resume token
@@ -2585,6 +2588,8 @@ def watch(
         :Returns:
           A :class:`~pymongo.change_stream.CollectionChangeStream` cursor.
 
+        .. versionchanged:: 4.2
+            Added ``full_document_before_change`` parameter.
 
         .. versionchanged:: 4.1
            Added ``comment`` parameter.
@@ -2613,7 +2618,8 @@ def watch(
             start_at_operation_time,
             session,
             start_after,
-            comment=comment,
+            comment,
+            full_document_before_change,
         )
 
     def rename(

pymongo/database.py

@@ -389,6 +389,8 @@ def create_collection(
           - ``pipeline`` (list): a list of aggregation pipeline stages
           - ``comment`` (str): a user-provided comment to attach to this command.
             This option is only supported on MongoDB >= 4.4.
+          - ``changeStreamPreAndPostImages`` (dict): a document with a boolean field ``enabled`` for
+            enabling pre- and post-images.
 
         .. versionchanged:: 4.2
            Added ``encrypted_fields`` parameter.
@@ -530,6 +532,7 @@ def watch(
         session: Optional["ClientSession"] = None,
         start_after: Optional[Mapping[str, Any]] = None,
         comment: Optional[Any] = None,
+        full_document_before_change: Optional[str] = None,
     ) -> DatabaseChangeStream[_DocumentType]:
         """Watch changes on this database.
 
@@ -576,11 +579,13 @@ def watch(
             pipeline stages are valid after a ``$changeStream`` stage, see the
             MongoDB documentation on change streams for the supported stages.
           - `full_document` (optional): The fullDocument to pass as an option
-            to the ``$changeStream`` stage. Allowed values: 'updateLookup'.
+            to the ``$changeStream`` stage. Allowed values: 'updateLookup', 'whenAvailable', 'required'.
             When set to 'updateLookup', the change notification for partial
             updates will include both a delta describing the changes to the
             document, as well as a copy of the entire document that was
             changed from some time after the change occurred.
+          - `full_document_before_change`: Allowed values: `whenAvailable` and `required`. Change events
+            may now result in a `fullDocumentBeforeChange` response field.
           - `resume_after` (optional): A resume token. If provided, the
             change stream will start returning changes that occur directly
             after the operation specified in the resume token. A resume token
@@ -607,6 +612,9 @@ def watch(
         :Returns:
           A :class:`~pymongo.change_stream.DatabaseChangeStream` cursor.
 
+        .. versionchanged:: 4.2
+            Added ``full_document_before_change`` parameter.
+
         .. versionchanged:: 4.1
            Added ``comment`` parameter.
 
@@ -631,7 +639,8 @@ def watch(
             start_at_operation_time,
             session,
             start_after,
-            comment=comment,
+            comment,
+            full_document_before_change,
         )
 
     def _command(

pymongo/mongo_client.py

@@ -871,6 +871,7 @@ def watch(
         session: Optional[client_session.ClientSession] = None,
         start_after: Optional[Mapping[str, Any]] = None,
         comment: Optional[Any] = None,
+        full_document_before_change: Optional[str] = None,
     ) -> ChangeStream[_DocumentType]:
         """Watch changes on this cluster.
 
@@ -922,6 +923,8 @@ def watch(
             updates will include both a delta describing the changes to the
             document, as well as a copy of the entire document that was
             changed from some time after the change occurred.
+          - `full_document_before_change`: Allowed values: `whenAvailable` and `required`. Change events
+            may now result in a `fullDocumentBeforeChange` response field.
           - `resume_after` (optional): A resume token. If provided, the
             change stream will start returning changes that occur directly
             after the operation specified in the resume token. A resume token
@@ -948,6 +951,9 @@ def watch(
         :Returns:
           A :class:`~pymongo.change_stream.ClusterChangeStream` cursor.
 
+        .. versionchanged:: 4.2
+            Added ``full_document_before_change`` parameter.
+
         .. versionchanged:: 4.1
            Added ``comment`` parameter.
 
@@ -972,7 +978,8 @@ def watch(
             start_at_operation_time,
             session,
             start_after,
-            comment=comment,
+            comment,
+            full_document_before_change,
         )
 
     @property