def __unit_ids_to_plugin_unit_keys(self, unit_ids_by_type, repo_ids):
        """
        Parse a dictionary of unit ids keyed by content type id and return a dictionary of 
        corresponding plugin unit keys keyed by content type id.

        :param unit_ids_by_type: dictionary of <content type id> : <list of unit ids>
        :type unit_ids_by_type: dict

        :return: if units are specified, return the corresponding plugin unit_keys. If unit_ids_by_type dict
                 is empty, return plugin unit keys corresponging to all units in given repo ids.
                 If unit ids list for a particular unit type is empty, return all plugin unit_keys
                 in given repo ids with that unit type.
        :rtype: dict
        """
        repo_unit_association_query_manager = managers.repo_unit_association_query_manager()
        content_query_manager = managers.content_query_manager()

        result_unit_keys = {}

        if unit_ids_by_type is not None:
            for unit_type_id, unit_ids in unit_ids_by_type.items():
                # Get unit type specific collection
                collection = content_query_manager.get_content_unit_collection(type_id=unit_type_id)
                type_def = content_types_db.type_definition(unit_type_id)
                if not unit_ids:
                    # If unit_list is empty for a unit_type, consider all units of specific type
                    criteria = UnitAssociationCriteria(unit_fields = ['unit_id'])
                    for repo_id in repo_ids:
                        repo_units = repo_unit_association_query_manager.get_units_by_type(repo_id, unit_type_id, criteria)
                        # Get metadata for each unit from type specific collection
                        pulp_units = [collection.find_one({'_id': u['unit_id']}) for u in repo_units]
                        # Convert pulp units to plugin unit keys
                        plugin_unit_keys = [common_utils.to_plugin_unit(u, type_def).unit_key for u in pulp_units]
                        result_unit_keys.setdefault(unit_type_id, []).extend(plugin_unit_keys)
                else:
                    # Get metadata for each unit from type specific collection
                    pulp_units = [collection.find_one({'_id': unit_id}) for unit_id in unit_ids]
                    # Convert pulp units to plugin unit keys
                    plugin_unit_keys = [common_utils.to_plugin_unit(u, type_def).unit_key for u in pulp_units]
                    result_unit_keys.setdefault(unit_type_id, []).extend(plugin_unit_keys)
        else:
            # If units are not specified, consider all units in given repos.
            for repo_id in repo_ids:
                all_unit_type_ids = content_types_db.all_type_ids()
                for unit_type_id in all_unit_type_ids:
                    criteria = UnitAssociationCriteria(type_ids=[unit_type_id], unit_fields = ['unit_id', 'unit_type_id'])
                    repo_units = repo_unit_association_query_manager.get_units(repo_id, criteria)

                    # Get unit metadata for each unit from type specific collection
                    collection = content_query_manager.get_content_unit_collection(type_id=unit_type_id)
                    pulp_units = [collection.find_one({'_id': u['unit_id']}) for u in repo_units]
  
                    # Convert pulp units to plugin unit keys
                    type_def = content_types_db.type_definition(unit_type_id)
                    plugin_unit_keys = [common_utils.to_plugin_unit(u, type_def).unit_key for u in pulp_units]
                    result_unit_keys.setdefault(unit_type_id, []).extend(plugin_unit_keys)

        return result_unit_keys

    def __parse_units(self, user_units, repo_ids):
        """
        Parse units specified by user and return a dictionary of all plugin unit_keys 
        to be considered for applicability keyed by unit_type_id.

        :param user_units: dictionary of unit metadata filters keyed by unit-type-id specified by user
        :type user_units: dict

        :return: if specific units are specified, return the corresponding plugin unit_keys. If units dict is empty, 
                 return all plugin unit_keys corresponging to units in given repo ids keyed by unit_type_id. 
                 If units list for a particular unit type in units is empty, return all plugin unit_keys 
                 in given repo ids with that unit type keyed by unit_type_id.
        :rtype: dict
        """
        repo_unit_association_query_manager = managers.repo_unit_association_query_manager()
        content_query_manager = managers.content_query_manager()

        result_unit_keys = {}
        if user_units is not None:
            for unit_type_id, unit_list in user_units.items():
                # Get unit type specific collection
                collection = content_query_manager.get_content_unit_collection(type_id=unit_type_id)
                type_def = content_types_db.type_definition(unit_type_id)
                if not unit_list:
                    # If unit_list is empty for a unit_type, consider all units of specific type
                    criteria = UnitAssociationCriteria(unit_fields = ['unit_id'])
                    for repo_id in repo_ids:
                        repo_units = repo_unit_association_query_manager.get_units_by_type(repo_id, unit_type_id, criteria)
                        # Get unit metadata for each unit from type specific collection
                        pulp_units = [collection.find_one({'_id': u['unit_id']}) for u in repo_units]
                        plugin_units = [common_utils.to_plugin_unit(u, type_def) for u in pulp_units]
                        plugin_unit_keys = [u.unit_key for u in plugin_units]
                        result_unit_keys.setdefault(unit_type_id, []).extend(plugin_unit_keys)
                else:
                    for unit in unit_list:
                        criteria = UnitAssociationCriteria(unit_filters=unit, unit_fields = ['unit_id'])
                        for repo_id in repo_ids:
                            repo_units = repo_unit_association_query_manager.get_units_by_type(repo_id, unit_type_id, criteria)
                            # Get unit metadata for each unit from type specific collection
                            pulp_units = [collection.find_one({'_id': u['unit_id']}) for u in repo_units]
                            plugin_units = [common_utils.to_plugin_unit(u, type_def) for u in pulp_units]
                            plugin_unit_keys = [u.unit_key for u in plugin_units]
                            result_unit_keys.setdefault(unit_type_id, []).extend(plugin_unit_keys)
        else:
            # If units are not specified, consider all units in repo_ids list.
            for repo_id in repo_ids:
                criteria = UnitAssociationCriteria(unit_fields = ['unit_id','unit_type_id'])
                repo_units = repo_unit_association_query_manager.get_units(repo_id, criteria)
                # Get unit metadata for each unit from type specific collection
                for repo_unit in repo_units:
                    collection = content_query_manager.get_content_unit_collection(type_id=repo_unit['unit_type_id'])
                    type_def = content_types_db.type_definition(repo_unit['unit_type_id'])
                    pulp_unit = collection.find_one({'_id': repo_unit['unit_id']})
                    plugin_unit = common_utils.to_plugin_unit(pulp_unit, type_def)
                    result_unit_keys.setdefault(repo_unit['unit_type_id'], []).append(plugin_unit.unit_key)

        return result_unit_keys

    def search_all_units(self, type_id, criteria):
        """
        Searches for units of a given type in the server, regardless of their
        associations to any repositories.

        @param type_id: indicates the type of units being retrieved
        @type  type_id: str
        @param criteria: used to query which units are returned
        @type  criteria: pulp.server.db.model.criteria.Criteria

        @return: list of unit instances
        @rtype:  list of L{Unit}
        """

        try:
            query_manager = manager_factory.content_query_manager()
            units = query_manager.find_by_criteria(type_id, criteria)
            type_def = types_db.type_definition(type_id)

            transfer_units = []
            for pulp_unit in units:
                u = common_utils.to_plugin_unit(pulp_unit, type_def)
                transfer_units.append(u)

            return transfer_units

        except Exception, e:
            _LOG.exception('Exception from server requesting all units of type [%s]' % type_id)
            raise self.exception_class(e), None, sys.exc_info()[2]

def do_get_repo_units(repo_id, criteria, exception_class):
    """
    Performs a repo unit association query. This is split apart so we can have
    custom mixins with different signatures.
    """
    try:
        association_query_manager = manager_factory.repo_unit_association_query_manager()
        units = association_query_manager.get_units(repo_id, criteria=criteria)

        all_units = []

        # Load all type definitions in use so we don't hammer the database
        unique_type_defs = set([u['unit_type_id'] for u in units])
        type_defs = {}
        for def_id in unique_type_defs:
            type_def = types_db.type_definition(def_id)
            type_defs[def_id] = type_def

        # Convert to transfer object
        for unit in units:
            type_id = unit['unit_type_id']
            u = common_utils.to_plugin_unit(unit, type_defs[type_id])
            all_units.append(u)

        return all_units

    except Exception, e:
        _LOG.exception('Exception from server requesting all content units for repository [%s]' % repo_id)
        raise exception_class(e), None, sys.exc_info()[2]

 def link_referenced_content_units(self, from_type, from_id, to_type, to_ids):
     """
     Link referenced content units.
     @param from_type: unique id of the parent content collection
     @type from_type: str
     @param from_id: unique id of the parent content unit
     @type from_id: str
     @param to_type: unique id of the child content collection
     @type to_type: str
     @param to_ids: list of unique ids of child content units
     @types child_ids: tuple of list
     """
     collection = content_types_db.type_units_collection(from_type)
     parent = collection.find_one({'_id': from_id})
     if parent is None:
         raise InvalidValue(['from_type'])
     parent_type_def = content_types_db.type_definition(from_type)
     if to_type not in parent_type_def['referenced_types']:
         raise Exception()
     children = parent.setdefault('_%s_references' % to_type, [])
     for id_ in to_ids:
         if id_ in children:
             continue
         children.append(id_)
     collection.update({'_id': from_id}, parent, safe=True)

def _check_content_definitions(descriptors):
    """
    Check whether the given content definitions exist in the database. This method
    does not make any changes to the content definitions or any indexes.

    :param descriptors: A list of content descriptors
    :type  descriptors: list of TypeDescriptor

    :return: A list of content types that would have been created or updated by
             _load_type_definitions
    :rtype:  list of TypeDefinition
    """
    definitions = parser.parse(descriptors)
    old_content_types = []

    # Ensure all the content types exist and match the definitions
    for definition in definitions:
        content_type = database.type_definition(definition.id)
        if content_type is None:
            old_content_types.append(definition)
            continue

        dict_definition = definition.__dict__
        for key, value in dict_definition.items():
            if key not in content_type or content_type[key] != value:
                old_content_types.append(definition)
                break

    return old_content_types

def get_unit_key_fields_for_type(type_id):
    """
    Based on a unit type ID, determine the fields that compose that type's unit key.

    This supports both the new mongoengine models and the old "types_def" collection, so the caller
    need not worry about whether a type has been converted to use mongoengine or not.

    :param type_id: unique ID for a unit type
    :type  type_id: str

    :return:    tuple containing the name of each field in the unit key
    :rtype:     tuple

    :raises ValueError: if the type ID is not found
    """
    model_class = plugin_api.get_unit_model_by_id(type_id)
    if model_class is not None:
        return model_class.unit_key_fields

    type_def = types_db.type_definition(type_id)
    if type_def is not None:
        # this is an "old style" model
        return tuple(type_def['unit_key'])

    raise ValueError

    def resolve_dependencies_by_units(repo_id, units, options):
        """
        Calculates dependencies for the given set of units in the given
        repository.

        :param repo_id:         identifies the repository
        :type  repo_id:         str
        :param units:           list of database representations of units to resolve dependencies
                                for
        :type  units:           list
        :param options:         dict of options to pass the importer to drive the resolution
        :type  options:         dict or None
        :return:                report from the plugin
        :rtype:                 object
        :raise MissingResource: if the repo does not exist or does not have an importer
        """
        # Validation
        repo_query_manager = manager_factory.repo_query_manager()
        importer_manager = manager_factory.repo_importer_manager()

        # The following will raise MissingResource as appropriate
        repo = repo_query_manager.get_repository(repo_id)
        repo_importer = importer_manager.get_importer(repo_id)

        try:
            importer_instance, plugin_config = plugin_api.get_importer_by_id(
                repo_importer['importer_type_id'])
        except plugin_exceptions.PluginNotFound:
            raise MissingResource(repo_id), None, sys.exc_info()[2]

        # Package for the importer call
        call_config = PluginCallConfiguration(plugin_config, repo_importer['config'], options)
        transfer_repo = common_utils.to_transfer_repo(repo)

        conduit = DependencyResolutionConduit(repo_id, repo_importer['id'])

        # Convert all of the units into the plugin standard representation
        transfer_units = []

        # Preload all the type defs so we don't hammer the database unnecessarily
        type_defs = {}
        all_type_def_ids = set([u['unit_type_id'] for u in units])
        for def_id in all_type_def_ids:
            type_def = types_db.type_definition(def_id)
            type_defs[def_id] = type_def

        for unit in units:
            type_id = unit['unit_type_id']
            u = conduit_common_utils.to_plugin_associated_unit(unit, type_defs[type_id])
            transfer_units.append(u)

        # Invoke the importer
        try:
            dep_report = importer_instance.resolve_dependencies(transfer_repo, transfer_units,
                                                                conduit, call_config)
        except Exception:
            raise PulpExecutionException(), None, sys.exc_info()[2]

        return dep_report

    def get_repo_units(self, repo_id, content_type_id, additional_unit_fields=[]):
        """
        Searches for units in the given repository with given content type 
        and returns a plugin unit containing unit id, unit key and any additional
        fields requested.

        :param repo_id: repo id
        :type  repo_id: str

        :param content_type_id: content type id of the units
        :type  content_type_id: str

        :param additional_unit_fields: additional fields from the unit metadata to be added 
                                       in the result
        :type additional_unit_fields: list of str

        :return: list of unit instances
        :rtype:  list of pulp.plugins.model.Unit
        """
        try:
            # Get type definition and unit_key for given content type
            type_def = types_db.type_definition(content_type_id)
            unit_key_fields = type_def['unit_key']

            # Query repo association manager to get all units of given type
            # associated with given repo. Limit data by requesting only the fields
            # that are needed.
            query_manager = managers.repo_unit_association_query_manager()
            unit_fields = list(set(unit_key_fields + additional_unit_fields))
            criteria = UnitAssociationCriteria(association_fields=['unit_id'],
                                               unit_fields=unit_fields)
            units = query_manager.get_units_by_type(repo_id, content_type_id, criteria)

            # Convert units to plugin units with unit_key and required metadata values for each unit
            all_units = []
            for unit in units:
                unit_key = {}
                metadata = {}
                for k in unit_key_fields:
                    unit_key[k] = unit['metadata'].pop(k)
                # Add unit_id and any additional unit fields requested by plugins
                metadata['unit_id'] = unit.pop('unit_id')
                for field in additional_unit_fields:
                    metadata[field] = unit['metadata'].pop(field, None)

                u = Unit(content_type_id, unit_key, metadata, None)
                all_units.append(u)

            return all_units

        except Exception, e:
            _LOG.exception(_('Exception from server getting units from repo [%s]' % repo_id))
            raise self.exception_class(e), None, sys.exc_info()[2]

def create_transfer_units(associate_units, associated_unit_type_ids):
    type_defs = {}
    for def_id in associated_unit_type_ids:
        type_def = types_db.type_definition(def_id)
        type_defs[def_id] = type_def

    transfer_units = []
    for unit in associate_units:
        type_id = unit['unit_type_id']
        u = conduit_common_utils.to_plugin_associated_unit(unit, type_defs[type_id])
        transfer_units.append(u)

    return transfer_units

    def validate_type(content_type_id):
        """
        Validate the provided content type id.

        :param content_type_id: unique id of the content type
        :type content_type_id: str
        :raises MissingResource: if there is no such content type id

        :return: content type definition
        :rtype: dict
        """
        content_type_definition = content_types_db.type_definition(content_type_id)
        if content_type_definition is None:
            raise pulp_exceptions.MissingResource(content_type_id=content_type_id)
        return content_type_definition

    def generate_orphans_by_type_with_unit_keys(content_type_id):
        """
        Return an generator of all orphaned content units of the given content type.

        Each content unit will contain the fields specified in the content type
        definition's search indexes.

        :param content_type_id: id of the content type
        :type content_type_id: basestring
        :return: generator of orphaned content units for the given content type
        :rtype: generator
        """
        content_type_definition = content_types_db.type_definition(content_type_id)
        fields = ['_id', '_content_type_id']
        fields.extend(content_type_definition['unit_key'])

        for content_unit in OrphanManager.generate_orphans_by_type(content_type_id, fields):
            yield  content_unit

    def find_unit_by_unit_key(self, type_id, unit_key):
        """
        Finds a unit based on its unit key. If more than one unit comes back,
        an exception will be raised.

        @param type_id: indicates the type of units being retrieved
        @type  type_id: str
        @param unit_key: the unit key for the unit
        @type  unit_key: dict

        @return: a single unit
        @rtype:  L{Unit}
        """
        content_query_manager = manager_factory.content_query_manager()
        try:
            # this call returns a unit or raises MissingResource
            existing_unit = content_query_manager.get_content_unit_by_keys_dict(type_id, unit_key)
            type_def = types_db.type_definition(type_id)
            plugin_unit = common_utils.to_plugin_unit(existing_unit, type_def)
            return plugin_unit
        except MissingResource:
            return None

    def get_source_units(self, criteria=None):
        """
        Returns the collection of content units associated with the source
        repository for a unit import.

        Units returned from this call will have the id field populated and are
        useable in any calls in this conduit that require the id field.

        @param criteria: used to scope the returned results or the data within;
               the Criteria class can be imported from this module
        @type  criteria: L{Criteria}

        @return: list of unit instances
        @rtype:  list of L{AssociatedUnit}
        """

        try:
            units = self.__association_query_manager.get_units_across_types(self.source_repo_id, criteria=criteria)

            all_units = []

            # Load all type definitions in use so we don't hammer the database
            unique_type_defs = set([u['unit_type_id'] for u in units])
            type_defs = {}
            for def_id in unique_type_defs:
                type_def = types_db.type_definition(def_id)
                type_defs[def_id] = type_def

            # Convert to transfer object
            for unit in units:
                type_id = unit['unit_type_id']
                u = common_utils.to_plugin_unit(unit, type_defs[type_id])
                all_units.append(u)

            return all_units

        except Exception, e:
            _LOG.exception('Exception from server requesting all content units for repository [%s]' % self.repo_id)
            raise UnitImportConduitException(e), None, sys.exc_info()[2]

 def get_content_type(self, type_id):
     """
     """
     return content_types_db.type_definition(type_id)

 def get(self, type_id):
     typedef = self.cached.get(type_id)
     if typedef is None:
         typedef = types_db.type_definition(type_id)
         self.cached[type_id] = typedef
     return typedef

    def associate_from_repo(self, source_repo_id, dest_repo_id, criteria=None):
        """
        Creates associations in a repository based on the contents of a source
        repository. Units from the source repository can be filtered by
        specifying a criteria object.

        The destination repository must have an importer that can support
        the types of units being associated. This is done by analyzing the
        unit list and the importer metadata and takes place before the
        destination repository is called.

        Pulp does not actually perform the associations as part of this call.
        The unit list is determined and passed to the destination repository's
        importer. It is the job of the importer to make the associate calls
        back into Pulp where applicable.

        If criteria is None, the effect of this call is to copy the source
        repository's associations into the destination repository.

        @param source_repo_id: identifies the source repository
        @type  source_repo_id: str

        @param dest_repo_id: identifies the destination repository
        @type  dest_repo_id: str

        @param criteria: optional; if specified, will filter the units retrieved
                         from the source repository
        @type  criteria: L{Criteria}

        @raise MissingResource: if either of the specified repositories don't exist
        """

        # Validation
        repo_query_manager = manager_factory.repo_query_manager()
        importer_manager = manager_factory.repo_importer_manager()
        dependency_manager = manager_factory.dependency_manager()
        association_query_manager = manager_factory.repo_unit_association_query_manager()

        source_repo = repo_query_manager.get_repository(source_repo_id)
        dest_repo = repo_query_manager.get_repository(dest_repo_id)

        # This will raise MissingResource if there isn't one, which is the
        # behavior we want this method to exhibit, so just let it bubble up.
        dest_repo_importer = importer_manager.get_importer(dest_repo_id)
        source_repo_importer = importer_manager.get_importer(source_repo_id)

        # The docs are incorrect on the list_importer_types call; it actually
        # returns a dict with the types under key "types" for some reason.
        supported_type_ids = plugin_api.list_importer_types(dest_repo_importer['importer_type_id'])['types']

        # If criteria is specified, retrieve the list of units now
        associate_us = None
        if criteria is not None:
            criteria.association_fields = None
            criteria.unit_fields = None

            # Retrieve the units to be associated
            associate_us = association_query_manager.get_units(source_repo_id, criteria=criteria)

            # If units were supposed to be filtered but none matched, we're done
            if len(associate_us) is 0:
                return

        # Now we can make sure the destination repository's importer is capable
        # of importing either the selected units or all of the units
        if associate_us is not None:
            associated_unit_type_ids = set([u['unit_type_id'] for u in associate_us])
        else:
            association_query_manager = manager_factory.repo_unit_association_query_manager()

            # We may want to make a call here that only retrieves the unique
            # type IDs instead of all of the units, but for now it doesn't exist
            # and I'm not entirely sure this will be a huge problem.
            all_units = association_query_manager.get_units(source_repo_id)
            associated_unit_type_ids = set(u['unit_type_id'] for u in all_units)

        unsupported_types = [t for t in associated_unit_type_ids if t not in supported_type_ids]

        if len(unsupported_types) > 0:
            raise exceptions.InvalidValue(['types'])

        # Convert all of the units into the plugin standard representation if
        # a filter was specified
        transfer_units = None
        if associate_us is not None:
            type_defs = {}
            for def_id in associated_unit_type_ids:
                type_def = types_db.type_definition(def_id)
                type_defs[def_id] = type_def

            transfer_units = []
            for unit in associate_us:
                type_id = unit['unit_type_id']
                u = conduit_common_utils.to_plugin_unit(unit, type_defs[type_id])
                transfer_units.append(u)

        # Convert the two repos into the plugin API model
        transfer_dest_repo = common_utils.to_transfer_repo(dest_repo)
        transfer_dest_repo.working_dir = common_utils.importer_working_dir(dest_repo_importer['importer_type_id'], dest_repo['id'], mkdir=True)

#.........这里部分代码省略.........