From cf58875cbd7ecda653161655ba0d9edbeaac09a6 Mon Sep 17 00:00:00 2001 From: Brian Beggs Date: Wed, 15 Jul 2015 10:12:15 -0400 Subject: [PATCH] pymongo 3.0.3 migration --- AUTHORS | 2 + common/djangoapps/track/backends/mongodb.py | 2 +- .../lib/xmodule/xmodule/contentstore/mongo.py | 10 ++--- .../xmodule/xmodule/modulestore/mongo/base.py | 42 +++++++++---------- .../xmodule/modulestore/mongo/draft.py | 8 ++-- .../split_mongo/mongo_connection.py | 27 ++++-------- .../xmodule/modulestore/split_mongo/split.py | 4 +- .../tests/test_mixed_modulestore.py | 8 ++-- requirements/edx/base.txt | 5 ++- 9 files changed, 49 insertions(+), 59 deletions(-) diff --git a/AUTHORS b/AUTHORS index 1c7c99e46186..0ac637d92fe7 100644 --- a/AUTHORS +++ b/AUTHORS @@ -229,3 +229,5 @@ Pan Luo Tyler Nickerson Vedran Karačić William Ono +Brian Beggs + diff --git a/common/djangoapps/track/backends/mongodb.py b/common/djangoapps/track/backends/mongodb.py index 71845129af89..4a732662671f 100644 --- a/common/djangoapps/track/backends/mongodb.py +++ b/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. diff --git a/common/lib/xmodule/xmodule/contentstore/mongo.py b/common/lib/xmodule/xmodule/contentstore/mongo.py index d03de7a8fca8..dda757d9ddf5 100644 --- a/common/lib/xmodule/xmodule/contentstore/mongo.py +++ b/common/lib/xmodule/xmodule/contentstore/mongo.py @@ -51,7 +51,7 @@ 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): """ @@ -59,7 +59,7 @@ def _drop_database(self): 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) diff --git a/common/lib/xmodule/xmodule/modulestore/mongo/base.py b/common/lib/xmodule/xmodule/modulestore/mongo/base.py index 898db379c07c..3b5ec4cb8ff1 100644 --- a/common/lib/xmodule/xmodule/modulestore/mongo/base.py +++ b/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 diff --git a/common/lib/xmodule/xmodule/modulestore/mongo/draft.py b/common/lib/xmodule/xmodule/modulestore/mongo/draft.py index 8c181b2639e7..c470fcb1573b 100644 --- a/common/lib/xmodule/xmodule/modulestore/mongo/draft.py +++ b/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) diff --git a/common/lib/xmodule/xmodule/modulestore/split_mongo/mongo_connection.py b/common/lib/xmodule/xmodule/modulestore/split_mongo/mongo_connection.py index abe9fd154388..c3f72665fe44 100644 --- a/common/lib/xmodule/xmodule/modulestore/split_mongo/mongo_connection.py +++ b/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): """ diff --git a/common/lib/xmodule/xmodule/modulestore/split_mongo/split.py b/common/lib/xmodule/xmodule/modulestore/split_mongo/split.py index f29a0b04a10a..85893d78f4d2 100644 --- a/common/lib/xmodule/xmodule/modulestore/split_mongo/split.py +++ b/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() diff --git a/common/lib/xmodule/xmodule/modulestore/tests/test_mixed_modulestore.py b/common/lib/xmodule/xmodule/modulestore/tests/test_mixed_modulestore.py index 1872a481d1f8..0eabc051bcf1 100644 --- a/common/lib/xmodule/xmodule/modulestore/tests/test_mixed_modulestore.py +++ b/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)}} ) diff --git a/requirements/edx/base.txt b/requirements/edx/base.txt index efa4f34121fb..de7ad5ee6425 100644 --- a/requirements/edx/base.txt +++ b/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