PLAT-723 - pymongo 3.0.3 migration

AUTHORS

@@ -229,3 +229,5 @@ Pan Luo <pan.luo@ubc.ca>
 Tyler Nickerson <nickersoft@gmail.com>
 Vedran Karačić <vedran@edx.org>
 William Ono <william.ono@ubc.ca>
+Brian Beggs <bbeggs@edx.org>
+

common/djangoapps/track/backends/mongodb.py

@@ -89,7 +89,7 @@ def _create_indexes(self):
     def send(self, event):
         """Insert the event in to the Mongo collection"""
         try:
-            self.collection.insert(event, manipulate=False)
+            self.collection.insert_one(event)
         except (PyMongoError, BSONError):
             # The event will be lost in case of a connection error or any error
             # that occurs when trying to insert the event into Mongo.

common/lib/xmodule/xmodule/contentstore/mongo.py

@@ -51,15 +51,15 @@ def close_connections(self):
         """
         Closes any open connections to the underlying databases
         """
-        self.fs_files.database.connection.close()
+        self.fs_files.database.client.close()
 
     def _drop_database(self):
         """
         A destructive operation to drop the underlying database and close all connections.
         Intended to be used by test code for cleanup.
         """
         self.close_connections()
-        self.fs_files.database.connection.drop_database(self.fs_files.database)
+        self.fs_files.database.client.drop_database(self.fs_files.database)
 
     def save(self, content):
         content_id, content_son = self.asset_db_key(content.location)
@@ -196,9 +196,9 @@ def remove_redundant_content_for_courses(self):
             items = self.fs_files.find(query)
             assets_to_delete = assets_to_delete + items.count()
             for asset in items:
-                self.fs.delete(asset[prefix])
+                self.fs.delete_one(asset[prefix])
 
-            self.fs_files.remove(query)
+            self.fs_files.delete_many(query)
         return assets_to_delete
 
     def _get_all_content_for_course(self,
@@ -279,7 +279,7 @@ def set_attrs(self, location, attr_dict):
                 raise AttributeError("{} is a protected attribute.".format(attr))
         asset_db_key, __ = self.asset_db_key(location)
         # catch upsert error and raise NotFoundError if asset doesn't exist
-        result = self.fs_files.update({'_id': asset_db_key}, {"$set": attr_dict}, upsert=False)
+        result = self.fs_files.update_one({'_id': asset_db_key}, {"$set": attr_dict}, upsert=False)
         if not result.get('updatedExisting', True):
             raise NotFoundError(asset_db_key)
 

common/lib/xmodule/xmodule/modulestore/mongo/base.py

@@ -558,8 +558,8 @@ def do_connection(
             """
             Create & open the connection, authenticate, and provide pointers to the collection
             """
-            # Remove the replicaSet parameter.
-            kwargs.pop('replicaSet', None)
+            # Set the write concern to 1 (data has been written to the primary) to ensure data integrity.
+            kwargs['w'] = kwargs.get('w', 1)
 
             self.database = MongoProxy(
                 pymongo.database.Database(
@@ -586,9 +586,6 @@ def do_connection(
 
         do_connection(**doc_store_config)
 
-        # Force mongo to report errors, at the expense of performance
-        self.collection.write_concern = {'w': 1}
-
         if default_class is not None:
             module_path, _, class_name = default_class.rpartition('.')
             class_ = getattr(import_module(module_path), class_name)
@@ -616,6 +613,7 @@ def mongo_wire_version(self):
         Returns the wire version for mongo. Only used to unit tests which instrument the connection.
         """
         self.database.connection._ensure_connected()
+        # TODO bbeggs max_wire_version if deprecated... Replace? Remove?
         return self.database.connection.max_wire_version
 
     def _drop_database(self):
@@ -626,7 +624,7 @@ def _drop_database(self):
         # drop the assets
         super(MongoModuleStore, self)._drop_database()
 
-        connection = self.collection.database.connection
+        connection = self.collection.database.client
         connection.drop_database(self.collection.database.proxied_object)
         connection.close()
 
@@ -1066,7 +1064,7 @@ def has_course(self, course_key, ignore_case=False, **kwargs):
                     course_query[key] = re.compile(r"(?i)^{}$".format(course_query[key]))
         else:
             course_query = {'_id': location.to_deprecated_son()}
-        course = self.collection.find_one(course_query, fields={'_id': True})
+        course = self.collection.find_one(course_query)
         if course:
             return SlashSeparatedCourseKey(course['_id']['org'], course['_id']['course'], course['_id']['name'])
         else:
@@ -1227,7 +1225,7 @@ def create_course(self, org, course, run, user_id, fields=None, **kwargs):
             ('_id.course', re.compile(u'^{}$'.format(course_id.course), re.IGNORECASE)),
             ('_id.category', 'course'),
         ])
-        courses = self.collection.find(course_search_location, fields=('_id'))
+        courses = self.collection.find(course_search_location)
         if courses.count() > 0:
             raise DuplicateCourseError(course_id, courses[0]['_id'])
 
@@ -1393,14 +1391,12 @@ def _update_single_item(self, location, update, allow_not_found=False):
         bulk_record.dirty = True
         # See http://www.mongodb.org/display/DOCS/Updating for
         # atomic update syntax
-        result = self.collection.update(
+        result = self.collection.update_one(
             {'_id': location.to_deprecated_son()},
             {'$set': update},
-            multi=False,
-            upsert=allow_not_found,
-            w=1,  # wait until primary commits
+            upsert=True,
         )
-        if result['n'] == 0:
+        if result.raw_result['n'] == 0:
             raise ItemNotFoundError(location)
 
     def _update_ancestors(self, location, update):
@@ -1530,10 +1526,9 @@ def _get_non_orphan_parents(self, location, parents, revision):
                     bulk_record.dirty = True
                     # The parent is an orphan, so remove all the children including
                     # the location whose parent we are looking for from orphan parent
-                    self.collection.update(
+                    self.collection.update_one(
                         {'_id': parent_loc.to_deprecated_son()},
                         {'$set': {'definition.children': []}},
-                        multi=False,
                         upsert=True,
                     )
                 elif ancestor_loc.category == 'course':
@@ -1727,13 +1722,13 @@ def _find_course_assets(self, course_key):
             else:
                 # Course exists, so create matching assets document.
                 course_assets = {'course_id': unicode(course_key), 'assets': {}}
-                doc_id = self.asset_collection.insert(course_assets)
+                doc_id = self.asset_collection.insert_one(course_assets).inserted_id
         elif isinstance(course_assets['assets'], list):
             # This record is in the old course assets format.
             # Ensure that no data exists before updating the format.
             assert len(course_assets['assets']) == 0
             # Update the format to a dict.
-            self.asset_collection.update(
+            self.asset_collection.update_one(
                 {'_id': doc_id},
                 {'$set': {'assets': {}}}
             )
@@ -1767,7 +1762,8 @@ def _save_asset_metadata_list(self, asset_metadata_list, user_id, import_only):
             updates_by_type[self._make_mongo_asset_key(asset_type)] = assets.as_list()
 
         # Update the document.
-        self.asset_collection.update(
+        #TODO BBEGGS should this be an upsert?
+        self.asset_collection.update_one(
             {'_id': course_assets.doc_id},
             {'$set': updates_by_type}
         )
@@ -1817,9 +1813,9 @@ def copy_all_asset_metadata(self, source_course_key, dest_course_key, user_id):
         """
         source_assets = self._find_course_assets(source_course_key)
         dest_assets = {'assets': source_assets.asset_md.copy(), 'course_id': unicode(dest_course_key)}
-        self.asset_collection.remove({'course_id': unicode(dest_course_key)})
+        self.asset_collection.delete_one({'course_id': unicode(dest_course_key)})
         # Update the document.
-        self.asset_collection.insert(dest_assets)
+        self.asset_collection.insert_one(dest_assets)
 
     @contract(asset_key='AssetKey', attr_dict=dict, user_id='int|long')
     def set_asset_metadata_attrs(self, asset_key, attr_dict, user_id):
@@ -1847,7 +1843,7 @@ def set_asset_metadata_attrs(self, asset_key, attr_dict, user_id):
         # Generate a Mongo doc from the metadata and update the course asset info.
         all_assets[asset_idx] = md.to_storable()
 
-        self.asset_collection.update(
+        self.asset_collection.update_one(
             {'_id': course_assets.doc_id},
             {"$set": {self._make_mongo_asset_key(asset_key.asset_type): all_assets}}
         )
@@ -1871,7 +1867,7 @@ def delete_asset_metadata(self, asset_key, user_id):
         all_asset_info.pop(asset_idx)
 
         # Update the document.
-        self.asset_collection.update(
+        self.asset_collection.update_one(
             {'_id': course_assets.doc_id},
             {'$set': {self._make_mongo_asset_key(asset_key.asset_type): all_asset_info}}
         )
@@ -1890,7 +1886,7 @@ def delete_all_asset_metadata(self, course_key, user_id):
         # A single document exists per course to store the course asset metadata.
         try:
             course_assets = self._find_course_assets(course_key)
-            self.asset_collection.remove(course_assets.doc_id)
+            self.asset_collection.delete_one(course_assets.doc_id)
         except ItemNotFoundError:
             # When deleting asset metadata, if a course's asset metadata is not present, no big deal.
             pass

common/lib/xmodule/xmodule/modulestore/mongo/draft.py

@@ -164,7 +164,7 @@ def delete_course(self, course_key, user_id):
 
         # delete all of the db records for the course
         course_query = self._course_key_to_son(course_key)
-        self.collection.remove(course_query, multi=True)
+        self.collection.delete_many(course_query)
         self.delete_all_asset_metadata(course_key, user_id)
 
     def clone_course(self, source_course_id, dest_course_id, user_id, fields=None, **kwargs):
@@ -432,7 +432,7 @@ def convert_item(item, to_be_deleted):
             bulk_record = self._get_bulk_ops_record(location.course_key)
             bulk_record.dirty = True
             try:
-                self.collection.insert(item)
+                self.collection.insert_one(item)
             except pymongo.errors.DuplicateKeyError:
                 # prevent re-creation of DRAFT versions, unless explicitly requested to ignore
                 if not ignore_if_draft:
@@ -635,7 +635,7 @@ def _internal(tier):
         if len(to_be_deleted) > 0:
             bulk_record = self._get_bulk_ops_record(root_usages[0].course_key)
             bulk_record.dirty = True
-            self.collection.remove({'_id': {'$in': to_be_deleted}}, safe=self.collection.safe)
+            self.collection.delete_many({'_id': {'$in': to_be_deleted}})
 
     @memoize_in_request_cache('request_cache')
     def has_changes(self, xblock):
@@ -739,7 +739,7 @@ def _internal_depth_first(item_location, is_root):
         bulk_record = self._get_bulk_ops_record(course_key)
         if len(to_be_deleted) > 0:
             bulk_record.dirty = True
-            self.collection.remove({'_id': {'$in': to_be_deleted}})
+            self.collection.delete_many({'_id': {'$in': to_be_deleted}})
 
         self._flag_publish_event(course_key)
 

common/lib/xmodule/xmodule/modulestore/split_mongo/mongo_connection.py

@@ -271,12 +271,10 @@ def __init__(
         """
         Create & open the connection, authenticate, and provide pointers to the collections
         """
-        if kwargs.get('replicaSet') is None:
-            kwargs.pop('replicaSet', None)
-            mongo_class = pymongo.MongoClient
-        else:
-            mongo_class = pymongo.MongoReplicaSetClient
-        _client = mongo_class(
+        # Set the write concern to 1 (data has been written to the primary) to ensure data integrity.
+        kwargs['w'] = kwargs.get('w', 1)
+
+        _client = pymongo.MongoClient(
             host=host,
             port=port,
             tz_aware=tz_aware,
@@ -294,13 +292,6 @@ def __init__(
         self.structures = self.database[collection + '.structures']
         self.definitions = self.database[collection + '.definitions']
 
-        # every app has write access to the db (v having a flag to indicate r/o v write)
-        # Force mongo to report errors, at the expense of performance
-        # pymongo docs suck but explanation:
-        # http://api.mongodb.org/java/2.10.1/com/mongodb/WriteConcern.html
-        self.course_index.write_concern = {'w': 1}
-        self.structures.write_concern = {'w': 1}
-        self.definitions.write_concern = {'w': 1}
 
     def heartbeat(self):
         """
@@ -404,7 +395,7 @@ def insert_structure(self, structure, course_context=None):
         """
         with TIMER.timer("insert_structure", course_context) as tagger:
             tagger.measure("blocks", len(structure["blocks"]))
-            self.structures.insert(structure_to_mongo(structure, course_context))
+            self.structures.insert_one(structure_to_mongo(structure, course_context))
 
     def get_course_index(self, key, ignore_case=False):
         """
@@ -454,7 +445,7 @@ def insert_course_index(self, course_index, course_context=None):
         """
         with TIMER.timer("insert_course_index", course_context):
             course_index['last_update'] = datetime.datetime.now(pytz.utc)
-            self.course_index.insert(course_index)
+            self.course_index.insert_one(course_index)
 
     def update_course_index(self, course_index, from_index=None, course_context=None):
         """
@@ -477,7 +468,7 @@ def update_course_index(self, course_index, from_index=None, course_context=None
                     'run': course_index['run'],
                 }
             course_index['last_update'] = datetime.datetime.now(pytz.utc)
-            self.course_index.update(query, course_index, upsert=False,)
+            self.course_index.update_one(query, {'$set': course_index}, upsert=False)
 
     def delete_course_index(self, course_key):
         """
@@ -488,7 +479,7 @@ def delete_course_index(self, course_key):
                 key_attr: getattr(course_key, key_attr)
                 for key_attr in ('org', 'course', 'run')
             }
-            return self.course_index.remove(query)
+            return self.course_index.delete_many(query)
 
     def get_definition(self, key, course_context=None):
         """
@@ -516,7 +507,7 @@ def insert_definition(self, definition, course_context=None):
         with TIMER.timer("insert_definition", course_context) as tagger:
             tagger.measure('fields', len(definition['fields']))
             tagger.tag(block_type=definition['block_type'])
-            self.definitions.insert(definition)
+            self.definitions.insert_one(definition)
 
     def ensure_indexes(self):
         """

common/lib/xmodule/xmodule/modulestore/split_mongo/split.py

@@ -678,7 +678,7 @@ def close_connections(self):
         """
         Closes any open connections to the underlying databases
         """
-        self.db.connection.close()
+        self.db.client.close()
 
     def mongo_wire_version(self):
         """
@@ -694,7 +694,7 @@ def _drop_database(self):
         # drop the assets
         super(SplitMongoModuleStore, self)._drop_database()
 
-        connection = self.db.connection
+        connection = self.db.client
         connection.drop_database(self.db.name)
         connection.close()
 

common/lib/xmodule/xmodule/modulestore/tests/test_mixed_modulestore.py

@@ -1180,7 +1180,7 @@ def test_get_parent_location_draft(self, default_ms):
 
         mongo_store = self.store._get_modulestore_for_courselike(course_id)  # pylint: disable=protected-access
         # add another parent (unit) "vertical_x1b" for problem "problem_x1a_1"
-        mongo_store.collection.update(
+        mongo_store.collection.update_many(
             self.vertical_x1b.to_deprecated_son('_id.'),
             {'$push': {'definition.children': unicode(self.problem_x1a_1)}}
         )
@@ -1455,11 +1455,11 @@ def test_get_non_orphan_parents(self, default_ms):
         self.assertEqual(len(set(found_orphans)), 2)
 
         # add orphan vertical and sequential as another parents of problem "problem_x1a_1"
-        mongo_store.collection.update(
+        mongo_store.collection.update_many(
             orphan_sequential.to_deprecated_son('_id.'),
             {'$push': {'definition.children': unicode(self.problem_x1a_1)}}
         )
-        mongo_store.collection.update(
+        mongo_store.collection.update_many(
             orphan_vertical.to_deprecated_son('_id.'),
             {'$push': {'definition.children': unicode(self.problem_x1a_1)}}
         )
@@ -1470,7 +1470,7 @@ def test_get_non_orphan_parents(self, default_ms):
             self.assertEqual(parent, self.vertical_x1a)
 
         # now add valid published vertical as another parent of problem
-        mongo_store.collection.update(
+        mongo_store.collection.update_many(
             self.sequential_x1.to_deprecated_son('_id.'),
             {'$push': {'definition.children': unicode(self.problem_x1a_1)}}
         )

requirements/edx/base.txt

@@ -51,7 +51,7 @@ Markdown==2.2.1
 --allow-external meliae
 --allow-unverified meliae
 meliae==0.4.0
-mongoengine==0.7.10
+mongoengine==0.10.0
 networkx==1.7
 nose==1.3.3
 oauthlib==0.7.2
@@ -63,7 +63,7 @@ pycrypto>=2.6
 pygments==2.0.1
 pygraphviz==1.1
 PyJWT==1.0.1
-pymongo==2.7.2
+pymongo==3.0.3
 pyparsing==2.0.1
 python-memcached==1.48
 python-openid==2.2.5
@@ -156,3 +156,4 @@ analytics-python==0.4.4
 # Needed for mailchimp(mailing djangoapp)
 mailsnake==1.6.2
 jsonfield==1.0.3
+blinker