def create_bind_payload(repo_id, distributor_id, binding_config):
    """
    Requests the distributor plugin to generate the consumer bind payload.

    :param repo_id: identifies the repo being bound
    :type  repo_id: basestring
    :param distributor_id: identifies the distributor
    :type  distributor_id: basestring
    :param binding_config: config applicable only to the binding whose payload is being created
    :type  binding_config: object or None

    :return: payload to pass to the consumer
    :rtype:  dict

    :raise PulpExecutionException: if the distributor raises an error
    """
    dist = model.Distributor.objects.get_or_404(repo_id=repo_id, distributor_id=distributor_id)
    repo_obj = model.Repository.objects.get_repo_or_missing_resource(repo_id)

    distributor_instance, plugin_config = plugin_api.get_distributor_by_id(dist.distributor_type_id)

    # Let the distributor plugin verify the configuration
    call_config = PluginCallConfiguration(plugin_config, dist.config)
    transfer_repo = repo_obj.to_transfer_repo()

    try:
        return distributor_instance.create_consumer_payload(transfer_repo, call_config,
                                                            binding_config)
    except Exception:
        msg = _('Exception raised from distributor [%(d)s] generating consumer payload')
        msg = msg % {'d': distributor_id}
        _logger.exception(msg)
        raise exceptions.PulpExecutionException(), None, sys.exc_info()[2]

    def remove_distributor(repo_id, distributor_id):
        """
        Removes a distributor from a repository.

        @param repo_id: identifies the repo
        @type  repo_id: str

        @param distributor_id: identifies the distributor to delete
        @type  distributor_id: str

        @raise MissingResource: if repo_id doesn't correspond to a valid repo
        @raise MissingResource: if there is no distributor with the given ID
        """

        distributor_coll = RepoDistributor.get_collection()
        repo_obj = model.Repository.objects.get_repo_or_missing_resource(repo_id)

        repo_distributor = distributor_coll.find_one({'repo_id': repo_id, 'id': distributor_id})
        if repo_distributor is None:
            raise MissingResource(distributor=distributor_id)

        # remove schedules
        RepoPublishScheduleManager().delete_by_distributor_id(repo_id, repo_distributor['id'])

        # Call the distributor's cleanup method
        distributor_type_id = repo_distributor['distributor_type_id']
        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        call_config = PluginCallConfiguration(plugin_config, repo_distributor['config'])

        transfer_repo = repo_obj.to_transfer_repo()
        distributor_instance.distributor_removed(transfer_repo, call_config)

        # Update the database to reflect the removal
        distributor_coll.remove({'_id': repo_distributor['_id']}, safe=True)

    def create_bind_payload(self, repo_id, distributor_id):
        """
        Requests the distributor plugin to generate the consumer bind payload.

        @param repo_id: identifies the repo being bound
        @type  repo_id: str

        @param distributor_id: identifies the distributor
        @type  distributor_id: str

        @return: payload object to pass to the consumer
        @rtype:  dict

        @raise MissingResource: if the repo or distributor do not exist
        @raise PulpExecutionException: if the distributor raises an error
        """

        # Input Validation
        repo_distributor = self.get_distributor(repo_id, distributor_id)
        repo = Repo.get_collection().find_one({'id' : repo_id})

        distributor_type_id = repo_distributor['distributor_type_id']
        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        # Let the distributor plugin verify the configuration
        call_config = PluginCallConfiguration(plugin_config, repo_distributor['config'])
        transfer_repo = common_utils.to_transfer_repo(repo)
        transfer_repo.working_dir = common_utils.distributor_working_dir(distributor_type_id, repo_id)

        try:
            payload = distributor_instance.create_consumer_payload(transfer_repo, call_config)
            return payload
        except Exception:
            _LOG.exception('Exception raised from distributor [%s] generating consumer payload' % distributor_id)
            raise PulpExecutionException(), None, sys.exc_info()[2]

def delete(repo_id, dist_id):
    """
    Removes a distributor from a repository and unbinds any bound consumers.

    :param distributor: distributor to be deleted
    :type  distributor: pulp.server.db.model.Distributor

    :return: result containing any errors and tasks spawned
    :rtype pulp.server.async.tasks.TaskResult
    """

    distributor = model.Distributor.objects.get_or_404(repo_id=repo_id, distributor_id=dist_id)
    managers.repo_publish_schedule_manager().delete_by_distributor_id(repo_id, dist_id)

    # Call the distributor's cleanup method
    dist_instance, plugin_config = plugin_api.get_distributor_by_id(distributor.distributor_type_id)

    call_config = PluginCallConfiguration(plugin_config, distributor.config)
    repo = model.Repository.objects.get_repo_or_missing_resource(repo_id)
    dist_instance.distributor_removed(repo.to_transfer_repo(), call_config)
    distributor.delete()

    unbind_errors = []
    additional_tasks = []
    options = {}

    bind_manager = managers.consumer_bind_manager()
    for bind in bind_manager.find_by_distributor(repo_id, dist_id):
        try:
            report = bind_manager.unbind(bind['consumer_id'], bind['repo_id'],
                                         bind['distributor_id'], options)
            if report:
                additional_tasks.extend(report.spawned_tasks)
        except Exception, e:
            unbind_errors.append(e)

 def _get_distributor_instance_and_config(self, repo_id, distributor_id):
     repo_distributor_manager = manager_factory.repo_distributor_manager()
     try:
         repo_distributor = repo_distributor_manager.get_distributor(repo_id, distributor_id)
         distributor, config = plugin_api.get_distributor_by_id(repo_distributor['distributor_type_id'])
     except (MissingResource, plugin_exceptions.PluginNotFound):
         distributor = None
         config = None
     return distributor, config

 def test_distributors(self):
     # listing
     distributors = api.list_distributors()
     self.assertEqual(len(distributors), 1)
     self.assertEqual(distributors, {DISTRIBUTOR_ID: METADATA})
     # list types
     self.assertEqual(api.list_distributor_types(DISTRIBUTOR_ID), METADATA)
     # by id
     distributor = api.get_distributor_by_id(DISTRIBUTOR_ID)
     self.assertFalse(distributor is None)
     self.assertTrue(isinstance(distributor[0], MockDistributor))
     self.assertRaises(PluginNotFound, api.get_distributor_by_id, 'not-valid')
     # is_valid
     self.assertTrue(api.is_valid_distributor(DISTRIBUTOR_ID))
     self.assertFalse(api.is_valid_distributor('not-valid'))

def _get_distributor_instance_and_config(repo_id, distributor_id):
    """
    For a given repository and distributor, retrieve the instance of the distributor and its
    configuration from the plugin api.

    :param repo_id: identifies the repository
    :type  repo_id: str
    :param distributor_id: identifies the distributor
    :type  distributor_id: str

    :return: distributor instance and config
    :rtype:  tuple
    """
    dist = model.Distributor.objects.get_or_404(repo_id=repo_id, distributor_id=distributor_id)
    distributor, config = plugin_api.get_distributor_by_id(dist.distributor_type_id)
    return distributor, config

def _get_distributor_instance_and_config(repo_id, distributor_id):
    """
    For a given repository and distributor, retrieve the instance of the distributor and its
    configuration from the plugin api.

    :param repo_id: identifies the repository
    :type  repo_id: str
    :param distributor_id: identifies the distributor
    :type  distributor_id: str

    :return: distributor instance and config
    :rtype:  tuple
    """
    repo_distributor_manager = manager_factory.repo_distributor_manager()
    repo_distributor = repo_distributor_manager.get_distributor(repo_id, distributor_id)
    distributor, config = plugin_api.get_distributor_by_id(repo_distributor['distributor_type_id'])
    return distributor, config

    def remove_distributor(self, repo_id, distributor_id):
        """
        Removes a distributor from a repository.

        @param repo_id: identifies the repo
        @type  repo_id: str

        @param distributor_id: identifies the distributor to delete
        @type  distributor_id: str

        @raise MissingResource: if repo_id doesn't correspond to a valid repo
        @raise MissingResource: if there is no distributor with the given ID
        """

        repo_coll = Repo.get_collection()
        distributor_coll = RepoDistributor.get_collection()

        # Validation
        repo = repo_coll.find_one({'id' : repo_id})
        if repo is None:
            raise MissingResource(repo_id)

        repo_distributor = distributor_coll.find_one({'repo_id' : repo_id, 'id' : distributor_id})
        if repo_distributor is None:
            raise MissingResource(distributor_id)

        # Call the distributor's cleanup method
        distributor_type_id = repo_distributor['distributor_type_id']
        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        call_config = PluginCallConfiguration(plugin_config, repo_distributor['config'])

        transfer_repo = common_utils.to_transfer_repo(repo)
        transfer_repo.working_dir = common_utils.distributor_working_dir(distributor_type_id, repo_id)

        distributor_instance.distributor_removed(transfer_repo, call_config)

        # clean up binds
        bind_manager = manager_factory.consumer_bind_manager()
        bind_manager.distributor_deleted(repo_id, distributor_id)

        # Update the database to reflect the removal
        distributor_coll.remove(repo_distributor, safe=True)

    def create_bind_payload(self, repo_id, distributor_id, binding_config):
        """
        Requests the distributor plugin to generate the consumer bind payload.

        @param repo_id: identifies the repo being bound
        @type  repo_id: str

        @param distributor_id: identifies the distributor
        @type  distributor_id: str

        @param binding_config: configuration applicable only to the binding whose
               payload is being created; may be None
        @type  binding_config: object or None

        @return: payload object to pass to the consumer
        @rtype:  dict

        @raise MissingResource: if the repo or distributor do not exist
        @raise PulpExecutionException: if the distributor raises an error
        """

        # Input Validation
        repo_distributor = self.get_distributor(repo_id, distributor_id)
        repo_obj = model.Repository.objects.get_repo_or_missing_resource(repo_id)

        distributor_type_id = repo_distributor['distributor_type_id']
        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        # Let the distributor plugin verify the configuration
        call_config = PluginCallConfiguration(plugin_config, repo_distributor['config'])
        transfer_repo = repo_obj.to_transfer_repo()

        try:
            payload = distributor_instance.create_consumer_payload(transfer_repo, call_config,
                                                                   binding_config)
            return payload
        except Exception:
            msg = _('Exception raised from distributor [%(d)s] generating consumer payload')
            msg = msg % {'d': distributor_id}
            _logger.exception(msg)
            raise PulpExecutionException(), None, sys.exc_info()[2]

    def add_distributor(repo_id, distributor_type_id, repo_plugin_config,
                        auto_publish, distributor_id=None):
        """
        Adds an association from the given repository to a distributor. The
        association will be tracked through the distributor_id; each distributor
        on a given repository must have a unique ID. If this is not specified,
        one will be generated. If a distributor already exists on the repo for
        the given ID, the existing one will be removed and replaced with the
        newly configured one.

        :param repo_id:                         identifies the repo
        :type  repo_id:                         str
        :param distributor_type_id:             identifies the distributor; must correspond to a
                                                distributor loaded at server startup
        :type  distributor_type_id:             str
        :param repo_plugin_config:              configuration the repo will use with this
                                                distributor; may be None
        :type  repo_plugin_config:              dict
        :param auto_publish:                    if true, this distributor will be invoked at the end
                                                of every sync
        :type  auto_publish:                    bool
        :param distributor_id:                  unique ID to refer to this distributor for this repo
        :type  distributor_id:                  str
        :return:                                ID assigned to the distributor (only valid in
                                                conjunction with the repo)
        :raise MissingResource:                 if the given repo_id does not refer to a valid repo
        :raise InvalidValue:                    if the distributor ID is provided and unacceptable
        :raise InvalidDistributorConfiguration: if the distributor plugin does not accept the given
                                                configuration
        """

        distributor_coll = RepoDistributor.get_collection()
        repo_obj = model.Repository.objects.get_repo_or_missing_resource(repo_id)

        if not plugin_api.is_valid_distributor(distributor_type_id):
            raise InvalidValue(['distributor_type_id'])

        # Determine the ID for this distributor on this repo; will be
        # unique for all distributors on this repository but not globally
        if distributor_id is None:
            distributor_id = str(uuid.uuid4())
        else:
            # Validate if one was passed in
            if not is_distributor_id_valid(distributor_id):
                raise InvalidValue(['distributor_id'])

        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        # Convention is that a value of None means unset. Remove any keys that
        # are explicitly set to None so the plugin will default them.
        if repo_plugin_config is not None:
            clean_config = dict([(k, v) for k, v in repo_plugin_config.items() if v is not None])
        else:
            clean_config = None

        # Let the distributor plugin verify the configuration
        call_config = PluginCallConfiguration(plugin_config, clean_config)
        config_conduit = RepoConfigConduit(distributor_type_id)

        transfer_repo = repo_obj.to_transfer_repo()
        result = distributor_instance.validate_config(transfer_repo, call_config, config_conduit)

        # For backward compatibility with plugins that don't yet return the tuple
        if isinstance(result, bool):
            valid_config = result
            message = None
        else:
            valid_config, message = result

        if not valid_config:
            raise PulpDataException(message)

        # Remove the old distributor if it exists
        try:
            RepoDistributorManager.remove_distributor(repo_id, distributor_id)
        except MissingResource:
            pass  # if it didn't exist, no problem

        # Let the distributor plugin initialize the repository
        try:
            distributor_instance.distributor_added(transfer_repo, call_config)
        except Exception:
            msg = _('Error initializing distributor [%(d)s] for repo [%(r)s]')
            msg = msg % {'d': distributor_type_id, 'r': repo_id}
            _logger.exception(msg)
            raise PulpExecutionException(), None, sys.exc_info()[2]

        # Database Update
        distributor = RepoDistributor(repo_id, distributor_id, distributor_type_id, clean_config,
                                      auto_publish)
        distributor_coll.save(distributor, safe=True)

        return distributor

    def add_distributor(self, repo_id, distributor_type_id, repo_plugin_config,
                        auto_publish, distributor_id=None):
        """
        Adds an association from the given repository to a distributor. The
        association will be tracked through the distributor_id; each distributor
        on a given repository must have a unique ID. If this is not specified,
        one will be generated. If a distributor already exists on the repo for
        the given ID, the existing one will be removed and replaced with the
        newly configured one.

        @param repo_id: identifies the repo
        @type  repo_id: str

        @param distributor_type_id: identifies the distributor; must correspond
                                    to a distributor loaded at server startup
        @type  distributor_type_id: str

        @param repo_plugin_config: configuration the repo will use with this distributor; may be None
        @type  repo_plugin_config: dict

        @param auto_publish: if true, this distributor will be invoked at
                             the end of every sync
        @type  auto_publish: bool

        @param distributor_id: unique ID to refer to this distributor for this repo
        @type  distributor_id: str

        @return: ID assigned to the distributor (only valid in conjunction with the repo)

        @raise MissingResource: if the given repo_id does not refer to a valid repo
        @raise InvalidValue: if the distributor ID is provided and unacceptable
        @raise InvalidDistributorConfiguration: if the distributor plugin does not
               accept the given configuration
        """

        repo_coll = Repo.get_collection()
        distributor_coll = RepoDistributor.get_collection()

        # Validation
        repo = repo_coll.find_one({'id' : repo_id})
        if repo is None:
            raise MissingResource(repo_id)

        if not plugin_api.is_valid_distributor(distributor_type_id):
            raise InvalidValue(['distributor_type_id'])

        # Determine the ID for this distributor on this repo; will be
        # unique for all distributors on this repository but not globally
        if distributor_id is None:
            distributor_id = str(uuid.uuid4())
        else:
            # Validate if one was passed in
            if not is_distributor_id_valid(distributor_id):
                raise InvalidValue(['distributor_id'])

        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        # Convention is that a value of None means unset. Remove any keys that
        # are explicitly set to None so the plugin will default them.
        if repo_plugin_config is not None:
            clean_config = dict([(k, v) for k, v in repo_plugin_config.items() if v is not None])
        else:
            clean_config = None

        # Let the distributor plugin verify the configuration
        call_config = PluginCallConfiguration(plugin_config, clean_config)
        transfer_repo = common_utils.to_transfer_repo(repo)
        transfer_repo.working_dir = common_utils.distributor_working_dir(distributor_type_id, repo_id)

        query_manager = manager_factory.repo_query_manager()
        related_repos = query_manager.find_with_distributor_type(distributor_type_id)

        transfer_related_repos = []
        for r in related_repos:
            all_configs = [d['config'] for d in r['distributors']]
            trr = common_utils.to_related_repo(r, all_configs)
            transfer_related_repos.append(trr)

        try:
            result = distributor_instance.validate_config(transfer_repo, call_config, transfer_related_repos)

            # For backward compatibility with plugins that don't yet return the tuple
            if isinstance(result, bool):
                valid_config = result
                message = None
            else:
                valid_config, message = result
        except Exception, e:
            _LOG.exception('Exception received from distributor [%s] while validating config' % distributor_type_id)
            raise PulpDataException(e.args), None, sys.exc_info()[2]

def update(repo_id, dist_id, config=None, delta=None):
    """
    Update the distributor and (re)bind any bound consumers.

    :param distributor: distributor to be updated
    :type  distributor: pulp.server.db.model.Distributor
    :param config: A configuration dictionary for a distributor instance. The contents of this dict
                   depends on the type of distributor. Values of None will remove they key from the
                   config. Keys ommited from this dictionary will remain unchanged.
    :type  config: dict
    :param delta: A dictionary used to change conf values for a distributor instance. This currently
                  only supports the 'auto_publish' keyword, which should have a value of type bool
    :type  delta: dict or None

    :return: result containing any errors and tasks spawned
    :rtype pulp.server.async.tasks.TaskResult
    """
    repo = model.Repository.objects.get_repo_or_missing_resource(repo_id)
    distributor = model.Distributor.objects.get_or_404(repo_id=repo_id, distributor_id=dist_id)

    for k, v in config.iteritems():
        if v is None:
            distributor.config.pop(k)
        else:
            distributor.config[k] = v

    auto_publish = delta.get('auto_publish') if delta else None
    if isinstance(auto_publish, bool):
        distributor.auto_publish = auto_publish
    elif not isinstance(auto_publish, type(None)):
        raise exceptions.InvalidValue(['auto_publish'])

    # Let the distributor plugin verify the configuration
    distributor_instance, plugin_config = plugin_api.get_distributor_by_id(
        distributor.distributor_type_id)
    call_config = PluginCallConfiguration(plugin_config, distributor.config)
    transfer_repo = repo.to_transfer_repo()
    config_conduit = RepoConfigConduit(distributor.distributor_type_id)

    result = distributor_instance.validate_config(transfer_repo, call_config,
                                                  config_conduit)

    # For backward compatibility with plugins that don't yet return the tuple
    if isinstance(result, bool):
        valid_config = result
        message = None
    else:
        valid_config, message = result

    if not valid_config:
        raise exceptions.PulpDataException(message)
    distributor.save()

    unbind_errors = []
    additional_tasks = []
    options = {}
    bind_manager = managers.consumer_bind_manager()
    for bind in bind_manager.find_by_distributor(distributor.repo_id, distributor.distributor_id):
        try:
            report = bind_manager.bind(bind['consumer_id'], bind['repo_id'], bind['distributor_id'],
                                       bind['notify_agent'], bind['binding_config'], options)
            if report:
                additional_tasks.extend(report.spawned_tasks)
        except Exception, e:
            unbind_errors.append(e)

    def update_distributor_config(self, repo_id, distributor_id, distributor_config):
        """
        Attempts to update the saved configuration for the given distributor.
        The distributor will be asked if the new configuration is valid. If not,
        this method will raise an error and the existing configuration will
        remain unchanged.

        @param repo_id: identifies the repo
        @type  repo_id: str

        @param distributor_id: identifies the distributor on the repo
        @type  distributor_id: str

        @param distributor_config: new configuration values to use
        @type  distributor_config: dict

        @return: the updated distributor
        @rtype:  dict

        @raise MissingResource: if the given repo or distributor doesn't exist
        @raise PulpDataException: if the plugin rejects the given changes
        """

        repo_coll = Repo.get_collection()
        distributor_coll = RepoDistributor.get_collection()

        # Input Validation
        repo = repo_coll.find_one({'id' : repo_id})
        if repo is None:
            raise MissingResource(repo_id)

        repo_distributor = distributor_coll.find_one({'repo_id' : repo_id, 'id' : distributor_id})
        if repo_distributor is None:
            raise MissingResource(distributor_id)

        distributor_type_id = repo_distributor['distributor_type_id']
        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        # The supplied config is a delta of changes to make to the existing config.
        # The plugin expects a full configuration, so we apply those changes to
        # the original config and pass that to the plugin's validate method.
        merged_config = dict(repo_distributor['config'])

        # The convention is that None in an update is removing the value and
        # setting it to the default. Find all such properties in this delta and
        # remove them from the existing config if they are there.
        unset_property_names = [k for k in distributor_config if distributor_config[k] is None]
        for key in unset_property_names:
            merged_config.pop(key, None)
            distributor_config.pop(key, None)

        # Whatever is left over are the changed/added values, so merge them in.
        merged_config.update(distributor_config)

        # Let the distributor plugin verify the configuration
        call_config = PluginCallConfiguration(plugin_config, merged_config)
        transfer_repo = common_utils.to_transfer_repo(repo)
        transfer_repo.working_dir = common_utils.distributor_working_dir(distributor_type_id, repo_id)

        query_manager = manager_factory.repo_query_manager()
        related_repos = query_manager.find_with_distributor_type(distributor_type_id)

        transfer_related_repos = []
        for r in related_repos:

            # Don't include the repo being updated in this list
            if r['id'] == repo_id:
                continue

            all_configs = [d['config'] for d in r['distributors']]
            trr = common_utils.to_related_repo(r, all_configs)
            transfer_related_repos.append(trr)

        try:
            result = distributor_instance.validate_config(transfer_repo, call_config, transfer_related_repos)

            # For backward compatibility with plugins that don't yet return the tuple
            if isinstance(result, bool):
                valid_config = result
                message = None
            else:
                valid_config, message = result
        except Exception, e:
            _LOG.exception('Exception raised from distributor [%s] while validating config for repo [%s]' % (distributor_type_id, repo_id))
            raise PulpDataException(e.args), None, sys.exc_info()[2]

def add_distributor(repo_id, distributor_type_id, repo_plugin_config,
                    auto_publish, distributor_id=None):
    """
    Adds an association from the given repository to a distributor. The distributor_id is unique for
    a given repository. If distributor_id is not specified, one will be generated. If a distributor
    already exists on the repo for the given ID, the existing one will be removed and replaced with
    the newly configured one.

    :param repo_id: identifies the repo
    :type  repo_id: basestring
    :param distributor_type_id: must correspond to a distributor type loaded at server startup
    :type  distributor_type_id: basestring
    :param repo_plugin_config: configuration the repo will use with this distributor
    :type  repo_plugin_config: dict or None
    :param auto_publish: if True, this distributor will be invoked at the end of every sync
    :type  auto_publish: bool
    :param distributor_id: unique ID to refer to this distributor for this repo
    :type  distributor_id: basestring
    :return: distributor object
    :rtype:  pulp.server.db.model.Distributor

    :raise InvalidValue: if the distributor ID is provided and unacceptable
    :raise exceptions.PulpDataException: if the plugin returns that the config is invalid
    """

    repo_obj = model.Repository.objects.get_repo_or_missing_resource(repo_id)

    if not plugin_api.is_valid_distributor(distributor_type_id):
        raise exceptions.InvalidValue(['distributor_type_id'])

    if distributor_id is None:
        distributor_id = str(uuid.uuid4())

    distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

    # Remove any keys whose values are explicitly set to None so the plugin will default them.
    if repo_plugin_config is not None:
        clean_config = dict([(k, v) for k, v in repo_plugin_config.items() if v is not None])
    else:
        clean_config = None

    # Let the distributor plugin verify the configuration
    call_config = PluginCallConfiguration(plugin_config, clean_config)
    config_conduit = RepoConfigConduit(distributor_type_id)
    transfer_repo = repo_obj.to_transfer_repo()
    result = distributor_instance.validate_config(transfer_repo, call_config, config_conduit)

    # For backward compatibility with plugins that don't yet return the tuple
    if isinstance(result, bool):
        valid_config = result
        message = None
    else:
        valid_config, message = result

    if not valid_config:
        raise exceptions.PulpDataException(message)

    try:
        model.Distributor.objects.get_or_404(repo_id=repo_id, distributor_id=distributor_id)
        delete(repo_id, distributor_id)
    except exceptions.MissingResource:
        pass  # if it didn't exist, no problem

    distributor_instance.distributor_added(transfer_repo, call_config)
    distributor = model.Distributor(repo_id, distributor_id, distributor_type_id, clean_config,
                                    auto_publish)
    distributor.save()
    return distributor

    def update_distributor_config(repo_id, distributor_id, distributor_config, auto_publish=None):
        """
        Attempts to update the saved configuration for the given distributor.
        The distributor will be asked if the new configuration is valid. If not,
        this method will raise an error and the existing configuration will
        remain unchanged.

        :param repo_id: identifies the repo
        :type  repo_id: str

        :param distributor_id: identifies the distributor on the repo
        :type  distributor_id: str

        :param distributor_config: new configuration values to use
        :type  distributor_config: dict

        :param auto_publish: If true, this distributor is used automatically during a sync operation
        :type auto_publish: bool

        :return: the updated distributor
        :rtype:  dict

        :raise MissingResource: if the given repo or distributor doesn't exist
        :raise PulpDataException: if the plugin rejects the given changes
        """

        distributor_coll = RepoDistributor.get_collection()
        repo_obj = model.Repository.objects.get_repo_or_missing_resource(repo_id)
        repo_distributor = distributor_coll.find_one({'repo_id': repo_id, 'id': distributor_id})
        if repo_distributor is None:
            raise MissingResource(distributor=distributor_id)

        distributor_type_id = repo_distributor['distributor_type_id']
        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        # The supplied config is a delta of changes to make to the existing config.
        # The plugin expects a full configuration, so we apply those changes to
        # the original config and pass that to the plugin's validate method.
        merged_config = dict(repo_distributor['config'])

        # The convention is that None in an update is removing the value and
        # setting it to the default. Find all such properties in this delta and
        # remove them from the existing config if they are there.
        unset_property_names = [k for k in distributor_config if distributor_config[k] is None]
        for key in unset_property_names:
            merged_config.pop(key, None)
            distributor_config.pop(key, None)

        # Whatever is left over are the changed/added values, so merge them in.
        merged_config.update(distributor_config)

        # Let the distributor plugin verify the configuration
        call_config = PluginCallConfiguration(plugin_config, merged_config)
        transfer_repo = repo_obj.to_transfer_repo()
        config_conduit = RepoConfigConduit(distributor_type_id)

        result = distributor_instance.validate_config(transfer_repo, call_config,
                                                      config_conduit)

        # For backward compatibility with plugins that don't yet return the tuple
        if isinstance(result, bool):
            valid_config = result
            message = None
        else:
            valid_config, message = result

        if not valid_config:
            raise PulpDataException(message)

        # Confirm that the auto_publish value is sane before updating the value, if it exists
        if auto_publish is not None:
            if isinstance(auto_publish, bool):
                repo_distributor['auto_publish'] = auto_publish
            else:
                raise InvalidValue(['auto_publish'])

        # If we got this far, the new config is valid, so update the database
        repo_distributor['config'] = merged_config
        distributor_coll.save(repo_distributor, safe=True)

        return repo_distributor

    def update_distributor_config(repo_id, distributor_id, distributor_config, auto_publish=None):
        """
        Attempts to update the saved configuration for the given distributor.
        The distributor will be asked if the new configuration is valid. If not,
        this method will raise an error and the existing configuration will
        remain unchanged.

        :param repo_id: identifies the repo
        :type  repo_id: str

        :param distributor_id: identifies the distributor on the repo
        :type  distributor_id: str

        :param distributor_config: new configuration values to use
        :type  distributor_config: dict

        :param auto_publish: If true, this distributor is used automatically during a sync operation
        :type auto_publish: bool

        :return: the updated distributor
        :rtype:  dict

        :raise MissingResource: if the given repo or distributor doesn't exist
        :raise PulpDataException: if the plugin rejects the given changes
        """

        repo_coll = Repo.get_collection()
        distributor_coll = RepoDistributor.get_collection()

        # Input Validation
        repo = repo_coll.find_one({'id': repo_id})
        if repo is None:
            raise MissingResource(repository=repo_id)

        repo_distributor = distributor_coll.find_one({'repo_id': repo_id, 'id': distributor_id})
        if repo_distributor is None:
            raise MissingResource(distributor=distributor_id)

        distributor_type_id = repo_distributor['distributor_type_id']
        distributor_instance, plugin_config = plugin_api.get_distributor_by_id(distributor_type_id)

        # The supplied config is a delta of changes to make to the existing config.
        # The plugin expects a full configuration, so we apply those changes to
        # the original config and pass that to the plugin's validate method.
        merged_config = dict(repo_distributor['config'])

        # The convention is that None in an update is removing the value and
        # setting it to the default. Find all such properties in this delta and
        # remove them from the existing config if they are there.
        unset_property_names = [k for k in distributor_config if distributor_config[k] is None]
        for key in unset_property_names:
            merged_config.pop(key, None)
            distributor_config.pop(key, None)

        # Whatever is left over are the changed/added values, so merge them in.
        merged_config.update(distributor_config)

        # Let the distributor plugin verify the configuration
        call_config = PluginCallConfiguration(plugin_config, merged_config)
        transfer_repo = common_utils.to_transfer_repo(repo)
        transfer_repo.working_dir = common_utils.distributor_working_dir(distributor_type_id,
                                                                         repo_id)
        config_conduit = RepoConfigConduit(distributor_type_id)

        try:
            result = distributor_instance.validate_config(transfer_repo, call_config,
                                                          config_conduit)

            # For backward compatibility with plugins that don't yet return the tuple
            if isinstance(result, bool):
                valid_config = result
                message = None
            else:
                valid_config, message = result
        except Exception, e:
            msg = _('Exception raised from distributor [%(d)s] while validating config for repo '
                    '[%(r)s]')
            msg = msg % {'d': distributor_type_id, 'r': repo_id}
            _logger.exception(msg)
            raise PulpDataException(e.args), None, sys.exc_info()[2]

 def _get_distributor_instance_and_config(repo_id, distributor_id):
     repo_distributor_manager = manager_factory.repo_distributor_manager()
     repo_distributor = repo_distributor_manager.get_distributor(repo_id, distributor_id)
     distributor, config = plugin_api.get_distributor_by_id(
         repo_distributor['distributor_type_id'])
     return distributor, config