def revert_to_published(self, location, user_id):
        """
        Reverts an item to its last published version (recursively traversing all of its descendants).
        If no published version exists, a VersionConflictError is thrown.

        If a published version exists but there is no draft version of this item or any of its descendants, this
        method is a no-op.

        :raises InvalidVersionError: if no published version exists for the location specified
        """
        if location.category in DIRECT_ONLY_CATEGORIES:
            return

        draft_course_key = location.course_key.for_branch(ModuleStoreEnum.BranchName.draft)
        with self.bulk_operations(draft_course_key):

            # get head version of Published branch
            published_course_structure = self._lookup_course(
                location.course_key.for_branch(ModuleStoreEnum.BranchName.published)
            ).structure
            published_block = self._get_block_from_structure(
                published_course_structure,
                BlockKey.from_usage_key(location)
            )
            if published_block is None:
                raise InvalidVersionError(location)

            # create a new versioned draft structure
            draft_course_structure = self._lookup_course(draft_course_key).structure
            new_structure = self.version_structure(draft_course_key, draft_course_structure, user_id)

            # remove the block and its descendants from the new structure
            self._remove_subtree(BlockKey.from_usage_key(location), new_structure['blocks'])

            # copy over the block and its descendants from the published branch
            def copy_from_published(root_block_id):
                """
                copies root_block_id and its descendants from published_course_structure to new_structure
                """
                self._update_block_in_structure(
                    new_structure,
                    root_block_id,
                    self._get_block_from_structure(published_course_structure, root_block_id)
                )
                block = self._get_block_from_structure(new_structure, root_block_id)
                original_parent_location = location.course_key.make_usage_key(root_block_id.type, root_block_id.id)
                for child_block_id in block.fields.get('children', []):
                    item_location = location.course_key.make_usage_key(child_block_id.type, child_block_id.id)
                    self.update_parent_if_moved(item_location, original_parent_location, new_structure, user_id)
                    copy_from_published(child_block_id)

            copy_from_published(BlockKey.from_usage_key(location))

            # update course structure and index
            self.update_structure(draft_course_key, new_structure)
            index_entry = self._get_index_if_valid(draft_course_key)
            if index_entry is not None:
                self._update_head(draft_course_key, index_entry, ModuleStoreEnum.BranchName.draft, new_structure['_id'])

    def _load_item(self, usage_key, course_entry_override=None, **kwargs):
        # usage_key is either a UsageKey or just the block_key. if a usage_key,
        if isinstance(usage_key, BlockUsageLocator):

            # trust the passed in key to know the caller's expectations of which fields are filled in.
            # particularly useful for strip_keys so may go away when we're version aware
            course_key = usage_key.course_key

            if isinstance(usage_key.block_id, LocalId):
                try:
                    return self.local_modules[usage_key]
                except KeyError:
                    raise ItemNotFoundError
            else:
                block_key = BlockKey.from_usage_key(usage_key)
        else:
            block_key = usage_key

            course_info = course_entry_override or self.course_entry
            course_key = CourseLocator(
                version_guid=course_info['structure']['_id'],
                org=course_info.get('org'),
                course=course_info.get('course'),
                run=course_info.get('run'),
                branch=course_info.get('branch'),
            )

        json_data = self.get_module_data(block_key, course_key)

        class_ = self.load_block_type(json_data.get('block_type'))
        # pass None for inherited_settings to signal that it should get the settings from cache
        new_item = self.xblock_from_json(class_, course_key, block_key, json_data, None, course_entry_override, **kwargs)
        return new_item

    def has_changes(self, xblock):
        """
        Checks if the given block has unpublished changes
        :param xblock: the block to check
        :return: True if the draft and published versions differ
        """
        def get_course(branch_name):
            return self._lookup_course(xblock.location.course_key.for_branch(branch_name)).structure

        def get_block(course_structure, block_key):
            return self._get_block_from_structure(course_structure, block_key)

        draft_course = get_course(ModuleStoreEnum.BranchName.draft)
        published_course = get_course(ModuleStoreEnum.BranchName.published)

        def has_changes_subtree(block_key):
            draft_block = get_block(draft_course, block_key)
            published_block = get_block(published_course, block_key)
            if not published_block:
                return True

            # check if the draft has changed since the published was created
            if self._get_version(draft_block) != self._get_version(published_block):
                return True

            # check the children in the draft
            if 'children' in draft_block.setdefault('fields', {}):
                return any(
                    [has_changes_subtree(child_block_id) for child_block_id in draft_block['fields']['children']]
                )

            return False

        return has_changes_subtree(BlockKey.from_usage_key(xblock.location))

    def _load_item(self, usage_key, course_entry_override=None, **kwargs):
        # usage_key is either a UsageKey or just the block_key. if a usage_key,
        if isinstance(usage_key, BlockUsageLocator):

            # trust the passed in key to know the caller's expectations of which fields are filled in.
            # particularly useful for strip_keys so may go away when we're version aware
            course_key = usage_key.course_key

            if isinstance(usage_key.block_id, LocalId):
                try:
                    return self.local_modules[usage_key]
                except KeyError:
                    raise ItemNotFoundError
            else:
                block_key = BlockKey.from_usage_key(usage_key)
        else:
            block_key = usage_key

            course_info = course_entry_override or self.course_entry
            course_key = course_info.course_key

        if course_entry_override:
            structure_id = course_entry_override.structure.get('_id')
        else:
            structure_id = self.course_entry.structure.get('_id')

        json_data = self.get_module_data(block_key, course_key)

        class_ = self.load_block_type(json_data.get('block_type'))
        return self.xblock_from_json(class_, course_key, block_key, json_data, course_entry_override, **kwargs)

 def _get_head(self, xblock, branch):
     """ Gets block at the head of specified branch """
     try:
         course_structure = self._lookup_course(xblock.location.course_key.for_branch(branch)).structure
     except ItemNotFoundError:
         # There is no published version xblock container, e.g. Library
         return None
     return self._get_block_from_structure(course_structure, BlockKey.from_usage_key(xblock.location))

    def get_subtree_edited_on(self, xblock):
        """
        See :class: cms.lib.xblock.runtime.EditInfoRuntimeMixin
        """
        if not hasattr(xblock, "_subtree_edited_on"):
            json_data = self.module_data[BlockKey.from_usage_key(xblock.location)]
            if "_subtree_edited_on" not in json_data.setdefault("edit_info", {}):
                self._compute_subtree_edited_internal(xblock.location.block_id, json_data, xblock.location.course_key)
            setattr(xblock, "_subtree_edited_on", json_data["edit_info"]["_subtree_edited_on"])

        return getattr(xblock, "_subtree_edited_on")

    def get_subtree_edited_by(self, xblock):
        """
        See :class: cms.lib.xblock.runtime.EditInfoRuntimeMixin
        """
        if not hasattr(xblock, '_subtree_edited_by'):
            json_data = self.module_data[BlockKey.from_usage_key(xblock.location)]
            if '_subtree_edited_by' not in json_data.setdefault('edit_info', {}):
                self._compute_subtree_edited_internal(
                    xblock.location.block_id, json_data, xblock.location.course_key
                )
            setattr(xblock, '_subtree_edited_by', json_data['edit_info']['_subtree_edited_by'])

        return getattr(xblock, '_subtree_edited_by')

    def get_definition_id(self, usage_id):
        if isinstance(usage_id.block_id, LocalId):
            # a LocalId indicates that this block hasn't been persisted yet, and is instead stored
            # in-memory in the local_modules dictionary.
            return self._cds.local_modules[usage_id].scope_ids.def_id
        else:
            block_key = BlockKey.from_usage_key(usage_id)
            module_data = self._cds.get_module_data(block_key, usage_id.course_key)

            if module_data.definition is not None:
                return DefinitionLocator(usage_id.block_type, module_data.definition)
            else:
                raise ValueError("All non-local blocks should have a definition specified")

    def get_subtree_edited_on(self, xblock):
        """
        See :class: cms.lib.xblock.runtime.EditInfoRuntimeMixin
        """
        # pylint: disable=protected-access
        if not hasattr(xblock, '_subtree_edited_on'):
            block_data = self.module_data[BlockKey.from_usage_key(xblock.location)]
            if block_data.edit_info._subtree_edited_on is None:
                self._compute_subtree_edited_internal(
                    block_data, xblock.location.course_key
                )
            xblock._subtree_edited_on = block_data.edit_info._subtree_edited_on

        return xblock._subtree_edited_on

    def update_parent_if_moved(self, item_location, original_parent_location, course_structure, user_id):
        """
        Update parent of an item if it has moved.

        Arguments:
            item_location (BlockUsageLocator)    : Locator of item.
            original_parent_location (BlockUsageLocator)  : Original parent block locator.
            course_structure (dict)  : course structure of the course.
            user_id (int)   : User id
        """
        parent_block_keys = self._get_parents_from_structure(BlockKey.from_usage_key(item_location), course_structure)
        for block_key in parent_block_keys:
            # Item's parent is different than its new parent - so it has moved.
            if block_key.id != original_parent_location.block_id:
                old_parent_location = original_parent_location.course_key.make_usage_key(block_key.type, block_key.id)
                self.update_item_parent(item_location, original_parent_location, old_parent_location, user_id)

    def _load_item(self, usage_key, course_entry_override=None, **kwargs):
        """
        Instantiate the xblock fetching it either from the cache or from the structure

        :param course_entry_override: the course_info with the course_key to use (defaults to cached)
        """
        # usage_key is either a UsageKey or just the block_key. if a usage_key,
        if isinstance(usage_key, BlockUsageLocator):

            # trust the passed in key to know the caller's expectations of which fields are filled in.
            # particularly useful for strip_keys so may go away when we're version aware
            course_key = usage_key.course_key

            if isinstance(usage_key.block_id, LocalId):
                try:
                    return self.local_modules[usage_key]
                except KeyError:
                    raise ItemNotFoundError
            else:
                block_key = BlockKey.from_usage_key(usage_key)
                version_guid = self.course_entry.course_key.version_guid
        else:
            block_key = usage_key

            course_info = course_entry_override or self.course_entry
            course_key = course_info.course_key
            version_guid = course_key.version_guid

        # look in cache
        cached_module = self.modulestore.get_cached_block(course_key, version_guid, block_key)
        if cached_module:
            return cached_module

        block_data = self.get_module_data(block_key, course_key)

        class_ = self.load_block_type(block_data.block_type)
        block = self.xblock_from_json(class_, course_key, block_key, block_data, course_entry_override, **kwargs)

        # TODO Once TNL-5092 is implemented, we can expose the course version
        # information within the key identifier of the block.  Until then, set
        # the course_version as a field on the returned block so higher layers
        # can use it when needed.
        block.course_version = version_guid

        self.modulestore.cache_block(course_key, version_guid, block_key, block)
        return block

    def _load_item(self, usage_key, course_entry_override=None, **kwargs):
        """
        Instantiate the xblock fetching it either from the cache or from the structure

        :param course_entry_override: the course_info with the course_key to use (defaults to cached)
        """
        # usage_key is either a UsageKey or just the block_key. if a usage_key,
        if isinstance(usage_key, BlockUsageLocator):

            # trust the passed in key to know the caller's expectations of which fields are filled in.
            # particularly useful for strip_keys so may go away when we're version aware
            course_key = usage_key.course_key

            if isinstance(usage_key.block_id, LocalId):
                try:
                    return self.local_modules[usage_key]
                except KeyError:
                    raise ItemNotFoundError
            else:
                block_key = BlockKey.from_usage_key(usage_key)
                version_guid = self.course_entry.course_key.version_guid
        else:
            block_key = usage_key

            course_info = course_entry_override or self.course_entry
            course_key = course_info.course_key
            version_guid = course_key.version_guid

        # look in cache
        cached_module = self.modulestore.get_cached_block(course_key, version_guid, block_key)
        if cached_module:
            return cached_module

        json_data = self.get_module_data(block_key, course_key)

        class_ = self.load_block_type(json_data.get('block_type'))
        block = self.xblock_from_json(class_, course_key, block_key, json_data, course_entry_override, **kwargs)
        self.modulestore.cache_block(course_key, version_guid, block_key, block)
        return block

 def _get_head(self, xblock, branch):
     course_structure = self._lookup_course(xblock.location.course_key.for_branch(branch)).structure
     return self._get_block_from_structure(course_structure, BlockKey.from_usage_key(xblock.location))