PYTHON-1598 Fix transaction write concern inheritance

ShaneHarvey · ShaneHarvey · commit 4f3b6467ab7d · 2018-06-26T12:07:12.000-04:00
diff --git a/pymongo/bulk.py b/pymongo/bulk.py
@@ -497,8 +497,7 @@ def execute(self, write_concern, session):
             raise InvalidOperation('Bulk operations can '
                                    'only be executed once.')
         self.executed = True
-        write_concern = (WriteConcern(**write_concern) if
-                         write_concern else self.collection.write_concern)
+        write_concern = write_concern or self.collection.write_concern
         session = _validate_session_write_concern(session, write_concern)
 
         if self.ordered:
@@ -690,5 +689,5 @@ def execute(self, write_concern=None):
             execution.
         """
         if write_concern is not None:
-            validate_is_mapping("write_concern", write_concern)
+            write_concern = WriteConcern(**write_concern)
         return self.__bulk.execute(write_concern, session=None)
diff --git a/pymongo/client_session.py b/pymongo/client_session.py
@@ -548,11 +548,6 @@ def _apply_to(self, command, is_retryable, read_preference):
             return
 
         if self._in_transaction:
-            # TODO: hack
-            name = next(iter(command))
-            if name not in ('commitTransaction', 'abortTransaction'):
-                command.pop('writeConcern', None)
-
             if read_preference != ReadPreference.PRIMARY:
                 raise InvalidOperation(
                     'read preference in a transaction must be primary, not: '
diff --git a/pymongo/collection.py b/pymongo/collection.py
@@ -254,7 +254,7 @@ def __create(self, options, collation, session):
         with self._socket_for_writes() as sock_info:
             self._command(
                 sock_info, cmd, read_preference=ReadPreference.PRIMARY,
-                write_concern=self.write_concern,
+                write_concern=self._write_concern_for(session),
                 collation=collation, session=session)
 
     def __getattr__(self, name):
@@ -493,7 +493,8 @@ def bulk_write(self, requests, ordered=True,
             except AttributeError:
                 raise TypeError("%r is not a valid request" % (request,))
 
-        bulk_api_result = blk.execute(self.write_concern.document, session)
+        write_concern = self._write_concern_for(session)
+        bulk_api_result = blk.execute(write_concern, session)
         if bulk_api_result is not None:
             return BulkWriteResult(bulk_api_result, True)
         return BulkWriteResult({}, False)
@@ -633,11 +634,11 @@ def gen():
                         ids.append(doc.get('_id'))
                     yield doc
 
-        concern = (write_concern or self.write_concern).document
+        write_concern = write_concern or self._write_concern_for(session)
         blk = _Bulk(self, ordered, bypass_doc_val)
         blk.ops = [(message._INSERT, doc) for doc in gen()]
         try:
-            blk.execute(concern, session=session)
+            blk.execute(write_concern, session=session)
         except BulkWriteError as bwe:
             _raise_last_error(bwe.details)
         return ids
@@ -684,11 +685,13 @@ def insert_one(self, document, bypass_document_validation=False,
         if not (isinstance(document, RawBSONDocument) or "_id" in document):
             document["_id"] = ObjectId()
 
+        write_concern = self._write_concern_for(session)
         return InsertOneResult(
             self._insert(document,
+                         write_concern=write_concern,
                          bypass_doc_val=bypass_document_validation,
                          session=session),
-            self.write_concern.acknowledged)
+            write_concern.acknowledged)
 
     def insert_many(self, documents, ordered=True,
                     bypass_document_validation=False, session=None):
@@ -744,10 +747,11 @@ def gen():
                     inserted_ids.append(document["_id"])
                 yield (message._INSERT, document)
 
+        write_concern = self._write_concern_for(session)
         blk = _Bulk(self, ordered, bypass_document_validation)
         blk.ops = [doc for doc in gen()]
-        blk.execute(self.write_concern.document, session=session)
-        return InsertManyResult(inserted_ids, self.write_concern.acknowledged)
+        blk.execute(write_concern, session=session)
+        return InsertManyResult(inserted_ids, write_concern.acknowledged)
 
     def _update(self, sock_info, criteria, document, upsert=False,
                 check_keys=True, multi=False, manipulate=False,
@@ -912,12 +916,14 @@ def replace_one(self, filter, replacement, upsert=False,
         common.validate_is_mapping("filter", filter)
         common.validate_ok_for_replace(replacement)
 
+        write_concern = self._write_concern_for(session)
         return UpdateResult(
             self._update_retryable(
                 filter, replacement, upsert,
+                write_concern=write_concern,
                 bypass_doc_val=bypass_document_validation,
                 collation=collation, session=session),
-            self.write_concern.acknowledged)
+            write_concern.acknowledged)
 
     def update_one(self, filter, update, upsert=False,
                    bypass_document_validation=False,
@@ -979,13 +985,15 @@ def update_one(self, filter, update, upsert=False,
         common.validate_ok_for_update(update)
         common.validate_list_or_none('array_filters', array_filters)
 
+        write_concern = self._write_concern_for(session)
         return UpdateResult(
             self._update_retryable(
                 filter, update, upsert, check_keys=False,
+                write_concern=write_concern,
                 bypass_doc_val=bypass_document_validation,
                 collation=collation, array_filters=array_filters,
                 session=session),
-            self.write_concern.acknowledged)
+            write_concern.acknowledged)
 
     def update_many(self, filter, update, upsert=False, array_filters=None,
                     bypass_document_validation=False, collation=None,
@@ -1047,13 +1055,15 @@ def update_many(self, filter, update, upsert=False, array_filters=None,
         common.validate_ok_for_update(update)
         common.validate_list_or_none('array_filters', array_filters)
 
+        write_concern = self._write_concern_for(session)
         return UpdateResult(
             self._update_retryable(
                 filter, update, upsert, check_keys=False, multi=True,
+                write_concern=write_concern,
                 bypass_doc_val=bypass_document_validation,
                 collation=collation, array_filters=array_filters,
                 session=session),
-            self.write_concern.acknowledged)
+            write_concern.acknowledged)
 
     def drop(self, session=None):
         """Alias for :meth:`~pymongo.database.Database.drop_collection`.
@@ -1173,10 +1183,13 @@ def delete_one(self, filter, collation=None, session=None):
 
         .. versionadded:: 3.0
         """
+        write_concern = self._write_concern_for(session)
         return DeleteResult(
             self._delete_retryable(
-                filter, False, collation=collation, session=session),
-            self.write_concern.acknowledged)
+                filter, False,
+                write_concern=write_concern,
+                collation=collation, session=session),
+            write_concern.acknowledged)
 
     def delete_many(self, filter, collation=None, session=None):
         """Delete one or more documents matching the filter.
@@ -1208,10 +1221,13 @@ def delete_many(self, filter, collation=None, session=None):
 
         .. versionadded:: 3.0
         """
+        write_concern = self._write_concern_for(session)
         return DeleteResult(
             self._delete_retryable(
-                filter, True, collation=collation, session=session),
-            self.write_concern.acknowledged)
+                filter, True,
+                write_concern=write_concern,
+                collation=collation, session=session),
+            write_concern.acknowledged)
 
     def find_one(self, filter=None, *args, **kwargs):
         """Get a single document from the database.
@@ -1809,7 +1825,7 @@ def gen_indexes():
             self._command(
                 sock_info, cmd, read_preference=ReadPreference.PRIMARY,
                 codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
-                write_concern=self.write_concern,
+                write_concern=self._write_concern_for(session),
                 session=session)
         return names
 
@@ -1840,7 +1856,7 @@ def __create_index(self, keys, index_options, session, **kwargs):
             self._command(
                 sock_info, cmd, read_preference=ReadPreference.PRIMARY,
                 codec_options=_UNICODE_REPLACE_CODEC_OPTIONS,
-                write_concern=self.write_concern,
+                write_concern=self._write_concern_for(session),
                 session=session)
 
     def create_index(self, keys, session=None, **kwargs):
@@ -2059,7 +2075,7 @@ def drop_index(self, index_or_name, session=None, **kwargs):
                           cmd,
                           read_preference=ReadPreference.PRIMARY,
                           allowable_errors=["ns not found"],
-                          write_concern=self.write_concern,
+                          write_concern=self._write_concern_for(session),
                           session=session)
 
     def reindex(self, session=None, **kwargs):
@@ -2268,7 +2284,7 @@ def _aggregate(self, pipeline, cursor_class, first_batch_size, session,
             else:
                 read_concern = None
             if 'writeConcern' not in cmd and dollar_out:
-                write_concern = self.write_concern
+                write_concern = self._write_concern_for(session)
             else:
                 write_concern = None
 
@@ -2588,7 +2604,7 @@ def rename(self, new_name, session=None, **kwargs):
         new_name = "%s.%s" % (self.__database.name, new_name)
         cmd = SON([("renameCollection", self.__full_name), ("to", new_name)])
         cmd.update(kwargs)
-        write_concern = self._write_concern_for_cmd(cmd)
+        write_concern = self._write_concern_for_cmd(cmd, session)
 
         with self._socket_for_writes() as sock_info:
             with self.__database.client._tmp_session(session) as s:
@@ -2724,7 +2740,7 @@ def map_reduce(self, map, reduce, out, full_response=False, session=None,
             else:
                 read_concern = None
             if 'writeConcern' not in cmd and not inline:
-                write_concern = self.write_concern
+                write_concern = self._write_concern_for(session)
             else:
                 write_concern = None
 
@@ -2798,12 +2814,12 @@ def inline_map_reduce(self, map, reduce, full_response=False, session=None,
         else:
             return res.get("results")
 
-    def _write_concern_for_cmd(self, cmd):
+    def _write_concern_for_cmd(self, cmd, session):
         raw_wc = cmd.get('writeConcern')
         if raw_wc is not None:
             return WriteConcern(**raw_wc)
         else:
-            return self.write_concern
+            return self._write_concern_for(session)
 
     def __find_and_modify(self, filter, projection, sort, upsert=None,
                           return_document=ReturnDocument.BEFORE,
@@ -2827,15 +2843,15 @@ def __find_and_modify(self, filter, projection, sort, upsert=None,
             common.validate_boolean("upsert", upsert)
             cmd["upsert"] = upsert
 
-        write_concern = self._write_concern_for_cmd(cmd)
+        write_concern = self._write_concern_for_cmd(cmd, session)
 
         def _find_and_modify(session, sock_info, retryable_write):
             if array_filters is not None:
                 if sock_info.max_wire_version < 6:
                     raise ConfigurationError(
                         'Must be connected to MongoDB 3.6+ to use '
                         'arrayFilters.')
-                if not self.write_concern.acknowledged:
+                if not write_concern.acknowledged:
                     raise ConfigurationError(
                         'arrayFilters is unsupported for unacknowledged '
                         'writes.')
@@ -3250,7 +3266,7 @@ def find_and_modify(self, query={}, update=None,
         cmd = SON([("findAndModify", self.__name)])
         cmd.update(kwargs)
 
-        write_concern = self._write_concern_for_cmd(cmd)
+        write_concern = self._write_concern_for_cmd(cmd, None)
 
         def _find_and_modify(session, sock_info, retryable_write):
             if (sock_info.max_wire_version >= 4 and
diff --git a/pymongo/common.py b/pymongo/common.py
@@ -33,7 +33,7 @@
 from pymongo.read_concern import ReadConcern
 from pymongo.read_preferences import _MONGOS_MODES, _ServerMode
 from pymongo.ssl_support import validate_cert_reqs
-from pymongo.write_concern import WriteConcern
+from pymongo.write_concern import DEFAULT_WRITE_CONCERN, WriteConcern
 
 try:
     from collections import OrderedDict
@@ -679,6 +679,14 @@ def write_concern(self):
         """
         return self.__write_concern
 
+    def _write_concern_for(self, session):
+        """Read only access to the write concern of this instance or session.
+        """
+        # Override this operation's write concern with the transaction's.
+        if session and session._in_transaction:
+            return DEFAULT_WRITE_CONCERN
+        return self.write_concern
+
     @property
     def read_preference(self):
         """Read only access to the read preference of this instance.
diff --git a/pymongo/database.py b/pymongo/database.py
@@ -32,7 +32,7 @@
 from pymongo.message import _first_batch
 from pymongo.read_preferences import ReadPreference
 from pymongo.son_manipulator import SONManipulator
-from pymongo.write_concern import WriteConcern
+from pymongo.write_concern import DEFAULT_WRITE_CONCERN
 
 
 _INDEX_REGEX = {"name": {"$regex": r"^(?!.*\$)"}}
@@ -749,7 +749,7 @@ def drop_collection(self, name_or_collection, session=None):
             return self._command(
                 sock_info, 'drop', value=_unicode(name),
                 allowable_errors=['ns not found'],
-                write_concern=self.write_concern,
+                write_concern=self._write_concern_for(session),
                 parse_write_concern_error=True,
                 session=session)
 
@@ -1362,7 +1362,7 @@ def __init__(self, database):
 
         if not database.write_concern.acknowledged:
             database = database.client.get_database(
-                database.name, write_concern=WriteConcern())
+                database.name, write_concern=DEFAULT_WRITE_CONCERN)
         # can't just assign it since we've overridden __setattr__
         object.__setattr__(self, "_db", database)
 
diff --git a/pymongo/mongo_client.py b/pymongo/mongo_client.py
@@ -71,7 +71,7 @@
 from pymongo.topology import Topology
 from pymongo.topology_description import TOPOLOGY_TYPE
 from pymongo.settings import TopologySettings
-from pymongo.write_concern import WriteConcern
+from pymongo.write_concern import DEFAULT_WRITE_CONCERN
 
 
 class MongoClient(common.BaseObject):
@@ -1716,7 +1716,7 @@ def drop_database(self, name_or_database, session=None):
                 sock_info,
                 "dropDatabase",
                 read_preference=ReadPreference.PRIMARY,
-                write_concern=self.write_concern,
+                write_concern=self._write_concern_for(session),
                 parse_write_concern_error=True,
                 session=session)
 
@@ -1802,7 +1802,7 @@ def _database_default_options(self, name):
         return self.get_database(
             name, codec_options=DEFAULT_CODEC_OPTIONS,
             read_preference=ReadPreference.PRIMARY,
-            write_concern=WriteConcern())
+            write_concern=DEFAULT_WRITE_CONCERN)
 
     @property
     def is_locked(self):
diff --git a/pymongo/write_concern.py b/pymongo/write_concern.py
@@ -117,3 +117,6 @@ def __eq__(self, other):
 
     def __ne__(self, other):
         return self.__document != other.document
+
+
+DEFAULT_WRITE_CONCERN = WriteConcern()
diff --git a/test/test_transactions.py b/test/test_transactions.py
