Add bulk operations to split modulestore

common/lib/xmodule/xmodule/modulestore/exceptions.py

@@ -54,20 +54,16 @@ class DuplicateItemError(Exception):
             self, Exception.__str__(self, *args, **kwargs)
         )
 
+
 class VersionConflictError(Exception):
     """
     The caller asked for either draft or published head and gave a version which conflicted with it.
     """
     def __init__(self, requestedLocation, currentHeadVersionGuid):
-        super(VersionConflictError, self).__init__()
-        self.requestedLocation = requestedLocation
-        self.currentHeadVersionGuid = currentHeadVersionGuid
-
-    def __str__(self, *args, **kwargs):
-        """
-        Print requested and current head info
-        """
-        return u'Requested {} but {} is current head'.format(self.requestedLocation, self.currentHeadVersionGuid)
+        super(VersionConflictError, self).__init__(u'Requested {}, but current head is {}'.format(
+            requestedLocation,
+            currentHeadVersionGuid
+        ))
 
 
 class DuplicateCourseError(Exception):

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

@@ -17,6 +17,7 @@ import sys
 import logging
 import copy
 import re
+import threading
 from uuid import uuid4
 
 from bson.son import SON
@@ -439,7 +440,7 @@ class MongoModuleStore(ModuleStoreDraftAndPublished, ModuleStoreWriteBase):
         """
         Prevent updating the meta-data inheritance cache for the given course
         """
-        if not hasattr(self.ignore_write_events_on_courses.courses):
+        if not hasattr(self.ignore_write_events_on_courses, 'courses'):
             self.ignore_write_events_on_courses.courses = set()
 
         self.ignore_write_events_on_courses.courses.add(course_id)
@@ -449,18 +450,18 @@ class MongoModuleStore(ModuleStoreDraftAndPublished, ModuleStoreWriteBase):
         Restart updating the meta-data inheritance cache for the given course.
         Refresh the meta-data inheritance cache now since it was temporarily disabled.
         """
-        if not hasattr(self.ignore_write_events_on_courses.courses):
+        if not hasattr(self.ignore_write_events_on_courses, 'courses'):
             return
 
-        if course_id in self.ignore_write_events_on_courses:
-            self.ignore_write_events_on_courses.remove(course_id)
+        if course_id in self.ignore_write_events_on_courses.courses:
+            self.ignore_write_events_on_courses.courses.remove(course_id)
             self.refresh_cached_metadata_inheritance_tree(course_id)
 
     def _is_bulk_write_in_progress(self, course_id):
         """
         Returns whether a bulk write operation is in progress for the given course.
         """
-        if not hasattr(self.ignore_write_events_on_courses.courses):
+        if not hasattr(self.ignore_write_events_on_courses, 'courses'):
             return False
 
         course_id = course_id.for_branch(None)

common/lib/xmodule/xmodule/modulestore/split_migrator.py

@@ -55,24 +55,25 @@ class SplitMigrator(object):
             new_run = source_course_key.run
 
         new_course_key = CourseLocator(new_org, new_course, new_run, branch=ModuleStoreEnum.BranchName.published)
-        new_fields = self._get_fields_translate_references(original_course, new_course_key, None)
-        if fields:
-            new_fields.update(fields)
-        new_course = self.split_modulestore.create_course(
-            new_org, new_course, new_run, user_id,
-            fields=new_fields,
-            master_branch=ModuleStoreEnum.BranchName.published,
-            skip_auto_publish=True,
-            **kwargs
-        )
-
-        with self.split_modulestore.bulk_write_operations(new_course.id):
-            self._copy_published_modules_to_course(
-                new_course, original_course.location, source_course_key, user_id, **kwargs
+        with self.split_modulestore.bulk_write_operations(new_course_key):
+            new_fields = self._get_fields_translate_references(original_course, new_course_key, None)
+            if fields:
+                new_fields.update(fields)
+            new_course = self.split_modulestore.create_course(
+                new_org, new_course, new_run, user_id,
+                fields=new_fields,
+                master_branch=ModuleStoreEnum.BranchName.published,
+                skip_auto_publish=True,
+                **kwargs
             )
-        # create a new version for the drafts
-        with self.split_modulestore.bulk_write_operations(new_course.id):
-            self._add_draft_modules_to_course(new_course.location, source_course_key, user_id, **kwargs)
+
+            with self.split_modulestore.bulk_write_operations(new_course.id):
+                self._copy_published_modules_to_course(
+                    new_course, original_course.location, source_course_key, user_id, **kwargs
+                )
+
+                # create a new version for the drafts
+                self._add_draft_modules_to_course(new_course.location, source_course_key, user_id, **kwargs)
 
         return new_course.id
 
@@ -101,7 +102,6 @@ class SplitMigrator(object):
                     fields=self._get_fields_translate_references(
                         module, course_version_locator, new_course.location.block_id
                     ),
-                    continue_version=True,
                     skip_auto_publish=True,
                     **kwargs
                 )
@@ -109,7 +109,7 @@ class SplitMigrator(object):
         index_info = self.split_modulestore.get_course_index_info(course_version_locator)
         versions = index_info['versions']
         versions[ModuleStoreEnum.BranchName.draft] = versions[ModuleStoreEnum.BranchName.published]
-        self.split_modulestore.update_course_index(index_info)
+        self.split_modulestore.update_course_index(course_version_locator, index_info)
 
         # clean up orphans in published version: in old mongo, parents pointed to the union of their published and draft
         # children which meant some pointers were to non-existent locations in 'direct'

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

@@ -2,7 +2,7 @@ import sys
 import logging
 from xblock.runtime import KvsFieldData
 from xblock.fields import ScopeIds
-from opaque_keys.edx.locator import BlockUsageLocator, LocalId, CourseLocator
+from opaque_keys.edx.locator import BlockUsageLocator, LocalId, CourseLocator, DefinitionLocator
 from xmodule.mako_module import MakoDescriptorSystem
 from xmodule.error_module import ErrorDescriptor
 from xmodule.errortracker import exc_info_to_str
@@ -10,6 +10,7 @@ from xmodule.modulestore.split_mongo import encode_key_for_mongo
 from ..exceptions import ItemNotFoundError
 from .split_mongo_kvs import SplitMongoKVS
 from fs.osfs import OSFS
+from .definition_lazy_loader import DefinitionLazyLoader
 
 log = logging.getLogger(__name__)
 
@@ -120,9 +121,24 @@ class CachingDescriptorSystem(MakoDescriptorSystem):
             self.course_entry['org'] = course_entry_override['org']
             self.course_entry['course'] = course_entry_override['course']
             self.course_entry['run'] = course_entry_override['run']
-        # most likely a lazy loader or the id directly
-        definition = json_data.get('definition', {})
-        definition_id = self.modulestore.definition_locator(definition)
+
+        definition_id = json_data.get('definition')
+        block_type = json_data['category']
+
+        if definition_id is not None and not json_data.get('definition_loaded', False):
+            definition_loader = DefinitionLazyLoader(
+                self.modulestore, block_type, definition_id,
+                lambda fields: self.modulestore.convert_references_to_keys(
+                    course_key, self.load_block_type(block_type),
+                    fields, self.course_entry['structure']['blocks'],
+                )
+            )
+        else:
+            definition_loader = None
+
+        # If no definition id is provide, generate an in-memory id
+        if definition_id is None:
+            definition_id = LocalId()
 
         # If no usage id is provided, generate an in-memory id
         if block_id is None:
@@ -130,7 +146,7 @@ class CachingDescriptorSystem(MakoDescriptorSystem):
 
         block_locator = BlockUsageLocator(
             course_key,
-            block_type=json_data.get('category'),
+            block_type=block_type,
             block_id=block_id,
         )
 
@@ -138,7 +154,7 @@ class CachingDescriptorSystem(MakoDescriptorSystem):
             block_locator.course_key, class_, json_data.get('fields', {}), self.course_entry['structure']['blocks'],
         )
         kvs = SplitMongoKVS(
-            definition,
+            definition_loader,
             converted_fields,
             json_data.get('_inherited_settings'),
             **kwargs
@@ -148,7 +164,7 @@ class CachingDescriptorSystem(MakoDescriptorSystem):
         try:
             module = self.construct_xblock_from_class(
                 class_,
-                ScopeIds(None, json_data.get('category'), definition_id, block_locator),
+                ScopeIds(None, block_type, definition_id, block_locator),
                 field_data,
             )
         except Exception:
@@ -174,7 +190,7 @@ class CachingDescriptorSystem(MakoDescriptorSystem):
         module.previous_version = edit_info.get('previous_version')
         module.update_version = edit_info.get('update_version')
         module.source_version = edit_info.get('source_version', None)
-        module.definition_locator = definition_id
+        module.definition_locator = DefinitionLocator(block_type, definition_id)
         # decache any pending field settings
         module.save()
 

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

@@ -1,4 +1,5 @@
 from opaque_keys.edx.locator import DefinitionLocator
+from bson import SON
 
 
 class DefinitionLazyLoader(object):
@@ -24,3 +25,9 @@ class DefinitionLazyLoader(object):
         loader pointer with the result so as not to fetch more than once
         """
         return self.modulestore.db_connection.get_definition(self.definition_locator.definition_id)
+
+    def as_son(self):
+        return SON((
+            ('category', self.definition_locator.block_type),
+            ('definition', self.definition_locator.definition_id)
+        ))

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

@@ -76,6 +76,12 @@ class MongoConnection(object):
         """
         self.structures.update({'_id': structure['_id']}, structure)
 
+    def upsert_structure(self, structure):
+        """
+        Update the db record for structure, creating that record if it doesn't already exist
+        """
+        self.structures.update({'_id': structure['_id']}, structure, upsert=True)
+
     def get_course_index(self, key, ignore_case=False):
         """
         Get the course_index from the persistence mechanism whose id is the given key
@@ -101,13 +107,21 @@ class MongoConnection(object):
         """
         self.course_index.insert(course_index)
 
-    def update_course_index(self, course_index):
+    def update_course_index(self, course_index, from_index=None):
         """
-        Update the db record for course_index
+        Update the db record for course_index.
+
+        Arguments:
+            from_index: If set, only update an index if it matches the one specified in `from_index`.
         """
         self.course_index.update(
-            son.SON([('org', course_index['org']), ('course', course_index['course']), ('run', course_index['run'])]),
-            course_index
+            from_index or son.SON([
+                ('org', course_index['org']),
+                ('course', course_index['course']),
+                ('run', course_index['run'])
+            ]),
+            course_index,
+            upsert=False,
         )
 
     def delete_course_index(self, course_index):

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

@@ -57,6 +57,7 @@ from path import path
 import copy
 from pytz import UTC
 from bson.objectid import ObjectId
+from pymongo.errors import DuplicateKeyError
 
 from xblock.core import XBlock
 from xblock.fields import Scope, Reference, ReferenceList, ReferenceValueDict
@@ -72,7 +73,6 @@ from xmodule.modulestore import (
 )
 
 from ..exceptions import ItemNotFoundError
-from .definition_lazy_loader import DefinitionLazyLoader
 from .caching_descriptor_system import CachingDescriptorSystem
 from xmodule.modulestore.split_mongo.mongo_connection import MongoConnection
 from xmodule.error_module import ErrorDescriptor
@@ -82,6 +82,7 @@ from types import NoneType
 
 
 log = logging.getLogger(__name__)
+
 #==============================================================================
 #
 # Known issue:
@@ -104,7 +105,220 @@ log = logging.getLogger(__name__)
 EXCLUDE_ALL = '*'
 
 
-class SplitMongoModuleStore(ModuleStoreWriteBase):
+class BulkWriteRecord(object):
+    def __init__(self):
+        self.active_count = 0
+        self.dirty_branches = set()
+        self.initial_index = None
+        self.index = None
+        self._structures = {}
+
+    def set_structure(self, branch, structure):
+        self.index['versions'][branch] = structure['_id']
+        self._structures[branch] = structure
+        self.dirty_branches.add(branch)
+
+    def get_structure(self, branch):
+        return self._structures.get(branch)
+
+    def __repr__(self):
+        return u"BulkWriteRecord<{}, {}, {}, {}, {}>".format(
+            self.active_count,
+            self.dirty_branches,
+            self.initial_index,
+            self.index,
+            self._structures,
+        )
+
+
+class BulkWriteMixin(object):
+    """
+    This implements the :meth:`bulk_write_operations` modulestore semantics for the :class:`SplitMongoModuleStore`.
+
+    In particular, it implements :meth:`_begin_bulk_write_operation` and
+    :meth:`_end_bulk_write_operation` to provide the external interface, and then exposes a set of methods
+    for interacting with course_indexes and structures that can be used by :class:`SplitMongoModuleStore`.
+
+    Internally, this mixin records the set of all active bulk operations (keyed on the active course),
+    and only writes those values to ``self.mongo_connection`` when :meth:`_end_bulk_write_operation` is called.
+    If a bulk write operation isn't active, then the changes are immediately written to the underlying
+    mongo_connection.
+    """
+    def __init__(self, *args, **kwargs):
+        super(BulkWriteMixin, self).__init__(*args, **kwargs)
+        self._active_bulk_writes = threading.local()
+
+    def _get_bulk_write_record(self, course_key, ignore_case=False):
+        """
+        Return the :class:`.BulkWriteRecord` for this course.
+        """
+        if course_key is None:
+            return BulkWriteRecord()
+
+        if not isinstance(course_key, CourseLocator):
+            raise TypeError(u'{!r} is not a CourseLocator'.format(course_key))
+        if not hasattr(self._active_bulk_writes, 'records'):
+            self._active_bulk_writes.records = defaultdict(BulkWriteRecord)
+
+        # Retrieve the bulk record based on matching org/course/run (possibly ignoring case)
+        if course_key.org and course_key.course and course_key.run:
+            if ignore_case:
+                for key, record in self._active_bulk_writes.records.iteritems():
+                    if (
+                        key.org.lower() == course_key.org.lower() and
+                        key.course.lower() == course_key.course.lower() and
+                        key.run.lower() == course_key.run.lower()
+                    ):
+                        return record
+                # If nothing matches case-insesitively, fall through to creating a new record with the passed in case
+            return self._active_bulk_writes.records[course_key.replace(branch=None, version_guid=None)]
+        else:
+            # If nothing org/course/run aren't set, use a bulk record that is identified just by the version_guid
+            return self._active_bulk_writes.records[course_key.replace(org=None, course=None, run=None, branch=None)]
+
+    def _clear_bulk_write_record(self, course_key):
+        if not isinstance(course_key, CourseLocator):
+            raise TypeError('{!r} is not a CourseLocator'.format(course_key))
+
+        if not hasattr(self._active_bulk_writes, 'records'):
+            return
+
+        if course_key.org and course_key.course and course_key.run:
+            del self._active_bulk_writes.records[course_key.replace(branch=None, version_guid=None)]
+        else:
+            del self._active_bulk_writes.records[course_key.replace(org=None, course=None, run=None, branch=None)]
+
+    def _begin_bulk_write_operation(self, course_key):
+        """
+        Begin a bulk write operation on course_key.
+        """
+        bulk_write_record = self._get_bulk_write_record(course_key)
+        bulk_write_record.active_count += 1
+
+        if bulk_write_record.active_count > 1:
+            return
+
+        bulk_write_record.initial_index = bulk_write_record.index = self.db_connection.get_course_index(course_key)
+
+    def _end_bulk_write_operation(self, course_key):
+        """
+        End the active bulk write operation on course_key.
+        """
+        # If no bulk write is active, return
+        bulk_write_record = self._get_bulk_write_record(course_key)
+        if bulk_write_record.active_count == 0:
+            return
+
+        bulk_write_record.active_count -= 1
+
+        # If more than one nested bulk write is active, decrement and continue
+        if bulk_write_record.active_count > 0:
+            return
+
+        # If this is the last active bulk write, and the content is dirty,
+        # then mark it as inactive, and update the database
+        if bulk_write_record.dirty_branches:
+            for branch in bulk_write_record.dirty_branches:
+                try:
+                    self.db_connection.insert_structure(bulk_write_record.get_structure(branch))
+                except DuplicateKeyError:
+                    pass  # The structure already exists, so we don't have to write it out again
+
+            if bulk_write_record.index is not None and bulk_write_record.index != bulk_write_record.initial_index:
+                if bulk_write_record.initial_index is None:
+                    self.db_connection.insert_course_index(bulk_write_record.index)
+                else:
+                    self.db_connection.update_course_index(bulk_write_record.index, from_index=bulk_write_record.initial_index)
+            self._clear_bulk_write_record(course_key)
+
+    def _is_in_bulk_write_operation(self, course_key, ignore_case=False):
+        """
+        Return whether a bulk write is active on `course_key`.
+        """
+        return self._get_bulk_write_record(course_key, ignore_case).active_count > 0
+
+    def get_course_index(self, course_key, ignore_case=False):
+        """
+        Return the index for course_key.
+        """
+        if self._is_in_bulk_write_operation(course_key, ignore_case):
+            return self._get_bulk_write_record(course_key, ignore_case).index
+        else:
+            return self.db_connection.get_course_index(course_key, ignore_case)
+
+    def insert_course_index(self, course_key, index_entry):
+        bulk_write_record = self._get_bulk_write_record(course_key)
+        if bulk_write_record.active_count > 0:
+            bulk_write_record.index = index_entry
+        else:
+            self.db_connection.insert_course_index(index_entry)
+
+    def update_course_index(self, course_key, updated_index_entry):
+        """
+        Change the given course's index entry.
+
+        Note, this operation can be dangerous and break running courses.
+
+        Does not return anything useful.
+        """
+        bulk_write_record = self._get_bulk_write_record(course_key)
+        if bulk_write_record.active_count > 0:
+            bulk_write_record.index = updated_index_entry
+        else:
+            self.db_connection.update_course_index(updated_index_entry)
+
+    def get_structure(self, course_key, version_guid):
+        bulk_write_record = self._get_bulk_write_record(course_key)
+        if bulk_write_record.active_count > 0:
+            structure = bulk_write_record.get_structure(course_key.branch)
+
+            # The structure hasn't been loaded from the db yet, so load it
+            if structure is None:
+                structure = self.db_connection.get_structure(bulk_write_record.index['versions'][course_key.branch])
+                bulk_write_record.set_structure(course_key.branch, structure)
+
+            return structure
+        else:
+            # cast string to ObjectId if necessary
+            version_guid = course_key.as_object_id(version_guid)
+            return self.db_connection.get_structure(version_guid)
+
+    def update_structure(self, course_key, structure):
+        self._clear_cache(structure['_id'])
+        bulk_write_record = self._get_bulk_write_record(course_key)
+        if bulk_write_record.active_count > 0:
+            bulk_write_record.set_structure(course_key.branch, structure)
+        else:
+            self.db_connection.upsert_structure(structure)
+
+    def version_structure(self, course_key, structure, user_id):
+        """
+        Copy the structure and update the history info (edited_by, edited_on, previous_version)
+        :param structure:
+        :param user_id:
+        """
+        bulk_write_record = self._get_bulk_write_record(course_key)
+
+        # If we have an active bulk write, and it's already been edited, then just use that structure
+        if bulk_write_record.active_count > 0 and course_key.branch in bulk_write_record.dirty_branches:
+            return bulk_write_record.get_structure(course_key.branch)
+
+        # Otherwise, make a new structure
+        new_structure = copy.deepcopy(structure)
+        new_structure['_id'] = ObjectId()
+        new_structure['previous_version'] = structure['_id']
+        new_structure['edited_by'] = user_id
+        new_structure['edited_on'] = datetime.datetime.now(UTC)
+        new_structure['schema_version'] = self.SCHEMA_VERSION
+
+        # If we're in a bulk write, update the structure used there, and mark it as dirty
+        if bulk_write_record.active_count > 0:
+            bulk_write_record.set_structure(course_key.branch, new_structure)
+
+        return new_structure
+
+
+class SplitMongoModuleStore(ModuleStoreWriteBase, BulkWriteMixin):
     """
     A Mongodb backed ModuleStore supporting versions, inheritance,
     and sharing.
@@ -173,50 +387,45 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         '''
         Handles caching of items once inheritance and any other one time
         per course per fetch operations are done.
-        :param system: a CachingDescriptorSystem
-        :param base_block_ids: list of block_ids to fetch
-        :param course_key: the destination course providing the context
-        :param depth: how deep below these to prefetch
-        :param lazy: whether to fetch definitions or use placeholders
+
+        Arguments:
+            system: a CachingDescriptorSystem
+            base_block_ids: list of block_ids to fetch
+            course_key: the destination course providing the context
+            depth: how deep below these to prefetch
+            lazy: whether to fetch definitions or use placeholders
         '''
-        new_module_data = {}
-        for block_id in base_block_ids:
-            new_module_data = self.descendants(
-                system.course_entry['structure']['blocks'],
-                block_id,
-                depth,
-                new_module_data
-            )
-
-        if lazy:
-            for block in new_module_data.itervalues():
-                block['definition'] = DefinitionLazyLoader(
-                    self, block['category'], block['definition'],
-                    lambda fields: self.convert_references_to_keys(
-                        course_key, system.load_block_type(block['category']),
-                        fields, system.course_entry['structure']['blocks'],
-                    )
+        with self.bulk_write_operations(course_key):
+            new_module_data = {}
+            for block_id in base_block_ids:
+                new_module_data = self.descendants(
+                    system.course_entry['structure']['blocks'],
+                    block_id,
+                    depth,
+                    new_module_data
                 )
-        else:
-            # Load all descendants by id
-            descendent_definitions = self.db_connection.find_matching_definitions({
-                '_id': {'$in': [block['definition']
-                                for block in new_module_data.itervalues()]}})
-            # turn into a map
-            definitions = {definition['_id']: definition
-                           for definition in descendent_definitions}
 
-            for block in new_module_data.itervalues():
-                if block['definition'] in definitions:
-                    converted_fields = self.convert_references_to_keys(
-                        course_key, system.load_block_type(block['category']),
-                        definitions[block['definition']].get('fields'),
-                        system.course_entry['structure']['blocks'],
-                    )
-                    block['fields'].update(converted_fields)
+            if not lazy:
+                # Load all descendants by id
+                descendent_definitions = self.db_connection.find_matching_definitions({
+                    '_id': {'$in': [block['definition']
+                                    for block in new_module_data.itervalues()]}})
+                # turn into a map
+                definitions = {definition['_id']: definition
+                               for definition in descendent_definitions}
 
-        system.module_data.update(new_module_data)
-        return system.module_data
+                for block in new_module_data.itervalues():
+                    if block['definition'] in definitions:
+                        converted_fields = self.convert_references_to_keys(
+                            course_key, system.load_block_type(block['category']),
+                            definitions[block['definition']].get('fields'),
+                            system.course_entry['structure']['blocks'],
+                        )
+                        block['fields'].update(converted_fields)
+                        block['definition_loaded'] = True
+
+            system.module_data.update(new_module_data)
+            return system.module_data
 
     def _load_items(self, course_entry, block_ids, depth=0, lazy=True, **kwargs):
         '''
@@ -265,6 +474,8 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         :param course_version_guid: if provided, clear only this entry
         """
         if course_version_guid:
+            if not hasattr(self.thread_cache, 'course_cache'):
+                self.thread_cache.course_cache = {}
             try:
                 del self.thread_cache.course_cache[course_version_guid]
             except KeyError:
@@ -272,7 +483,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         else:
             self.thread_cache.course_cache = {}
 
-    def _lookup_course(self, course_locator):
+    def _lookup_course(self, course_key):
         '''
         Decode the locator into the right series of db access. Does not
         return the CourseDescriptor! It returns the actual db json from
@@ -283,40 +494,45 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         it raises VersionConflictError (the version now differs from what it was when you got your
         reference)
 
-        :param course_locator: any subclass of CourseLocator
+        :param course_key: any subclass of CourseLocator
         '''
-        if course_locator.org and course_locator.course and course_locator.run:
-            if course_locator.branch is None:
-                raise InsufficientSpecificationError(course_locator)
+        if course_key.org and course_key.course and course_key.run:
+            if course_key.branch is None:
+                raise InsufficientSpecificationError(course_key)
+
             # use the course id
-            index = self.db_connection.get_course_index(course_locator)
+            index = self.get_course_index(course_key)
+
             if index is None:
-                raise ItemNotFoundError(course_locator)
-            if course_locator.branch not in index['versions']:
-                raise ItemNotFoundError(course_locator)
-            version_guid = index['versions'][course_locator.branch]
-            if course_locator.version_guid is not None and version_guid != course_locator.version_guid:
+                raise ItemNotFoundError(course_key)
+            if course_key.branch not in index['versions']:
+                raise ItemNotFoundError(course_key)
+
+            version_guid = index['versions'][course_key.branch]
+
+            if course_key.version_guid is not None and version_guid != course_key.version_guid:
                 # This may be a bit too touchy but it's hard to infer intent
-                raise VersionConflictError(course_locator, version_guid)
-        elif course_locator.version_guid is None:
-            raise InsufficientSpecificationError(course_locator)
+                raise VersionConflictError(course_key, version_guid)
+
+        elif course_key.version_guid is None:
+            raise InsufficientSpecificationError(course_key)
         else:
             # TODO should this raise an exception if branch was provided?
-            version_guid = course_locator.version_guid
+            version_guid = course_key.version_guid
 
-        # cast string to ObjectId if necessary
-        version_guid = course_locator.as_object_id(version_guid)
-        entry = self.db_connection.get_structure(version_guid)
+        entry = self.get_structure(course_key, version_guid)
+        if entry is None:
+            raise ItemNotFoundError('Structure: {}'.format(version_guid))
 
         # b/c more than one course can use same structure, the 'org', 'course',
         # 'run', and 'branch' are not intrinsic to structure
         # and the one assoc'd w/ it by another fetch may not be the one relevant to this fetch; so,
         # add it in the envelope for the structure.
         envelope = {
-            'org': course_locator.org,
-            'course': course_locator.course,
-            'run': course_locator.run,
-            'branch': course_locator.branch,
+            'org': course_key.org,
+            'course': course_key.course,
+            'run': course_key.run,
+            'branch': course_key.branch,
             'structure': entry,
         }
         return envelope
@@ -401,7 +617,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
             # The supplied CourseKey is of the wrong type, so it can't possibly be stored in this modulestore.
             return False
 
-        course_index = self.db_connection.get_course_index(course_id, ignore_case)
+        course_index = self.get_course_index(course_id, ignore_case)
         return CourseLocator(course_index['org'], course_index['course'], course_index['run'], course_id.branch) if course_index else None
 
     def has_item(self, usage_key):
@@ -413,7 +629,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         if usage_key.block_id is None:
             raise InsufficientSpecificationError(usage_key)
         try:
-            course_structure = self._lookup_course(usage_key)['structure']
+            course_structure = self._lookup_course(usage_key.course_key)['structure']
         except ItemNotFoundError:
             # this error only occurs if the course does not exist
             return False
@@ -433,7 +649,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
             # The supplied UsageKey is of the wrong type, so it can't possibly be stored in this modulestore.
             raise ItemNotFoundError(usage_key)
 
-        course = self._lookup_course(usage_key)
+        course = self._lookup_course(usage_key.course_key)
         items = self._load_items(course, [usage_key.block_id], depth, lazy=True, **kwargs)
         if len(items) == 0:
             raise ItemNotFoundError(usage_key)
@@ -511,7 +727,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
 
         :param locator: BlockUsageLocator restricting search scope
         '''
-        course = self._lookup_course(locator)
+        course = self._lookup_course(locator.course_key)
         parent_id = self._get_parent_from_structure(locator.block_id, course['structure'])
         if parent_id is None:
             return None
@@ -541,12 +757,12 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
             for block_id in items
         ]
 
-    def get_course_index_info(self, course_locator):
+    def get_course_index_info(self, course_key):
         """
         The index records the initial creation of the indexed course and tracks the current version
         heads. This function is primarily for test verification but may serve some
         more general purpose.
-        :param course_locator: must have a org, course, and run set
+        :param course_key: must have a org, course, and run set
         :return {'org': string,
             versions: {'draft': the head draft version id,
                 'published': the head published version id if any,
@@ -555,24 +771,24 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
             'edited_on': when the course was originally created
         }
         """
-        if not (course_locator.course and course_locator.run and course_locator.org):
+        if not (course_key.course and course_key.run and course_key.org):
             return None
-        index = self.db_connection.get_course_index(course_locator)
+        index = self.get_course_index(course_key)
         return index
 
     # TODO figure out a way to make this info accessible from the course descriptor
-    def get_course_history_info(self, course_locator):
+    def get_course_history_info(self, course_key):
         """
         Because xblocks doesn't give a means to separate the course structure's meta information from
         the course xblock's, this method will get that info for the structure as a whole.
-        :param course_locator:
+        :param course_key:
         :return {'original_version': the version guid of the original version of this course,
             'previous_version': the version guid of the previous version,
             'edited_by': who made the last change,
             'edited_on': when the change was made
         }
         """
-        course = self._lookup_course(course_locator)['structure']
+        course = self._lookup_course(course_key)['structure']
         return {
             'original_version': course['original_version'],
             'previous_version': course['previous_version'],
@@ -628,7 +844,6 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
                     CourseLocator(version_guid=struct['_id']))
         return VersionTree(course_locator, result)
 
-
     def get_block_generations(self, block_locator):
         '''
         Find the history of this block. Return as a VersionTree of each place the block changed (except
@@ -639,7 +854,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
 
         '''
         # course_agnostic means we don't care if the head and version don't align, trust the version
-        course_struct = self._lookup_course(block_locator.course_agnostic())['structure']
+        course_struct = self._lookup_course(block_locator.course_key.course_agnostic())['structure']
         block_id = block_locator.block_id
         update_version_field = 'blocks.{}.edit_info.update_version'.format(block_id)
         all_versions_with_block = self.db_connection.find_matching_structures(
@@ -772,7 +987,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
     def create_item(
         self, user_id, course_key, block_type, block_id=None,
         definition_locator=None, fields=None,
-        force=False, continue_version=False, **kwargs
+        force=False, **kwargs
     ):
         """
         Add a descriptor to persistence as an element
@@ -799,10 +1014,8 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         :param block_id: if provided, must not already exist in the structure. Provides the block id for the
         new item in this structure. Otherwise, one is computed using the category appended w/ a few digits.
 
-        :param continue_version: continue changing the current structure at the head of the course. Very dangerous
-        unless used in the same request as started the change! See below about version conflicts.
-
-        This method creates a new version of the course structure unless continue_version is True.
+        This method creates a new version of the course structure unless the course has a bulk_write operation
+        active.
         It creates and inserts the new block, makes the block point
         to the definition which may be new or a new version of an existing or an existing.
 
@@ -818,91 +1031,82 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         the course id'd by version_guid but instead in one w/ a new version_guid. Ensure in this case that you get
         the new version_guid from the locator in the returned object!
         """
-        # split handles all the fields in one dict not separated by scope
-        fields = fields or {}
-        fields.update(kwargs.pop('metadata', {}) or {})
-        definition_data = kwargs.pop('definition_data', {})
-        if definition_data:
-            if not isinstance(definition_data, dict):
-                definition_data = {'data': definition_data}  # backward compatibility to mongo's hack
-            fields.update(definition_data)
+        with self.bulk_write_operations(course_key):
+            # split handles all the fields in one dict not separated by scope
+            fields = fields or {}
+            fields.update(kwargs.pop('metadata', {}) or {})
+            definition_data = kwargs.pop('definition_data', {})
+            if definition_data:
+                if not isinstance(definition_data, dict):
+                    definition_data = {'data': definition_data}  # backward compatibility to mongo's hack
+                fields.update(definition_data)
 
-        # find course_index entry if applicable and structures entry
-        index_entry = self._get_index_if_valid(course_key, force, continue_version)
-        structure = self._lookup_course(course_key)['structure']
+            # find course_index entry if applicable and structures entry
+            index_entry = self._get_index_if_valid(course_key, force)
+            structure = self._lookup_course(course_key)['structure']
 
-        partitioned_fields = self.partition_fields_by_scope(block_type, fields)
-        new_def_data = partitioned_fields.get(Scope.content, {})
-        # persist the definition if persisted != passed
-        if (definition_locator is None or isinstance(definition_locator.definition_id, LocalId)):
-            definition_locator = self.create_definition_from_data(new_def_data, block_type, user_id)
-        elif new_def_data is not None:
-            definition_locator, _ = self.update_definition_from_data(definition_locator, new_def_data, user_id)
+            partitioned_fields = self.partition_fields_by_scope(block_type, fields)
+            new_def_data = partitioned_fields.get(Scope.content, {})
+            # persist the definition if persisted != passed
+            if (definition_locator is None or isinstance(definition_locator.definition_id, LocalId)):
+                definition_locator = self.create_definition_from_data(new_def_data, block_type, user_id)
+            elif new_def_data is not None:
+                definition_locator, _ = self.update_definition_from_data(definition_locator, new_def_data, user_id)
 
-        # copy the structure and modify the new one
-        if continue_version:
-            new_structure = structure
-        else:
-            new_structure = self._version_structure(structure, user_id)
+            # copy the structure and modify the new one
+            new_structure = self.version_structure(course_key, structure, user_id)
 
-        new_id = new_structure['_id']
+            new_id = new_structure['_id']
 
-        edit_info = {
-            'edited_on': datetime.datetime.now(UTC),
-            'edited_by': user_id,
-            'previous_version': None,
-            'update_version': new_id,
-        }
-        # generate usage id
-        if block_id is not None:
-            if encode_key_for_mongo(block_id) in new_structure['blocks']:
-                raise DuplicateItemError(block_id, self, 'structures')
+            edit_info = {
+                'edited_on': datetime.datetime.now(UTC),
+                'edited_by': user_id,
+                'previous_version': None,
+                'update_version': new_id,
+            }
+            # generate usage id
+            if block_id is not None:
+                if encode_key_for_mongo(block_id) in new_structure['blocks']:
+                    raise DuplicateItemError(block_id, self, 'structures')
+                else:
+                    new_block_id = block_id
             else:
-                new_block_id = block_id
-        else:
-            new_block_id = self._generate_block_id(new_structure['blocks'], block_type)
+                new_block_id = self._generate_block_id(new_structure['blocks'], block_type)
 
-        block_fields = partitioned_fields.get(Scope.settings, {})
-        if Scope.children in partitioned_fields:
-            block_fields.update(partitioned_fields[Scope.children])
-        self._update_block_in_structure(new_structure, new_block_id, {
-            "category": block_type,
-            "definition": definition_locator.definition_id,
-            "fields": self._serialize_fields(block_type, block_fields),
-            'edit_info': edit_info,
-        })
+            block_fields = partitioned_fields.get(Scope.settings, {})
+            if Scope.children in partitioned_fields:
+                block_fields.update(partitioned_fields[Scope.children])
+            self._update_block_in_structure(new_structure, new_block_id, {
+                "category": block_type,
+                "definition": definition_locator.definition_id,
+                "fields": self._serialize_fields(block_type, block_fields),
+                'edit_info': edit_info,
+            })
 
-        if continue_version:
-            # db update
-            self.db_connection.update_structure(new_structure)
-            # clear cache so things get refetched and inheritance recomputed
-            self._clear_cache(new_id)
-        else:
-            self.db_connection.insert_structure(new_structure)
+            self.update_structure(course_key, new_structure)
 
-        # update the index entry if appropriate
-        if index_entry is not None:
-            # see if any search targets changed
-            if fields is not None:
-                self._update_search_targets(index_entry, fields)
-            if not continue_version:
-                self._update_head(index_entry, course_key.branch, new_id)
-            item_loc = BlockUsageLocator(
-                course_key.version_agnostic(),
-                block_type=block_type,
-                block_id=new_block_id,
-            )
-        else:
-            item_loc = BlockUsageLocator(
-                CourseLocator(version_guid=new_id),
-                block_type=block_type,
-                block_id=new_block_id,
-            )
+            # update the index entry if appropriate
+            if index_entry is not None:
+                # see if any search targets changed
+                if fields is not None:
+                    self._update_search_targets(index_entry, fields)
+                self._update_head(course_key, index_entry, course_key.branch, new_id)
+                item_loc = BlockUsageLocator(
+                    course_key.version_agnostic(),
+                    block_type=block_type,
+                    block_id=new_block_id,
+                )
+            else:
+                item_loc = BlockUsageLocator(
+                    CourseLocator(version_guid=new_id),
+                    block_type=block_type,
+                    block_id=new_block_id,
+                )
 
-        # reconstruct the new_item from the cache
-        return self.get_item(item_loc)
+            # reconstruct the new_item from the cache
+            return self.get_item(item_loc)
 
-    def create_child(self, user_id, parent_usage_key, block_type, block_id=None, fields=None, continue_version=False, **kwargs):
+    def create_child(self, user_id, parent_usage_key, block_type, block_id=None, fields=None, **kwargs):
         """
         Creates and saves a new xblock that as a child of the specified block
 
@@ -918,32 +1122,33 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
             fields (dict): A dictionary specifying initial values for some or all fields
                 in the newly created block
         """
-        xblock = self.create_item(
-            user_id, parent_usage_key.course_key, block_type, block_id=block_id, fields=fields,
-            continue_version=continue_version,
-            **kwargs)
+        with self.bulk_write_operations(parent_usage_key.course_key):
+            xblock = self.create_item(
+                user_id, parent_usage_key.course_key, block_type, block_id=block_id, fields=fields,
+                **kwargs)
 
-        # don't version the structure as create_item handled that already.
-        new_structure = self._lookup_course(xblock.location.course_key)['structure']
+            # don't version the structure as create_item handled that already.
+            new_structure = self._lookup_course(xblock.location.course_key)['structure']
 
-        # add new block as child and update parent's version
-        encoded_block_id = encode_key_for_mongo(parent_usage_key.block_id)
-        parent = new_structure['blocks'][encoded_block_id]
-        parent['fields'].setdefault('children', []).append(xblock.location.block_id)
-        if parent['edit_info']['update_version'] != new_structure['_id']:
-            # if the parent hadn't been previously changed in this bulk transaction, indicate that it's
-            # part of the bulk transaction
-            parent['edit_info'] = {
-                'edited_on': datetime.datetime.now(UTC),
-                'edited_by': user_id,
-                'previous_version': parent['edit_info']['update_version'],
-                'update_version': new_structure['_id'],
-            }
+            # add new block as child and update parent's version
+            encoded_block_id = encode_key_for_mongo(parent_usage_key.block_id)
+            if encoded_block_id not in new_structure['blocks']:
+                raise ItemNotFoundError(parent_usage_key)
 
-        # db update
-        self.db_connection.update_structure(new_structure)
-            # clear cache so things get refetched and inheritance recomputed
-        self._clear_cache(new_structure['_id'])
+            parent = new_structure['blocks'][encoded_block_id]
+            parent['fields'].setdefault('children', []).append(xblock.location.block_id)
+            if parent['edit_info']['update_version'] != new_structure['_id']:
+                # if the parent hadn't been previously changed in this bulk transaction, indicate that it's
+                # part of the bulk transaction
+                parent['edit_info'] = {
+                    'edited_on': datetime.datetime.now(UTC),
+                    'edited_by': user_id,
+                    'previous_version': parent['edit_info']['update_version'],
+                    'update_version': new_structure['_id'],
+                }
+
+            # db update
+            self.update_structure(parent_usage_key.course_key, new_structure)
 
         # don't need to update the index b/c create_item did it for this version
         return xblock
@@ -1023,7 +1228,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         assert master_branch is not None
         # check course and run's uniqueness
         locator = CourseLocator(org=org, course=course, run=run, branch=master_branch)
-        index = self.db_connection.get_course_index(locator)
+        index = self.get_course_index(locator)
         if index is not None:
             raise DuplicateCourseError(locator, index)
 
@@ -1061,18 +1266,16 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
             )
             new_id = draft_structure['_id']
 
-            self.db_connection.insert_structure(draft_structure)
-
             if versions_dict is None:
                 versions_dict = {master_branch: new_id}
             else:
                 versions_dict[master_branch] = new_id
 
-        elif definition_fields or block_fields:  # pointing to existing course w/ some overrides
+        elif block_fields or definition_fields:  # pointing to existing course w/ some overrides
             # just get the draft_version structure
             draft_version = CourseLocator(version_guid=versions_dict[master_branch])
             draft_structure = self._lookup_course(draft_version)['structure']
-            draft_structure = self._version_structure(draft_structure, user_id)
+            draft_structure = self.version_structure(locator, draft_structure, user_id)
             new_id = draft_structure['_id']
             encoded_block_id = encode_key_for_mongo(draft_structure['root'])
             root_block = draft_structure['blocks'][encoded_block_id]
@@ -1093,27 +1296,33 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
                 root_block['edit_info']['previous_version'] = root_block['edit_info'].get('update_version')
                 root_block['edit_info']['update_version'] = new_id
 
-            self.db_connection.insert_structure(draft_structure)
             versions_dict[master_branch] = new_id
+        else:  # Pointing to an existing course structure
+            new_id = versions_dict[master_branch]
+            draft_version = CourseLocator(version_guid=new_id)
+            draft_structure = self._lookup_course(draft_version)['structure']
 
-        index_entry = {
-            '_id': ObjectId(),
-            'org': org,
-            'course': course,
-            'run': run,
-            'edited_by': user_id,
-            'edited_on': datetime.datetime.now(UTC),
-            'versions': versions_dict,
-            'schema_version': self.SCHEMA_VERSION,
-            'search_targets': search_targets or {},
-        }
-        if fields is not None:
-            self._update_search_targets(index_entry, fields)
-        self.db_connection.insert_course_index(index_entry)
+        locator = locator.replace(version_guid=new_id)
+        with self.bulk_write_operations(locator):
+            self.update_structure(locator, draft_structure)
+            index_entry = {
+                '_id': ObjectId(),
+                'org': org,
+                'course': course,
+                'run': run,
+                'edited_by': user_id,
+                'edited_on': datetime.datetime.now(UTC),
+                'versions': versions_dict,
+                'schema_version': self.SCHEMA_VERSION,
+                'search_targets': search_targets or {},
+            }
+            if fields is not None:
+                self._update_search_targets(index_entry, fields)
+            self.insert_course_index(locator, index_entry)
 
-        # expensive hack to persist default field values set in __init__ method (e.g., wiki_slug)
-        course = self.get_course(locator, **kwargs)
-        return self.update_item(course, user_id, **kwargs)
+            # expensive hack to persist default field values set in __init__ method (e.g., wiki_slug)
+            course = self.get_course(locator, **kwargs)
+            return self.update_item(course, user_id, **kwargs)
 
     def update_item(self, descriptor, user_id, allow_not_found=False, force=False, **kwargs):
         """
@@ -1143,87 +1352,88 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         """
         Broke out guts of update_item for short-circuited internal use only
         """
-        if allow_not_found and isinstance(block_id, (LocalId, NoneType)):
-            fields = {}
-            for subfields in partitioned_fields.itervalues():
-                fields.update(subfields)
-            return self.create_item(
-                user_id, course_key, block_type, fields=fields, force=force
-            )
-
-        original_structure = self._lookup_course(course_key)['structure']
-        index_entry = self._get_index_if_valid(course_key, force)
-
-        original_entry = self._get_block_from_structure(original_structure, block_id)
-        if original_entry is None:
-            if allow_not_found:
+        with self.bulk_write_operations(course_key):
+            if allow_not_found and isinstance(block_id, (LocalId, NoneType)):
                 fields = {}
                 for subfields in partitioned_fields.itervalues():
                     fields.update(subfields)
                 return self.create_item(
-                    user_id, course_key, block_type, block_id=block_id, fields=fields, force=force,
+                    user_id, course_key, block_type, fields=fields, force=force
                 )
-            else:
-                raise ItemNotFoundError(course_key.make_usage_key(block_type, block_id))
 
-        is_updated = False
-        definition_fields = partitioned_fields[Scope.content]
-        if definition_locator is None:
-            definition_locator = DefinitionLocator(original_entry['category'], original_entry['definition'])
-        if definition_fields:
-            definition_locator, is_updated = self.update_definition_from_data(
-                definition_locator, definition_fields, user_id
-            )
+            original_structure = self._lookup_course(course_key)['structure']
+            index_entry = self._get_index_if_valid(course_key, force)
 
-        # check metadata
-        settings = partitioned_fields[Scope.settings]
-        settings = self._serialize_fields(block_type, settings)
-        if not is_updated:
-            is_updated = self._compare_settings(settings, original_entry['fields'])
+            original_entry = self._get_block_from_structure(original_structure, block_id)
+            if original_entry is None:
+                if allow_not_found:
+                    fields = {}
+                    for subfields in partitioned_fields.itervalues():
+                        fields.update(subfields)
+                    return self.create_item(
+                        user_id, course_key, block_type, block_id=block_id, fields=fields, force=force,
+                    )
+                else:
+                    raise ItemNotFoundError(course_key.make_usage_key(block_type, block_id))
 
-        # check children
-        if partitioned_fields.get(Scope.children, {}):  # purposely not 'is not None'
-            serialized_children = [child.block_id for child in partitioned_fields[Scope.children]['children']]
-            is_updated = is_updated or original_entry['fields'].get('children', []) != serialized_children
+            is_updated = False
+            definition_fields = partitioned_fields[Scope.content]
+            if definition_locator is None:
+                definition_locator = DefinitionLocator(original_entry['category'], original_entry['definition'])
+            if definition_fields:
+                definition_locator, is_updated = self.update_definition_from_data(
+                    definition_locator, definition_fields, user_id
+                )
+
+            # check metadata
+            settings = partitioned_fields[Scope.settings]
+            settings = self._serialize_fields(block_type, settings)
+            if not is_updated:
+                is_updated = self._compare_settings(settings, original_entry['fields'])
+
+            # check children
+            if partitioned_fields.get(Scope.children, {}):  # purposely not 'is not None'
+                serialized_children = [child.block_id for child in partitioned_fields[Scope.children]['children']]
+                is_updated = is_updated or original_entry['fields'].get('children', []) != serialized_children
+                if is_updated:
+                    settings['children'] = serialized_children
+
+            # if updated, rev the structure
             if is_updated:
-                settings['children'] = serialized_children
+                new_structure = self.version_structure(course_key, original_structure, user_id)
+                block_data = self._get_block_from_structure(new_structure, block_id)
 
-        # if updated, rev the structure
-        if is_updated:
-            new_structure = self._version_structure(original_structure, user_id)
-            block_data = self._get_block_from_structure(new_structure, block_id)
+                block_data["definition"] = definition_locator.definition_id
+                block_data["fields"] = settings
 
-            block_data["definition"] = definition_locator.definition_id
-            block_data["fields"] = settings
+                new_id = new_structure['_id']
+                block_data['edit_info'] = {
+                    'edited_on': datetime.datetime.now(UTC),
+                    'edited_by': user_id,
+                    'previous_version': block_data['edit_info']['update_version'],
+                    'update_version': new_id,
+                }
+                self.update_structure(course_key, new_structure)
+                # update the index entry if appropriate
+                if index_entry is not None:
+                    self._update_search_targets(index_entry, definition_fields)
+                    self._update_search_targets(index_entry, settings)
+                    course_key = CourseLocator(
+                        org=index_entry['org'],
+                        course=index_entry['course'],
+                        run=index_entry['run'],
+                        branch=course_key.branch,
+                        version_guid=new_id
+                    )
+                    self._update_head(course_key, index_entry, course_key.branch, new_id)
+                else:
+                    course_key = CourseLocator(version_guid=new_id)
 
-            new_id = new_structure['_id']
-            block_data['edit_info'] = {
-                'edited_on': datetime.datetime.now(UTC),
-                'edited_by': user_id,
-                'previous_version': block_data['edit_info']['update_version'],
-                'update_version': new_id,
-            }
-            self.db_connection.insert_structure(new_structure)
-            # update the index entry if appropriate
-            if index_entry is not None:
-                self._update_search_targets(index_entry, definition_fields)
-                self._update_search_targets(index_entry, settings)
-                self._update_head(index_entry, course_key.branch, new_id)
-                course_key = CourseLocator(
-                    org=index_entry['org'],
-                    course=index_entry['course'],
-                    run=index_entry['run'],
-                    branch=course_key.branch,
-                    version_guid=new_id
-                )
+                # fetch and return the new item--fetching is unnecessary but a good qc step
+                new_locator = course_key.make_usage_key(block_type, block_id)
+                return self.get_item(new_locator, **kwargs)
             else:
-                course_key = CourseLocator(version_guid=new_id)
-
-            # fetch and return the new item--fetching is unnecessary but a good qc step
-            new_locator = course_key.make_usage_key(block_type, block_id)
-            return self.get_item(new_locator, **kwargs)
-        else:
-            return None
+                return None
 
     # pylint: disable=unused-argument
     def create_xblock(
@@ -1286,23 +1496,25 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         :param user_id: who's doing the change
         """
         # find course_index entry if applicable and structures entry
-        index_entry = self._get_index_if_valid(xblock.location, force)
-        structure = self._lookup_course(xblock.location)['structure']
-        new_structure = self._version_structure(structure, user_id)
-        new_id = new_structure['_id']
-        is_updated = self._persist_subdag(xblock, user_id, new_structure['blocks'], new_id)
+        course_key = xblock.location.course_key
+        with self.bulk_write_operations(course_key):
+            index_entry = self._get_index_if_valid(course_key, force)
+            structure = self._lookup_course(course_key)['structure']
+            new_structure = self.version_structure(course_key, structure, user_id)
+            new_id = new_structure['_id']
+            is_updated = self._persist_subdag(xblock, user_id, new_structure['blocks'], new_id)
 
-        if is_updated:
-            self.db_connection.insert_structure(new_structure)
+            if is_updated:
+                self.update_structure(course_key, new_structure)
 
-            # update the index entry if appropriate
-            if index_entry is not None:
-                self._update_head(index_entry, xblock.location.branch, new_id)
+                # update the index entry if appropriate
+                if index_entry is not None:
+                    self._update_head(course_key, index_entry, xblock.location.branch, new_id)
 
-            # fetch and return the new item--fetching is unnecessary but a good qc step
-            return self.get_item(xblock.location.for_version(new_id))
-        else:
-            return xblock
+                # fetch and return the new item--fetching is unnecessary but a good qc step
+                return self.get_item(xblock.location.for_version(new_id))
+            else:
+                return xblock
 
     def _persist_subdag(self, xblock, user_id, structure_blocks, new_id):
         # persist the definition if persisted != passed
@@ -1415,71 +1627,63 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
                 subtree but the ancestors up to and including the course root are not published.
         """
         # get the destination's index, and source and destination structures.
-        source_structure = self._lookup_course(source_course)['structure']
-        index_entry = self.db_connection.get_course_index(destination_course)
-        if index_entry is None:
-            # brand new course
-            raise ItemNotFoundError(destination_course)
-        if destination_course.branch not in index_entry['versions']:
-            # must be copying the dag root if there's no current dag
-            root_block_id = source_structure['root']
-            if not any(root_block_id == subtree.block_id for subtree in subtree_list):
-                raise ItemNotFoundError(u'Must publish course root {}'.format(root_block_id))
-            root_source = source_structure['blocks'][root_block_id]
-            # create branch
-            destination_structure = self._new_structure(
-                user_id, root_block_id, root_category=root_source['category'],
-                # leave off the fields b/c the children must be filtered
-                definition_id=root_source['definition'],
-            )
-        else:
-            destination_structure = self._lookup_course(destination_course)['structure']
-            destination_structure = self._version_structure(destination_structure, user_id)
+        with self.bulk_write_operations(source_course):
+            with self.bulk_write_operations(destination_course):
+                source_structure = self._lookup_course(source_course)['structure']
+                index_entry = self.get_course_index(destination_course)
+                if index_entry is None:
+                    # brand new course
+                    raise ItemNotFoundError(destination_course)
+                if destination_course.branch not in index_entry['versions']:
+                    # must be copying the dag root if there's no current dag
+                    root_block_id = source_structure['root']
+                    if not any(root_block_id == subtree.block_id for subtree in subtree_list):
+                        raise ItemNotFoundError(u'Must publish course root {}'.format(root_block_id))
+                    root_source = source_structure['blocks'][root_block_id]
+                    # create branch
+                    destination_structure = self._new_structure(
+                        user_id, root_block_id, root_category=root_source['category'],
+                        # leave off the fields b/c the children must be filtered
+                        definition_id=root_source['definition'],
+                    )
+                else:
+                    destination_structure = self._lookup_course(destination_course)['structure']
+                    destination_structure = self.version_structure(destination_course, destination_structure, user_id)
 
-        if blacklist != EXCLUDE_ALL:
-            blacklist = [shunned.block_id for shunned in blacklist or []]
-        # iterate over subtree list filtering out blacklist.
-        orphans = set()
-        destination_blocks = destination_structure['blocks']
-        for subtree_root in subtree_list:
-            if subtree_root.block_id != source_structure['root']:
-                # find the parents and put root in the right sequence
-                parent = self._get_parent_from_structure(subtree_root.block_id, source_structure)
-                if parent is not None:  # may be a detached category xblock
-                    if not parent in destination_blocks:
-                        raise ItemNotFoundError(parent)
+                if blacklist != EXCLUDE_ALL:
+                    blacklist = [shunned.block_id for shunned in blacklist or []]
+                # iterate over subtree list filtering out blacklist.
+                orphans = set()
+                destination_blocks = destination_structure['blocks']
+                for subtree_root in subtree_list:
+                    if subtree_root.block_id != source_structure['root']:
+                        # find the parents and put root in the right sequence
+                        parent = self._get_parent_from_structure(subtree_root.block_id, source_structure)
+                        if parent is not None:  # may be a detached category xblock
+                            if not parent in destination_blocks:
+                                raise ItemNotFoundError(parent)
+                            orphans.update(
+                                self._sync_children(
+                                    source_structure['blocks'][parent],
+                                    destination_blocks[parent],
+                                    subtree_root.block_id
+                                )
+                            )
+                    # update/create the subtree and its children in destination (skipping blacklist)
                     orphans.update(
-                        self._sync_children(
-                            source_structure['blocks'][parent],
-                            destination_blocks[parent],
-                            subtree_root.block_id
+                        self._copy_subdag(
+                            user_id, destination_structure['_id'],
+                            subtree_root.block_id, source_structure['blocks'], destination_blocks, blacklist
                         )
                     )
-            # update/create the subtree and its children in destination (skipping blacklist)
-            orphans.update(
-                self._copy_subdag(
-                    user_id, destination_structure['_id'],
-                    subtree_root.block_id, source_structure['blocks'], destination_blocks, blacklist
-                )
-            )
-        # remove any remaining orphans
-        for orphan in orphans:
-            # orphans will include moved as well as deleted xblocks. Only delete the deleted ones.
-            self._delete_if_true_orphan(orphan, destination_structure)
+                # remove any remaining orphans
+                for orphan in orphans:
+                    # orphans will include moved as well as deleted xblocks. Only delete the deleted ones.
+                    self._delete_if_true_orphan(orphan, destination_structure)
 
-        # update the db
-        self.db_connection.insert_structure(destination_structure)
-        self._update_head(index_entry, destination_course.branch, destination_structure['_id'])
-
-    def update_course_index(self, updated_index_entry):
-        """
-        Change the given course's index entry.
-
-        Note, this operation can be dangerous and break running courses.
-
-        Does not return anything useful.
-        """
-        self.db_connection.update_course_index(updated_index_entry)
+                # update the db
+                self.update_structure(destination_course, destination_structure)
+                self._update_head(destination_course, index_entry, destination_course.branch, destination_structure['_id'])
 
     def delete_item(self, usage_locator, user_id, force=False):
         """
@@ -1500,46 +1704,47 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
             # The supplied UsageKey is of the wrong type, so it can't possibly be stored in this modulestore.
             raise ItemNotFoundError(usage_locator)
 
-        original_structure = self._lookup_course(usage_locator.course_key)['structure']
-        if original_structure['root'] == usage_locator.block_id:
-            raise ValueError("Cannot delete the root of a course")
-        if encode_key_for_mongo(usage_locator.block_id) not in original_structure['blocks']:
-            raise ValueError("Cannot delete a block that does not exist")
-        index_entry = self._get_index_if_valid(usage_locator, force)
-        new_structure = self._version_structure(original_structure, user_id)
-        new_blocks = new_structure['blocks']
-        new_id = new_structure['_id']
-        encoded_block_id = self._get_parent_from_structure(usage_locator.block_id, original_structure)
-        if encoded_block_id:
-            parent_block = new_blocks[encoded_block_id]
-            parent_block['fields']['children'].remove(usage_locator.block_id)
-            parent_block['edit_info']['edited_on'] = datetime.datetime.now(UTC)
-            parent_block['edit_info']['edited_by'] = user_id
-            parent_block['edit_info']['previous_version'] = parent_block['edit_info']['update_version']
-            parent_block['edit_info']['update_version'] = new_id
+        with self.bulk_write_operations(usage_locator.course_key):
+            original_structure = self._lookup_course(usage_locator.course_key)['structure']
+            if original_structure['root'] == usage_locator.block_id:
+                raise ValueError("Cannot delete the root of a course")
+            if encode_key_for_mongo(usage_locator.block_id) not in original_structure['blocks']:
+                raise ValueError("Cannot delete a block that does not exist")
+            index_entry = self._get_index_if_valid(usage_locator.course_key, force)
+            new_structure = self.version_structure(usage_locator.course_key, original_structure, user_id)
+            new_blocks = new_structure['blocks']
+            new_id = new_structure['_id']
+            encoded_block_id = self._get_parent_from_structure(usage_locator.block_id, original_structure)
+            if encoded_block_id:
+                parent_block = new_blocks[encoded_block_id]
+                parent_block['fields']['children'].remove(usage_locator.block_id)
+                parent_block['edit_info']['edited_on'] = datetime.datetime.now(UTC)
+                parent_block['edit_info']['edited_by'] = user_id
+                parent_block['edit_info']['previous_version'] = parent_block['edit_info']['update_version']
+                parent_block['edit_info']['update_version'] = new_id
 
-        def remove_subtree(block_id):
-            """
-            Remove the subtree rooted at block_id
-            """
-            encoded_block_id = encode_key_for_mongo(block_id)
-            for child in new_blocks[encoded_block_id]['fields'].get('children', []):
-                remove_subtree(child)
-            del new_blocks[encoded_block_id]
+            def remove_subtree(block_id):
+                """
+                Remove the subtree rooted at block_id
+                """
+                encoded_block_id = encode_key_for_mongo(block_id)
+                for child in new_blocks[encoded_block_id]['fields'].get('children', []):
+                    remove_subtree(child)
+                del new_blocks[encoded_block_id]
 
-        remove_subtree(usage_locator.block_id)
+            remove_subtree(usage_locator.block_id)
 
-        # update index if appropriate and structures
-        self.db_connection.insert_structure(new_structure)
+            # update index if appropriate and structures
+            self.update_structure(usage_locator.course_key, new_structure)
 
-        if index_entry is not None:
-            # update the index entry if appropriate
-            self._update_head(index_entry, usage_locator.branch, new_id)
-            result = usage_locator.course_key.for_version(new_id)
-        else:
-            result = CourseLocator(version_guid=new_id)
+            if index_entry is not None:
+                # update the index entry if appropriate
+                self._update_head(usage_locator.course_key, index_entry, usage_locator.branch, new_id)
+                result = usage_locator.course_key.for_version(new_id)
+            else:
+                result = CourseLocator(version_guid=new_id)
 
-        return result
+            return result
 
     def delete_course(self, course_key, user_id):
         """
@@ -1549,7 +1754,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         with a versions hash to restore the course; however, the edited_on and
         edited_by won't reflect the originals, of course.
         """
-        index = self.db_connection.get_course_index(course_key)
+        index = self.get_course_index(course_key)
         if index is None:
             raise ItemNotFoundError(course_key)
         # this is the only real delete in the system. should it do something else?
@@ -1610,23 +1815,10 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         if depth is None or depth > 0:
             depth = depth - 1 if depth is not None else None
             for child in descendent_map[block_id]['fields'].get('children', []):
-                descendent_map = self.descendants(block_map, child, depth,
-                    descendent_map)
+                descendent_map = self.descendants(block_map, child, depth, descendent_map)
 
         return descendent_map
 
-    def definition_locator(self, definition):
-        '''
-        Pull the id out of the definition w/ correct semantics for its
-        representation
-        '''
-        if isinstance(definition, DefinitionLazyLoader):
-            return definition.definition_locator
-        elif '_id' not in definition:
-            return DefinitionLocator(definition.get('category'), LocalId())
-        else:
-            return DefinitionLocator(definition['category'], definition['_id'])
-
     def get_modulestore_type(self, course_key=None):
         """
         Returns an enumeration-like type reflecting the type of this modulestore, per ModuleStoreEnum.Type.
@@ -1651,9 +1843,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
                     block_id for block_id in block['fields']["children"]
                     if encode_key_for_mongo(block_id) in original_structure['blocks']
                 ]
-        self.db_connection.update_structure(original_structure)
-        # clear cache again b/c inheritance may be wrong over orphans
-        self._clear_cache(original_structure['_id'])
+        self.update_structure(course_locator, original_structure)
 
     def convert_references_to_keys(self, course_key, xblock_class, jsonfields, blocks):
         """
@@ -1681,69 +1871,50 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
                 return course_key.make_usage_key('unknown', block_id)
 
         xblock_class = self.mixologist.mix(xblock_class)
-        for field_name, value in jsonfields.iteritems():
+        # Make a shallow copy, so that we aren't manipulating a cached field dictionary
+        output_fields = dict(jsonfields)
+        for field_name, value in output_fields.iteritems():
             if value:
                 field = xblock_class.fields.get(field_name)
                 if field is None:
                     continue
                 elif isinstance(field, Reference):
-                    jsonfields[field_name] = robust_usage_key(value)
+                    output_fields[field_name] = robust_usage_key(value)
                 elif isinstance(field, ReferenceList):
-                    jsonfields[field_name] = [robust_usage_key(ele) for ele in value]
+                    output_fields[field_name] = [robust_usage_key(ele) for ele in value]
                 elif isinstance(field, ReferenceValueDict):
                     for key, subvalue in value.iteritems():
                         assert isinstance(subvalue, basestring)
                         value[key] = robust_usage_key(subvalue)
-        return jsonfields
+        return output_fields
 
-    def _get_index_if_valid(self, locator, force=False, continue_version=False):
+    def _get_index_if_valid(self, course_key, force=False):
         """
-        If the locator identifies a course and points to its draft (or plausibly its draft),
+        If the course_key identifies a course and points to its draft (or plausibly its draft),
         then return the index entry.
 
         raises VersionConflictError if not the right version
 
-        :param locator: a courselocator
+        :param course_key: a CourseLocator
         :param force: if false, raises VersionConflictError if the current head of the course != the one identified
-        by locator. Cannot be True if continue_version is True
-        :param continue_version: if True, assumes this operation requires a head version and will not create a new
-        version but instead continue an existing transaction on this version. This flag cannot be True if force is True.
+        by course_key
         """
-        if locator.org is None or locator.course is None or locator.run is None or locator.branch is None:
-            if continue_version:
-                raise InsufficientSpecificationError(
-                    "To continue a version, the locator must point to one ({}).".format(locator)
-                )
-            else:
-                return None
+        if course_key.org is None or course_key.course is None or course_key.run is None or course_key.branch is None:
+            return None
         else:
-            index_entry = self.db_connection.get_course_index(locator)
+            index_entry = self.get_course_index(course_key)
             is_head = (
-                locator.version_guid is None or
-                index_entry['versions'][locator.branch] == locator.version_guid
+                course_key.version_guid is None or
+                index_entry['versions'][course_key.branch] == course_key.version_guid
             )
-            if (is_head or (force and not continue_version)):
+            if (is_head or force):
                 return index_entry
             else:
                 raise VersionConflictError(
-                    locator,
-                    index_entry['versions'][locator.branch]
+                    course_key,
+                    index_entry['versions'][course_key.branch]
                 )
 
-    def _version_structure(self, structure, user_id):
-        """
-        Copy the structure and update the history info (edited_by, edited_on, previous_version)
-        :param structure:
-        :param user_id:
-        """
-        new_structure = copy.deepcopy(structure)
-        new_structure['_id'] = ObjectId()
-        new_structure['previous_version'] = structure['_id']
-        new_structure['edited_by'] = user_id
-        new_structure['edited_on'] = datetime.datetime.now(UTC)
-        new_structure['schema_version'] = self.SCHEMA_VERSION
-        return new_structure
-
     def _find_local_root(self, element_to_find, possibility, tree):
         if possibility not in tree:
             return False
@@ -1766,7 +1937,7 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
             if field_name in self.SEARCH_TARGET_DICT:
                 index_entry.setdefault('search_targets', {})[field_name] = field_value
 
-    def _update_head(self, index_entry, branch, new_id):
+    def _update_head(self, course_key, index_entry, branch, new_id):
         """
         Update the active index for the given course's branch to point to new_id
 
@@ -1774,8 +1945,10 @@ class SplitMongoModuleStore(ModuleStoreWriteBase):
         :param course_locator:
         :param new_id:
         """
+        if not isinstance(new_id, ObjectId):
+            raise TypeError('new_id must be an ObjectId, but is {!r}'.format(new_id))
         index_entry['versions'][branch] = new_id
-        self.db_connection.update_course_index(index_entry)
+        self.insert_course_index(course_key, index_entry)
 
     def partition_xblock_fields_by_scope(self, xblock):
         """

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

@@ -9,6 +9,7 @@ from xmodule.modulestore.exceptions import InsufficientSpecificationError
 from xmodule.modulestore.draft_and_published import (
     ModuleStoreDraftAndPublished, DIRECT_ONLY_CATEGORIES, UnsupportedRevisionError
 )
+from opaque_keys.edx.locator import CourseLocator
 
 
 class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleStore):
@@ -30,24 +31,25 @@ class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleS
         Returns: a CourseDescriptor
         """
         master_branch = kwargs.pop('master_branch', ModuleStoreEnum.BranchName.draft)
-        item = super(DraftVersioningModuleStore, self).create_course(
-            org, course, run, user_id, master_branch=master_branch, **kwargs
-        )
-        if master_branch == ModuleStoreEnum.BranchName.draft and not skip_auto_publish:
-            # any other value is hopefully only cloning or doing something which doesn't want this value add
-            self._auto_publish_no_children(item.location, item.location.category, user_id, **kwargs)
+        with self.bulk_write_operations(CourseLocator(org, course, run)):
+            item = super(DraftVersioningModuleStore, self).create_course(
+                org, course, run, user_id, master_branch=master_branch, **kwargs
+            )
+            if master_branch == ModuleStoreEnum.BranchName.draft and not skip_auto_publish:
+                # any other value is hopefully only cloning or doing something which doesn't want this value add
+                self._auto_publish_no_children(item.location, item.location.category, user_id, **kwargs)
 
-            # create any other necessary things as a side effect: ensure they populate the draft branch
-            # and rely on auto publish to populate the published branch: split's create course doesn't
-            # call super b/c it needs the auto publish above to have happened before any of the create_items
-            # in this. The explicit use of SplitMongoModuleStore is intentional
-            with self.branch_setting(ModuleStoreEnum.Branch.draft_preferred, item.id):
-                # pylint: disable=bad-super-call
-                super(SplitMongoModuleStore, self).create_course(
-                    org, course, run, user_id, runtime=item.runtime, **kwargs
-                )
+                # create any other necessary things as a side effect: ensure they populate the draft branch
+                # and rely on auto publish to populate the published branch: split's create course doesn't
+                # call super b/c it needs the auto publish above to have happened before any of the create_items
+                # in this. The explicit use of SplitMongoModuleStore is intentional
+                with self.branch_setting(ModuleStoreEnum.Branch.draft_preferred, item.id):
+                    # pylint: disable=bad-super-call
+                    super(SplitMongoModuleStore, self).create_course(
+                        org, course, run, user_id, runtime=item.runtime, **kwargs
+                    )
 
-        return item
+            return item
 
     def get_course(self, course_id, depth=0, **kwargs):
         course_id = self._map_revision_to_branch(course_id)
@@ -87,15 +89,16 @@ class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleS
 
     def update_item(self, descriptor, user_id, allow_not_found=False, force=False, **kwargs):
         descriptor.location = self._map_revision_to_branch(descriptor.location)
-        item = super(DraftVersioningModuleStore, self).update_item(
-            descriptor,
-            user_id,
-            allow_not_found=allow_not_found,
-            force=force,
-            **kwargs
-        )
-        self._auto_publish_no_children(item.location, item.location.category, user_id, **kwargs)
-        return item
+        with self.bulk_write_operations(descriptor.location.course_key):
+            item = super(DraftVersioningModuleStore, self).update_item(
+                descriptor,
+                user_id,
+                allow_not_found=allow_not_found,
+                force=force,
+                **kwargs
+            )
+            self._auto_publish_no_children(item.location, item.location.category, user_id, **kwargs)
+            return item
 
     def create_item(
         self, user_id, course_key, block_type, block_id=None,
@@ -106,26 +109,28 @@ class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleS
         See :py:meth `ModuleStoreDraftAndPublished.create_item`
         """
         course_key = self._map_revision_to_branch(course_key)
-        item = super(DraftVersioningModuleStore, self).create_item(
-            user_id, course_key, block_type, block_id=block_id,
-            definition_locator=definition_locator, fields=fields,
-            force=force, continue_version=continue_version, **kwargs
-        )
-        if not skip_auto_publish:
-            self._auto_publish_no_children(item.location, item.location.category, user_id, **kwargs)
-        return item
+        with self.bulk_write_operations(course_key):
+            item = super(DraftVersioningModuleStore, self).create_item(
+                user_id, course_key, block_type, block_id=block_id,
+                definition_locator=definition_locator, fields=fields,
+                force=force, continue_version=continue_version, **kwargs
+            )
+            if not skip_auto_publish:
+                self._auto_publish_no_children(item.location, item.location.category, user_id, **kwargs)
+            return item
 
     def create_child(
             self, user_id, parent_usage_key, block_type, block_id=None,
             fields=None, continue_version=False, **kwargs
     ):
         parent_usage_key = self._map_revision_to_branch(parent_usage_key)
-        item = super(DraftVersioningModuleStore, self).create_child(
-            user_id, parent_usage_key, block_type, block_id=block_id,
-            fields=fields, continue_version=continue_version, **kwargs
-        )
-        self._auto_publish_no_children(parent_usage_key, item.location.category, user_id, **kwargs)
-        return item
+        with self.bulk_write_operations(parent_usage_key.course_key):
+            item = super(DraftVersioningModuleStore, self).create_child(
+                user_id, parent_usage_key, block_type, block_id=block_id,
+                fields=fields, continue_version=continue_version, **kwargs
+            )
+            self._auto_publish_no_children(parent_usage_key, item.location.category, user_id, **kwargs)
+            return item
 
     def delete_item(self, location, user_id, revision=None, **kwargs):
         """
@@ -141,26 +146,27 @@ class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleS
                     currently only provided by contentstore.views.item.orphan_handler
                 Otherwise, raises a ValueError.
         """
-        if revision == ModuleStoreEnum.RevisionOption.published_only:
-            branches_to_delete = [ModuleStoreEnum.BranchName.published]
-        elif revision == ModuleStoreEnum.RevisionOption.all:
-            branches_to_delete = [ModuleStoreEnum.BranchName.published, ModuleStoreEnum.BranchName.draft]
-        elif revision is None:
-            branches_to_delete = [ModuleStoreEnum.BranchName.draft]
-        else:
-            raise UnsupportedRevisionError(
-                [
-                    None,
-                    ModuleStoreEnum.RevisionOption.published_only,
-                    ModuleStoreEnum.RevisionOption.all
-                ]
-            )
+        with self.bulk_write_operations(location.course_key):
+            if revision == ModuleStoreEnum.RevisionOption.published_only:
+                branches_to_delete = [ModuleStoreEnum.BranchName.published]
+            elif revision == ModuleStoreEnum.RevisionOption.all:
+                branches_to_delete = [ModuleStoreEnum.BranchName.published, ModuleStoreEnum.BranchName.draft]
+            elif revision is None:
+                branches_to_delete = [ModuleStoreEnum.BranchName.draft]
+            else:
+                raise UnsupportedRevisionError(
+                    [
+                        None,
+                        ModuleStoreEnum.RevisionOption.published_only,
+                        ModuleStoreEnum.RevisionOption.all
+                    ]
+                )
 
-        for branch in branches_to_delete:
-            branched_location = location.for_branch(branch)
-            parent_loc = self.get_parent_location(branched_location)
-            SplitMongoModuleStore.delete_item(self, branched_location, user_id)
-            self._auto_publish_no_children(parent_loc, parent_loc.category, user_id, **kwargs)
+            for branch in branches_to_delete:
+                branched_location = location.for_branch(branch)
+                parent_loc = self.get_parent_location(branched_location)
+                SplitMongoModuleStore.delete_item(self, branched_location, user_id)
+                self._auto_publish_no_children(parent_loc, parent_loc.category, user_id, **kwargs)
 
     def _map_revision_to_branch(self, key, revision=None):
         """
@@ -231,7 +237,7 @@ class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleS
         :return: True if the draft and published versions differ
         """
         def get_block(branch_name):
-            course_structure = self._lookup_course(xblock.location.for_branch(branch_name))['structure']
+            course_structure = self._lookup_course(xblock.location.course_key.for_branch(branch_name))['structure']
             return self._get_block_from_structure(course_structure, xblock.location.block_id)
 
         draft_block = get_block(ModuleStoreEnum.BranchName.draft)
@@ -255,7 +261,9 @@ class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleS
             # because for_branch obliterates the version_guid and will lead to missed version conflicts.
             # TODO Instead, the for_branch implementation should be fixed in the Opaque Keys library.
             location.course_key.replace(branch=ModuleStoreEnum.BranchName.draft),
-            location.course_key.for_branch(ModuleStoreEnum.BranchName.published),
+            # We clear out the version_guid here because the location here is from the draft branch, and that
+            # won't have the same version guid
+            location.course_key.replace(branch=ModuleStoreEnum.BranchName.published, version_guid=None),
             [location],
             blacklist=blacklist
         )
@@ -266,8 +274,9 @@ class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleS
         Deletes the published version of the item.
         Returns the newly unpublished item.
         """
-        self.delete_item(location, user_id, revision=ModuleStoreEnum.RevisionOption.published_only)
-        return self.get_item(location.for_branch(ModuleStoreEnum.BranchName.draft), **kwargs)
+        with self.bulk_write_operations(location.course_key):
+            self.delete_item(location, user_id, revision=ModuleStoreEnum.RevisionOption.published_only)
+            return self.get_item(location.for_branch(ModuleStoreEnum.BranchName.draft), **kwargs)
 
     def revert_to_published(self, location, user_id):
         """
@@ -348,32 +357,33 @@ class DraftVersioningModuleStore(ModuleStoreDraftAndPublished, SplitMongoModuleS
         """
         Split-based modulestores need to import published blocks to both branches
         """
-        # hardcode course root block id
-        if block_type == 'course':
-            block_id = self.DEFAULT_ROOT_BLOCK_ID
-        new_usage_key = course_key.make_usage_key(block_type, block_id)
+        with self.bulk_write_operations(course_key):
+            # hardcode course root block id
+            if block_type == 'course':
+                block_id = self.DEFAULT_ROOT_BLOCK_ID
+            new_usage_key = course_key.make_usage_key(block_type, block_id)
 
-        if self.get_branch_setting() == ModuleStoreEnum.Branch.published_only:
-            # if importing a direct only, override existing draft
-            if block_type in DIRECT_ONLY_CATEGORIES:
-                draft_course = course_key.for_branch(ModuleStoreEnum.BranchName.draft)
-                with self.branch_setting(ModuleStoreEnum.Branch.draft_preferred, draft_course):
-                    draft = self.import_xblock(user_id, draft_course, block_type, block_id, fields, runtime)
-                    self._auto_publish_no_children(draft.location, block_type, user_id)
-                return self.get_item(new_usage_key.for_branch(ModuleStoreEnum.BranchName.published))
-            # if new to published
-            elif not self.has_item(new_usage_key.for_branch(ModuleStoreEnum.BranchName.published)):
-                # check whether it's new to draft
-                if not self.has_item(new_usage_key.for_branch(ModuleStoreEnum.BranchName.draft)):
-                    # add to draft too
+            if self.get_branch_setting() == ModuleStoreEnum.Branch.published_only:
+                # if importing a direct only, override existing draft
+                if block_type in DIRECT_ONLY_CATEGORIES:
                     draft_course = course_key.for_branch(ModuleStoreEnum.BranchName.draft)
                     with self.branch_setting(ModuleStoreEnum.Branch.draft_preferred, draft_course):
                         draft = self.import_xblock(user_id, draft_course, block_type, block_id, fields, runtime)
-                        return self.publish(draft.location, user_id, blacklist=EXCLUDE_ALL)
+                        self._auto_publish_no_children(draft.location, block_type, user_id)
+                    return self.get_item(new_usage_key.for_branch(ModuleStoreEnum.BranchName.published))
+                # if new to published
+                elif not self.has_item(new_usage_key.for_branch(ModuleStoreEnum.BranchName.published)):
+                    # check whether it's new to draft
+                    if not self.has_item(new_usage_key.for_branch(ModuleStoreEnum.BranchName.draft)):
+                        # add to draft too
+                        draft_course = course_key.for_branch(ModuleStoreEnum.BranchName.draft)
+                        with self.branch_setting(ModuleStoreEnum.Branch.draft_preferred, draft_course):
+                            draft = self.import_xblock(user_id, draft_course, block_type, block_id, fields, runtime)
+                            return self.publish(draft.location, user_id, blacklist=EXCLUDE_ALL)
 
-        # do the import
-        partitioned_fields = self.partition_fields_by_scope(block_type, fields)
-        course_key = self._map_revision_to_branch(course_key)  # cast to branch_setting
-        return self._update_item_from_fields(
-            user_id, course_key, block_type, block_id, partitioned_fields, None, allow_not_found=True, force=True
-        )
+            # do the import
+            partitioned_fields = self.partition_fields_by_scope(block_type, fields)
+            course_key = self._map_revision_to_branch(course_key)  # cast to branch_setting
+            return self._update_item_from_fields(
+                user_id, course_key, block_type, block_id, partitioned_fields, None, allow_not_found=True, force=True
+            )

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

@@ -298,4 +298,12 @@ def check_mongo_calls(mongo_store, num_finds=0, num_sends=None):
         finally:
             map(lambda wrap_patch: wrap_patch.stop(), wrap_patches)
             call_count = sum([find_wrap.call_count for find_wrap in find_wraps])
-            assert_equal(call_count, num_finds)
+            assert_equal(
+                call_count,
+                num_finds,
+                "Expected {} calls, {} were made. Calls: {}".format(
+                    num_finds,
+                    call_count,
+                    [find_wrap.call_args_list for find_wrap in find_wraps]
+                )
+            )

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

@@ -127,15 +127,16 @@ class TestMixedModuleStore(unittest.TestCase):
         Create a course w/ one item in the persistence store using the given course & item location.
         """
         # create course
-        self.course = self.store.create_course(course_key.org, course_key.course, course_key.run, self.user_id)
-        if isinstance(self.course.id, CourseLocator):
-            self.course_locations[self.MONGO_COURSEID] = self.course.location
-        else:
-            self.assertEqual(self.course.id, course_key)
+        with self.store.bulk_write_operations(course_key):
+            self.course = self.store.create_course(course_key.org, course_key.course, course_key.run, self.user_id)
+            if isinstance(self.course.id, CourseLocator):
+                self.course_locations[self.MONGO_COURSEID] = self.course.location
+            else:
+                self.assertEqual(self.course.id, course_key)
 
-        # create chapter
-        chapter = self.store.create_child(self.user_id, self.course.location, 'chapter', block_id='Overview')
-        self.writable_chapter_location = chapter.location
+            # create chapter
+            chapter = self.store.create_child(self.user_id, self.course.location, 'chapter', block_id='Overview')
+            self.writable_chapter_location = chapter.location
 
     def _create_block_hierarchy(self):
         """
@@ -188,8 +189,9 @@ class TestMixedModuleStore(unittest.TestCase):
                 create_sub_tree(block, tree)
             setattr(self, block_info.field_name, block.location)
 
-        for tree in trees:
-            create_sub_tree(self.course, tree)
+        with self.store.bulk_write_operations(self.course.id):
+            for tree in trees:
+                create_sub_tree(self.course, tree)
 
     def _course_key_from_string(self, string):
         """
@@ -349,10 +351,9 @@ class TestMixedModuleStore(unittest.TestCase):
             )
 
     # draft: 2 to look in draft and then published and then 5 for updating ancestors.
-    # split: 3 to get the course structure & the course definition (show_calculator is scope content)
-    #  before the change. 1 during change to refetch the definition. 3 afterward (b/c it calls get_item to return the "new" object).
+    # split: 1 for the course index, 1 for the course structure before the change, 1 for the structure after the change
     #  2 sends to update index & structure (calculator is a setting field)
-    @ddt.data(('draft', 7, 5), ('split', 6, 2))
+    @ddt.data(('draft', 7, 5), ('split', 3, 2))
     @ddt.unpack
     def test_update_item(self, default_ms, max_find, max_send):
         """
@@ -434,7 +435,7 @@ class TestMixedModuleStore(unittest.TestCase):
         component = self.store.publish(component.location, self.user_id)
         self.assertFalse(self.store.has_changes(component))
 
-    @ddt.data(('draft', 7, 2), ('split', 13, 4))
+    @ddt.data(('draft', 7, 2), ('split', 2, 4))
     @ddt.unpack
     def test_delete_item(self, default_ms, max_find, max_send):
         """
@@ -453,7 +454,7 @@ class TestMixedModuleStore(unittest.TestCase):
         with self.assertRaises(ItemNotFoundError):
             self.store.get_item(self.writable_chapter_location)
 
-    @ddt.data(('draft', 8, 2), ('split', 13, 4))
+    @ddt.data(('draft', 8, 2), ('split', 2, 4))
     @ddt.unpack
     def test_delete_private_vertical(self, default_ms, max_find, max_send):
         """
@@ -499,7 +500,7 @@ class TestMixedModuleStore(unittest.TestCase):
         self.assertFalse(self.store.has_item(leaf_loc))
         self.assertNotIn(vert_loc, course.children)
 
-    @ddt.data(('draft', 4, 1), ('split', 5, 2))
+    @ddt.data(('draft', 4, 1), ('split', 1, 2))
     @ddt.unpack
     def test_delete_draft_vertical(self, default_ms, max_find, max_send):
         """
@@ -579,7 +580,7 @@ class TestMixedModuleStore(unittest.TestCase):
             xml_store.create_course("org", "course", "run", self.user_id)
 
     # draft is 2 to compute inheritance
-    # split is 3 b/c it gets the definition to check whether wiki is set
+    # split is 3 (one for the index, one for the definition to check if the wiki is set, and one for the course structure
     @ddt.data(('draft', 2, 0), ('split', 3, 0))
     @ddt.unpack
     def test_get_course(self, default_ms, max_find, max_send):
@@ -884,7 +885,7 @@ class TestMixedModuleStore(unittest.TestCase):
         mongo_store = self.store._get_modulestore_for_courseid(self._course_key_from_string(self.MONGO_COURSEID))
         with check_mongo_calls(mongo_store, max_find, max_send):
             found_orphans = self.store.get_orphans(self.course_locations[self.MONGO_COURSEID].course_key)
-        self.assertEqual(set(found_orphans), set(orphan_locations))
+        self.assertItemsEqual(found_orphans, orphan_locations)
 
     @ddt.data('draft')
     def test_create_item_from_parent_location(self, default_ms):
@@ -953,7 +954,9 @@ class TestMixedModuleStore(unittest.TestCase):
         self.assertEqual(len(self.store.get_courses_for_wiki('edX.simple.2012_Fall')), 0)
         self.assertEqual(len(self.store.get_courses_for_wiki('no_such_wiki')), 0)
 
-    @ddt.data(('draft', 2, 6), ('split', 7, 2))
+    # Split takes 1 query to read the course structure, deletes all of the entries in memory, and loads the module from an in-memory cache
+    # Only writes the course structure back to the database once
+    @ddt.data(('draft', 2, 6), ('split', 1, 1))
     @ddt.unpack
     def test_unpublish(self, default_ms, max_find, max_send):
         """

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

@@ -2,12 +2,13 @@
     Test split modulestore w/o using any django stuff.
 """
 import datetime
+import random
+import re
 import unittest
 import uuid
 from importlib import import_module
+from mock import MagicMock, Mock, call
 from path import path
-import re
-import random
 
 from xmodule.course_module import CourseDescriptor
 from xmodule.modulestore import ModuleStoreEnum
@@ -20,7 +21,8 @@ from opaque_keys.edx.locator import CourseLocator, BlockUsageLocator, VersionTre
 from xmodule.modulestore.inheritance import InheritanceMixin
 from xmodule.x_module import XModuleMixin
 from xmodule.fields import Date, Timedelta
-from xmodule.modulestore.split_mongo.split import SplitMongoModuleStore
+from xmodule.modulestore.split_mongo.split import SplitMongoModuleStore, BulkWriteMixin
+from xmodule.modulestore.split_mongo.mongo_connection import MongoConnection
 from xmodule.modulestore.tests.test_modulestore import check_has_course_method
 from xmodule.modulestore.tests.mongo_connection import MONGO_PORT_NUM, MONGO_HOST
 
@@ -492,13 +494,14 @@ class SplitModuleTest(unittest.TestCase):
                     new_ele_dict[spec['id']] = child
                 course = split_store.persist_xblock_dag(course, revision['user_id'])
         # publish "testx.wonderful"
+        source_course = CourseLocator(org="testx", course="wonderful", run="run", branch=BRANCH_NAME_DRAFT)
         to_publish = BlockUsageLocator(
-            CourseLocator(org="testx", course="wonderful", run="run", branch=BRANCH_NAME_DRAFT),
+            source_course,
             block_type='course',
             block_id="head23456"
         )
         destination = CourseLocator(org="testx", course="wonderful", run="run", branch=BRANCH_NAME_PUBLISHED)
-        split_store.copy("test@edx.org", to_publish, destination, [to_publish], None)
+        split_store.copy("test@edx.org", source_course, destination, [to_publish], None)
 
     def setUp(self):
         self.user_id = random.getrandbits(32)
@@ -985,7 +988,7 @@ class TestItemCrud(SplitModuleTest):
         # grab link to course to ensure new versioning works
         locator = CourseLocator(org='testx', course='GreekHero', run="run", branch=BRANCH_NAME_DRAFT)
         premod_course = modulestore().get_course(locator)
-        premod_history = modulestore().get_course_history_info(premod_course.location)
+        premod_history = modulestore().get_course_history_info(locator)
         # add minimal one w/o a parent
         category = 'sequential'
         new_module = modulestore().create_item(
@@ -999,7 +1002,7 @@ class TestItemCrud(SplitModuleTest):
         current_course = modulestore().get_course(locator)
         self.assertEqual(new_module.location.version_guid, current_course.location.version_guid)
 
-        history_info = modulestore().get_course_history_info(current_course.location)
+        history_info = modulestore().get_course_history_info(current_course.location.course_key)
         self.assertEqual(history_info['previous_version'], premod_course.location.version_guid)
         self.assertEqual(history_info['original_version'], premod_history['original_version'])
         self.assertEqual(history_info['edited_by'], "user123")
@@ -1112,84 +1115,82 @@ class TestItemCrud(SplitModuleTest):
         chapter = modulestore().get_item(chapter_locator)
         self.assertIn(problem_locator, version_agnostic(chapter.children))
 
-    def test_create_continue_version(self):
+    def test_create_bulk_write_operations(self):
         """
-        Test create_item using the continue_version flag
+        Test create_item using bulk_write_operations
         """
         # start transaction w/ simple creation
         user = random.getrandbits(32)
-        new_course = modulestore().create_course('test_org', 'test_transaction', 'test_run', user, BRANCH_NAME_DRAFT)
-        new_course_locator = new_course.id
-        index_history_info = modulestore().get_course_history_info(new_course.location)
-        course_block_prev_version = new_course.previous_version
-        course_block_update_version = new_course.update_version
-        self.assertIsNotNone(new_course_locator.version_guid, "Want to test a definite version")
-        versionless_course_locator = new_course_locator.version_agnostic()
+        course_key = CourseLocator('test_org', 'test_transaction', 'test_run')
+        with modulestore().bulk_write_operations(course_key):
+            new_course = modulestore().create_course('test_org', 'test_transaction', 'test_run', user, BRANCH_NAME_DRAFT)
+            new_course_locator = new_course.id
+            index_history_info = modulestore().get_course_history_info(new_course.location.course_key)
+            course_block_prev_version = new_course.previous_version
+            course_block_update_version = new_course.update_version
+            self.assertIsNotNone(new_course_locator.version_guid, "Want to test a definite version")
+            versionless_course_locator = new_course_locator.version_agnostic()
 
-        # positive simple case: no force, add chapter
-        new_ele = modulestore().create_child(
-            user, new_course.location, 'chapter',
-            fields={'display_name': 'chapter 1'},
-            continue_version=True
-        )
-        # version info shouldn't change
-        self.assertEqual(new_ele.update_version, course_block_update_version)
-        self.assertEqual(new_ele.update_version, new_ele.location.version_guid)
-        refetch_course = modulestore().get_course(versionless_course_locator)
-        self.assertEqual(refetch_course.location.version_guid, new_course.location.version_guid)
-        self.assertEqual(refetch_course.previous_version, course_block_prev_version)
-        self.assertEqual(refetch_course.update_version, course_block_update_version)
-        refetch_index_history_info = modulestore().get_course_history_info(refetch_course.location)
-        self.assertEqual(refetch_index_history_info, index_history_info)
-        self.assertIn(new_ele.location.version_agnostic(), version_agnostic(refetch_course.children))
-
-        # try to create existing item
-        with self.assertRaises(DuplicateItemError):
-            _fail = modulestore().create_child(
+            # positive simple case: no force, add chapter
+            new_ele = modulestore().create_child(
                 user, new_course.location, 'chapter',
-                block_id=new_ele.location.block_id,
-                fields={'display_name': 'chapter 2'},
-                continue_version=True
+                fields={'display_name': 'chapter 1'},
             )
+            # version info shouldn't change
+            self.assertEqual(new_ele.update_version, course_block_update_version)
+            self.assertEqual(new_ele.update_version, new_ele.location.version_guid)
+            refetch_course = modulestore().get_course(versionless_course_locator)
+            self.assertEqual(refetch_course.location.version_guid, new_course.location.version_guid)
+            self.assertEqual(refetch_course.previous_version, course_block_prev_version)
+            self.assertEqual(refetch_course.update_version, course_block_update_version)
+            refetch_index_history_info = modulestore().get_course_history_info(refetch_course.location.course_key)
+            self.assertEqual(refetch_index_history_info, index_history_info)
+            self.assertIn(new_ele.location.version_agnostic(), version_agnostic(refetch_course.children))
+
+            # try to create existing item
+            with self.assertRaises(DuplicateItemError):
+                _fail = modulestore().create_child(
+                    user, new_course.location, 'chapter',
+                    block_id=new_ele.location.block_id,
+                    fields={'display_name': 'chapter 2'},
+                )
 
         # start a new transaction
-        new_ele = modulestore().create_child(
-            user, new_course.location, 'chapter',
-            fields={'display_name': 'chapter 2'},
-            continue_version=False
-        )
-        transaction_guid = new_ele.location.version_guid
-        # ensure force w/ continue gives exception
-        with self.assertRaises(VersionConflictError):
-            _fail = modulestore().create_child(
+        with modulestore().bulk_write_operations(course_key):
+            new_ele = modulestore().create_child(
                 user, new_course.location, 'chapter',
                 fields={'display_name': 'chapter 2'},
-                force=True, continue_version=True
             )
+            transaction_guid = new_ele.location.version_guid
+            # ensure force w/ continue gives exception
+            with self.assertRaises(VersionConflictError):
+                _fail = modulestore().create_child(
+                    user, new_course.location, 'chapter',
+                    fields={'display_name': 'chapter 2'},
+                    force=True
+                )
 
-        # ensure trying to continue the old one gives exception
-        with self.assertRaises(VersionConflictError):
-            _fail = modulestore().create_child(
-                user, new_course.location, 'chapter',
-                fields={'display_name': 'chapter 3'},
-                continue_version=True
+            # ensure trying to continue the old one gives exception
+            with self.assertRaises(VersionConflictError):
+                _fail = modulestore().create_child(
+                    user, new_course.location, 'chapter',
+                    fields={'display_name': 'chapter 3'},
+                )
+
+            # add new child to old parent in continued (leave off version_guid)
+            course_module_locator = new_course.location.version_agnostic()
+            new_ele = modulestore().create_child(
+                user, course_module_locator, 'chapter',
+                fields={'display_name': 'chapter 4'},
             )
+            self.assertNotEqual(new_ele.update_version, course_block_update_version)
+            self.assertEqual(new_ele.location.version_guid, transaction_guid)
 
-        # add new child to old parent in continued (leave off version_guid)
-        course_module_locator = new_course.location.version_agnostic()
-        new_ele = modulestore().create_child(
-            user, course_module_locator, 'chapter',
-            fields={'display_name': 'chapter 4'},
-            continue_version=True
-        )
-        self.assertNotEqual(new_ele.update_version, course_block_update_version)
-        self.assertEqual(new_ele.location.version_guid, transaction_guid)
-
-        # check children, previous_version
-        refetch_course = modulestore().get_course(versionless_course_locator)
-        self.assertIn(new_ele.location.version_agnostic(), version_agnostic(refetch_course.children))
-        self.assertEqual(refetch_course.previous_version, course_block_update_version)
-        self.assertEqual(refetch_course.update_version, transaction_guid)
+            # check children, previous_version
+            refetch_course = modulestore().get_course(versionless_course_locator)
+            self.assertIn(new_ele.location.version_agnostic(), version_agnostic(refetch_course.children))
+            self.assertEqual(refetch_course.previous_version, course_block_update_version)
+            self.assertEqual(refetch_course.update_version, transaction_guid)
 
     def test_update_metadata(self):
         """
@@ -1221,7 +1222,7 @@ class TestItemCrud(SplitModuleTest):
         current_course = modulestore().get_course(locator.course_key)
         self.assertEqual(updated_problem.location.version_guid, current_course.location.version_guid)
 
-        history_info = modulestore().get_course_history_info(current_course.location)
+        history_info = modulestore().get_course_history_info(current_course.location.course_key)
         self.assertEqual(history_info['previous_version'], pre_version_guid)
         self.assertEqual(history_info['edited_by'], self.user_id)
 
@@ -1396,16 +1397,13 @@ class TestCourseCreation(SplitModuleTest):
         )
         new_locator = new_course.location
         # check index entry
-        index_info = modulestore().get_course_index_info(new_locator)
+        index_info = modulestore().get_course_index_info(new_locator.course_key)
         self.assertEqual(index_info['org'], 'test_org')
         self.assertEqual(index_info['edited_by'], 'create_user')
         # check structure info
-        structure_info = modulestore().get_course_history_info(new_locator)
-        # TODO LMS-11098 "Implement bulk_write in Split"
-        # Right now, these assertions will not pass because create_course calls update_item,
-        #   resulting in two versions. Bulk updater will fix this.
-        # self.assertEqual(structure_info['original_version'], index_info['versions'][BRANCH_NAME_DRAFT])
-        # self.assertIsNone(structure_info['previous_version'])
+        structure_info = modulestore().get_course_history_info(new_locator.course_key)
+        self.assertEqual(structure_info['original_version'], index_info['versions'][BRANCH_NAME_DRAFT])
+        self.assertIsNone(structure_info['previous_version'])
 
         self.assertEqual(structure_info['edited_by'], 'create_user')
         # check the returned course object
@@ -1433,7 +1431,7 @@ class TestCourseCreation(SplitModuleTest):
         self.assertEqual(new_draft.edited_by, 'test@edx.org')
         self.assertEqual(new_draft_locator.version_guid, original_index['versions'][BRANCH_NAME_DRAFT])
         # however the edited_by and other meta fields on course_index will be this one
-        new_index = modulestore().get_course_index_info(new_draft_locator)
+        new_index = modulestore().get_course_index_info(new_draft_locator.course_key)
         self.assertEqual(new_index['edited_by'], 'leech_master')
 
         new_published_locator = new_draft_locator.course_key.for_branch(BRANCH_NAME_PUBLISHED)
@@ -1483,7 +1481,7 @@ class TestCourseCreation(SplitModuleTest):
         self.assertEqual(new_draft.edited_by, 'leech_master')
         self.assertNotEqual(new_draft_locator.version_guid, original_index['versions'][BRANCH_NAME_DRAFT])
         # however the edited_by and other meta fields on course_index will be this one
-        new_index = modulestore().get_course_index_info(new_draft_locator)
+        new_index = modulestore().get_course_index_info(new_draft_locator.course_key)
         self.assertEqual(new_index['edited_by'], 'leech_master')
         self.assertEqual(new_draft.display_name, fields['display_name'])
         self.assertDictEqual(
@@ -1504,13 +1502,13 @@ class TestCourseCreation(SplitModuleTest):
         head_course = modulestore().get_course(locator)
         versions = course_info['versions']
         versions[BRANCH_NAME_DRAFT] = head_course.previous_version
-        modulestore().update_course_index(course_info)
+        modulestore().update_course_index(None, course_info)
         course = modulestore().get_course(locator)
         self.assertEqual(course.location.version_guid, versions[BRANCH_NAME_DRAFT])
 
         # an allowed but not recommended way to publish a course
         versions[BRANCH_NAME_PUBLISHED] = versions[BRANCH_NAME_DRAFT]
-        modulestore().update_course_index(course_info)
+        modulestore().update_course_index(None, course_info)
         course = modulestore().get_course(locator.for_branch(BRANCH_NAME_PUBLISHED))
         self.assertEqual(course.location.version_guid, versions[BRANCH_NAME_DRAFT])
 
@@ -1715,6 +1713,7 @@ class TestPublish(SplitModuleTest):
                 dest_cursor += 1
         self.assertEqual(dest_cursor, len(dest_children))
 
+
 class TestSchema(SplitModuleTest):
     """
     Test the db schema (and possibly eventually migrations?)
@@ -1736,6 +1735,102 @@ class TestSchema(SplitModuleTest):
                 "{0.name} has records with wrong schema_version".format(collection)
             )
 
+
+class TestBulkWriteMixin(unittest.TestCase):
+    def setUp(self):
+        self.bulk = BulkWriteMixin()
+        self.clear_cache = self.bulk._clear_cache = Mock(name='_clear_cache')
+        self.conn = self.bulk.db_connection = MagicMock(name='db_connection', spec=MongoConnection)
+
+        self.course_key = Mock(name='course_key', spec=CourseLocator)
+        self.course_key_b = Mock(name='course_key_b', spec=CourseLocator)
+        self.version_guid = Mock(name='version_guid')
+        self.structure = MagicMock(name='structure')
+        self.index_entry = MagicMock(name='index_entry')
+
+    def assertConnCalls(self, *calls):
+        self.assertEqual(list(calls), self.conn.mock_calls)
+
+    def assertCacheNotCleared(self):
+        self.assertFalse(self.clear_cache.called)
+
+    def test_no_bulk_read_structure(self):
+        # Reading a structure when no bulk operation is active should just call
+        # through to the db_connection
+        result = self.bulk.get_structure(self.course_key, self.version_guid)
+        self.assertConnCalls(call.get_structure(self.course_key.as_object_id.return_value))
+        self.assertEqual(result, self.conn.get_structure.return_value)
+        self.assertCacheNotCleared()
+
+    def test_no_bulk_write_structure(self):
+        # Writing a structure when no bulk operation is active should just
+        # call through to the db_connection. It should also clear the
+        # system cache
+        self.bulk.update_structure(self.course_key, self.structure)
+        self.assertConnCalls(call.upsert_structure(self.structure))
+        self.clear_cache.assert_called_once_with(self.structure['_id'])
+
+    def test_no_bulk_read_index(self):
+        # Reading a course index when no bulk operation is active should just call
+        # through to the db_connection
+        result = self.bulk.get_course_index(self.course_key, ignore_case=True)
+        self.assertConnCalls(call.get_course_index(self.course_key, True))
+        self.assertEqual(result, self.conn.get_course_index.return_value)
+        self.assertCacheNotCleared()
+
+    def test_no_bulk_write_index(self):
+        # Writing a course index when no bulk operation is active should just call
+        # through to the db_connection
+        self.bulk.insert_course_index(self.course_key, self.index_entry)
+        self.assertConnCalls(call.insert_course_index(self.index_entry))
+        self.assertCacheNotCleared()
+
+    def test_read_structure_without_write_from_db(self):
+        # Reading a structure before it's been written (while in bulk operation mode)
+        # returns the structure from the database
+        self.bulk._begin_bulk_write_operation(self.course_key)
+        result = self.bulk.get_structure(self.course_key, self.version_guid)
+        self.assertEquals(self.conn.get_structure.call_count, 1)
+        self.assertEqual(result, self.conn.get_structure.return_value)
+        self.assertCacheNotCleared()
+
+    def test_read_structure_without_write_only_reads_once(self):
+        # Reading the same structure multiple times shouldn't hit the database
+        # more than once
+        self.test_read_structure_without_write_from_db()
+        result = self.bulk.get_structure(self.course_key, self.version_guid)
+        self.assertEquals(self.conn.get_structure.call_count, 1)
+        self.assertEqual(result, self.conn.get_structure.return_value)
+        self.assertCacheNotCleared()
+
+    def test_read_structure_after_write_no_db(self):
+        # Reading a structure that's already been written shouldn't hit the db at all
+        self.bulk._begin_bulk_write_operation(self.course_key)
+        self.bulk.update_structure(self.course_key, self.structure)
+        result = self.bulk.get_structure(self.course_key, self.version_guid)
+        self.assertEquals(self.conn.get_structure.call_count, 0)
+        self.assertEqual(result, self.structure)
+
+    def test_read_structure_after_write_after_read(self):
+        # Reading a structure that's been updated after being pulled from the db should
+        # still get the updated value
+        self.test_read_structure_without_write_only_reads_once()
+        self.conn.get_structure.reset_mock()
+        self.bulk.update_structure(self.course_key, self.structure)
+        result = self.bulk.get_structure(self.course_key, self.version_guid)
+        self.assertEquals(self.conn.get_structure.call_count, 0)
+        self.assertEqual(result, self.structure)
+
+    # read index after close
+    # read structure after close
+    # write index on close
+    # write structure on close
+    # close with new index
+    # close with existing index
+    # read index after update
+    # read index without update
+
+
 #===========================================
 def modulestore():
     """

common/lib/xmodule/xmodule/tests/__init__.py

@@ -287,9 +287,9 @@ class CourseComparisonTest(unittest.TestCase):
             self.assertEqual(expected_item.has_children, actual_item.has_children)
             if expected_item.has_children:
                 expected_children = [
-                    (course1_item_child.location.block_type, course1_item_child.location.block_id)
+                    (expected_item_child.location.block_type, expected_item_child.location.block_id)
                     # get_children() rather than children to strip privates from public parents
-                    for course1_item_child in expected_item.get_children()
+                    for expected_item_child in expected_item.get_children()
                 ]
                 actual_children = [
                     (item_child.location.block_type, item_child.location.block_id)

pavelib/tests.py

@@ -91,7 +91,10 @@ def test_lib(options):
     }
 
     if test_id:
-        lib = '/'.join(test_id.split('/')[0:3])
+        if '/' in test_id:
+            lib = '/'.join(test_id.split('/')[0:3])
+        else:
+            lib = 'common/lib/' + test_id.split('.')[0]
         opts['test_id'] = test_id
         lib_tests = [suites.LibTestSuite(lib, **opts)]
     else: