def create_sync_schedule(self, repo_id, importer_id, sync_options, schedule_data):
        """
        Create a new sync schedule for a given repository using the given importer.
        @param repo_id:
        @param importer_id:
        @param sync_options:
        @param schedule_data:
        @return:
        """

        # validate the input
        self._validate_importer(repo_id, importer_id)
        schedule_utils.validate_keys(sync_options, _SYNC_OPTION_KEYS)
        if 'schedule' not in schedule_data:
            raise pulp_exceptions.MissingValue(['schedule'])

        # build the sync call request
        args = [repo_id]
        kwargs = {'overrides': sync_options['override_config']}
        call_request = CallRequest(sync_with_auto_publish_itinerary, args, kwargs, weight=0)

        # schedule the sync
        scheduler = dispatch_factory.scheduler()
        schedule_id = scheduler.add(call_request, **schedule_data)
        importer_manager = managers_factory.repo_importer_manager()
        importer_manager.add_sync_schedule(repo_id, schedule_id)
        return schedule_id

    def POST(self, repo_id, distributor_id):
        distributor_manager = manager_factory.repo_distributor_manager()
        distributor_manager.get_distributor(repo_id, distributor_id)

        schedule_options = self.params()
        publish_options = {'override_config': schedule_options.pop('override_config', {})}

        schedule_manager = manager_factory.schedule_manager()
        resources = {dispatch_constants.RESOURCE_REPOSITORY_TYPE: {repo_id: dispatch_constants.RESOURCE_READ_OPERATION},
                     dispatch_constants.RESOURCE_REPOSITORY_DISTRIBUTOR_TYPE: {distributor_id: dispatch_constants.RESOURCE_UPDATE_OPERATION}}
        weight = pulp_config.config.getint('tasks', 'create_weight')
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id),
                resource_tag(dispatch_constants.RESOURCE_REPOSITORY_DISTRIBUTOR_TYPE, distributor_id),
                action_tag('create_publish_schedule')]
        call_request = CallRequest(schedule_manager.create_publish_schedule,
                                   [repo_id, distributor_id, publish_options, schedule_options],
                                   resources=resources,
                                   weight=weight,
                                   tags=tags,
                                   archive=True)
        schedule_id = execution.execute_sync(call_request)

        scheduler = dispatch_factory.scheduler()
        schedule = scheduler.get(schedule_id)
        obj = serialization.dispatch.scheduled_publish_obj(schedule)
        obj.update(serialization.link.child_link_obj(schedule_id))
        return self.created(obj['_href'], obj)

    def update_publish_schedule(self, repo_id, distributor_id, schedule_id, publish_options, schedule_data):
        """
        Update an existing scheduled publish for the given repository and distributor.
        @param repo_id:
        @param distributor_id:
        @param schedule_id:
        @param publish_options:
        @param schedule_data:
        @return:
        """

        # validate the input
        self._validate_distributor(repo_id, distributor_id)
        schedule_updates = copy.copy(schedule_data)

        # prepare the call request if there are changes to the publish itself
        scheduler = dispatch_factory.scheduler()
        if publish_options:
            report = scheduler.get(schedule_id)
            call_request = report['call_request']
            if 'override_config' in publish_options:
                call_request.kwargs = {'publish_config_override': publish_options['override_config']}
            schedule_updates['call_request'] = call_request

        # update the scheduled publish
        scheduler.update(schedule_id, **schedule_updates)

    def PUT(self, repo_id, importer_id, schedule_id):
        importer_manager = manager_factory.repo_importer_manager()
        schedule_list = importer_manager.list_sync_schedules(repo_id)
        if schedule_id not in schedule_list:
            raise exceptions.MissingResource(repo=repo_id, importer=importer_id, publish_schedule=schedule_id)

        sync_updates = {}
        schedule_updates = self.params()
        if 'override_config' in schedule_updates:
            sync_updates['override_config'] = schedule_updates.pop('override_config')

        schedule_manager = manager_factory.schedule_manager()
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id),
                resource_tag(dispatch_constants.RESOURCE_REPOSITORY_IMPORTER_TYPE, importer_id),
                resource_tag(dispatch_constants.RESOURCE_SCHEDULE_TYPE, schedule_id),
                action_tag('update_sync_schedule')]
        call_request = CallRequest(schedule_manager.update_sync_schedule,
                                   [repo_id, importer_id, schedule_id, sync_updates, schedule_updates],
                                   tags=tags,
                                   archive=True)
        call_request.reads_resource(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id)
        call_request.reads_resource(dispatch_constants.RESOURCE_REPOSITORY_IMPORTER_TYPE, importer_id)
        call_request.updates_resource(dispatch_constants.RESOURCE_SCHEDULE_TYPE, schedule_id)
        execution.execute(call_request)

        scheduler = dispatch_factory.scheduler()
        schedule = scheduler.get(schedule_id)
        obj = serialization.dispatch.scheduled_sync_obj(schedule)
        obj.update(serialization.link.current_link_obj())
        return self.ok(obj)

    def test_update_schedule(self):
        units = copy.copy(_TEST_UNITS)
        uninstall_options = {'options': {}}
        schedule_data = {'schedule': 'R1/P1DT'}

        schedule_id = self.schedule_manager.create_unit_uninstall_schedule(self.consumer_id, units, uninstall_options, schedule_data)

        scheduler = dispatch_factory.scheduler()
        scheduled_call = scheduler.get(schedule_id)

        self.assertFalse(scheduled_call is None)
        self.assertTrue(schedule_data['schedule'] == scheduled_call['schedule'])
        self.assertTrue(self.consumer_id in scheduled_call['call_request'].args)
        self.assertTrue(units == scheduled_call['call_request'].kwargs['units'])
        self.assertTrue(uninstall_options['options'] == scheduled_call['call_request'].kwargs['options'])

        units.append({'type_id': 'mock-type', 'unit_key': {'id': 'redis'}})
        uninstall_options['options'] = {'option': 'value'}
        schedule_data['schedule'] = 'R3/P1DT'

        self.schedule_manager.update_unit_uninstall_schedule(self.consumer_id, schedule_id, units, uninstall_options, schedule_data)

        updated_call = scheduler.get(schedule_id)

        self.assertFalse(updated_call is None)
        self.assertTrue(schedule_data['schedule'] == updated_call['schedule'], '%s != %s' % (schedule_data['schedule'], updated_call['schedule']))
        self.assertTrue(self.consumer_id in updated_call['call_request'].args)
        self.assertTrue(units == updated_call['call_request'].kwargs['units'])
        self.assertTrue(uninstall_options['options'] == updated_call['call_request'].kwargs['options'])

    def PUT(self, repo_id, distributor_id, schedule_id):
        distributor_manager = manager_factory.repo_distributor_manager()
        schedule_list = distributor_manager.list_publish_schedules(repo_id, distributor_id)
        if schedule_id not in schedule_list:
            raise exceptions.MissingResource(repo=repo_id, distributor=distributor_id, publish_schedule=schedule_id)

        publish_update = {}
        schedule_update = self.params()
        if 'override_config' in schedule_update:
            publish_update['override_config'] = schedule_update.pop('override_config')

        schedule_manager = manager_factory.schedule_manager()
        resources = {dispatch_constants.RESOURCE_REPOSITORY_TYPE: {repo_id: dispatch_constants.RESOURCE_READ_OPERATION},
                     dispatch_constants.RESOURCE_REPOSITORY_DISTRIBUTOR_TYPE: {distributor_id: dispatch_constants.RESOURCE_READ_OPERATION},
                     dispatch_constants.RESOURCE_SCHEDULE_TYPE: {schedule_id: dispatch_constants.RESOURCE_UPDATE_OPERATION}}
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id),
                resource_tag(dispatch_constants.RESOURCE_REPOSITORY_DISTRIBUTOR_TYPE, distributor_id),
                resource_tag(dispatch_constants.RESOURCE_SCHEDULE_TYPE, schedule_id),
                action_tag('update_publish_schedule')]
        call_request = CallRequest(schedule_manager.update_publish_schedule,
                                   [repo_id, distributor_id, schedule_id, publish_update, schedule_update],
                                   resources=resources,
                                   tags=tags,
                                   archive=True)
        execution.execute(call_request)

        scheduler = dispatch_factory.scheduler()
        schedule = scheduler.get(schedule_id)
        obj = serialization.dispatch.scheduled_publish_obj(schedule)
        obj.update(serialization.link.current_link_obj())
        return self.ok(obj)

    def POST(self, consumer_id):
        consumer_manager = managers.consumer_manager()
        consumer_manager.get_consumer(consumer_id)

        schedule_data = self.params()
        units = schedule_data.pop('units', None)
        uninstall_options = {'options': schedule_data.pop('options', {})}

        if not units:
            raise MissingValue(['units'])

        schedule_manager = managers.schedule_manager()

        weight = pulp_config.config.getint('tasks', 'create_weight')
        tags = [resource_tag(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id),
                action_tag('create_unit_uninstall_schedule')]

        call_request = CallRequest(schedule_manager.create_unit_uninstall_schedule,
                                   [consumer_id, units, uninstall_options, schedule_data],
                                   weight=weight,
                                   tags=tags,
                                   archive=True)
        call_request.reads_resource(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id)

        schedule_id = execution.execute_sync(call_request)

        scheduler = dispatch_factory.scheduler()
        scheduled_call = scheduler.get(schedule_id)

        scheduled_obj = serialization.dispatch.scheduled_unit_management_obj(scheduled_call)
        scheduled_obj.update(serialization.link.child_link_obj(schedule_id))
        return self.created(scheduled_obj['_href'], scheduled_obj)

    def PUT(self, consumer_id, schedule_id):
        consumer_manager = managers.consumer_manager()
        consumer_manager.get_consumer(consumer_id)

        schedule_data = self.params()
        install_options = None
        units = schedule_data.pop('units', None)

        if 'options' in schedule_data:
            install_options = {'options': schedule_data.pop('options')}

        schedule_manager = managers.schedule_manager()

        tags = [resource_tag(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id),
                resource_tag(dispatch_constants.RESOURCE_SCHEDULE_TYPE, schedule_id),
                action_tag('update_unit_uninstall_schedule')]

        call_request = CallRequest(schedule_manager.update_unit_uninstall_schedule,
                                   [consumer_id, schedule_id, units, install_options, schedule_data],
                                   tags=tags,
                                   archive=True)
        call_request.reads_resource(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id)
        call_request.updates_resource(dispatch_constants.RESOURCE_SCHEDULE_TYPE, schedule_id)

        execution.execute(call_request)

        scheduler = dispatch_factory.scheduler()
        scheduled_call = scheduler.get(schedule_id)

        scheduled_obj = serialization.dispatch.scheduled_unit_management_obj(scheduled_call)
        scheduled_obj.update(serialization.link.current_link_obj())
        return self.ok(scheduled_obj)

    def create_publish_schedule(self, repo_id, distributor_id, publish_options, schedule_data):
        """
        Create a new scheduled publish for the given repository and distributor.
        @param repo_id:
        @param distributor_id:
        @param publish_options:
        @param schedule_data:
        @return:
        """

        # validate the input
        self._validate_distributor(repo_id, distributor_id)
        self._validate_keys(publish_options, _PUBLISH_OPTION_KEYS)
        if 'schedule' not in schedule_data:
            raise pulp_exceptions.MissingValue(['schedule'])

        # build the publish call
        publish_manager = managers_factory.repo_publish_manager()
        args = [repo_id, distributor_id]
        kwargs = {'publish_config_override': publish_options['override_config']}
        weight = pulp_config.config.getint('tasks', 'publish_weight')
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id),
                resource_tag(dispatch_constants.RESOURCE_REPOSITORY_DISTRIBUTOR_TYPE, distributor_id)]
        call_request = CallRequest(publish_manager.publish, args, kwargs, weight=weight, tags=tags, archive=True)
        call_request.reads_resource(dispatch_constants.RESOURCE_REPOSITORY_DISTRIBUTOR_TYPE, distributor_id)
        call_request.updates_resource(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id)
        call_request.add_life_cycle_callback(dispatch_constants.CALL_ENQUEUE_LIFE_CYCLE_CALLBACK, publish_manager.prep_publish)

        # schedule the publish
        scheduler = dispatch_factory.scheduler()
        schedule_id = scheduler.add(call_request, **schedule_data)
        distributor_manager = managers_factory.repo_distributor_manager()
        distributor_manager.add_publish_schedule(repo_id, distributor_id, schedule_id)
        return schedule_id

    def create_publish_schedule(self, repo_id, distributor_id, publish_options, schedule_data):
        """
        Create a new scheduled publish for the given repository and distributor.
        @param repo_id:
        @param distributor_id:
        @param publish_options:
        @param schedule_data:
        @return:
        """

        # validate the input
        self._validate_distributor(repo_id, distributor_id)
        schedule_utils.validate_keys(publish_options, _PUBLISH_OPTION_KEYS)
        if 'schedule' not in schedule_data:
            raise pulp_exceptions.MissingValue(['schedule'])

        # build the publish call
        args = [repo_id, distributor_id]
        kwargs = {'overrides': publish_options['override_config']}
        call_request = CallRequest(publish_itinerary, args, kwargs, weight=0)

        # schedule the publish
        scheduler = dispatch_factory.scheduler()
        schedule_id = scheduler.add(call_request, **schedule_data)
        distributor_manager = managers_factory.repo_distributor_manager()
        distributor_manager.add_publish_schedule(repo_id, distributor_id, schedule_id)
        return schedule_id

    def update_unit_install_schedule(self, consumer_id, schedule_id, units=None, install_options=None, schedule_data=None):
        """
        Update an existing schedule for installing content units on a consumer.
        @param consumer_id: unique id for the consumer
        @param schedule_id: unique id for the schedule
        @param units: optional list of units to install
        @param install_options: optional options to pass to the install manager
        @param schedule_data: optional schedule updates
        """
        self._validate_consumer(consumer_id)
        schedule_updates = copy.copy(schedule_data) or {}

        scheduler = dispatch_factory.scheduler()
        report = scheduler.get(schedule_id)
        call_request = report['call_request']

        if units is not None:
            call_request.kwargs['units'] = units
            schedule_updates['call_request'] = call_request

        if install_options is not None and 'options' in install_options:
            call_request.kwargs['options'] = install_options['options']
            schedule_updates['call_request'] = call_request

        scheduler.update(schedule_id, **schedule_updates)

    def create_sync_schedule(self, repo_id, importer_id, sync_options, schedule_data):
        """
        Create a new sync schedule for a given repository using the given importer.
        @param repo_id:
        @param importer_id:
        @param sync_options:
        @param schedule_data:
        @return:
        """

        # validate the input
        self._validate_importer(repo_id, importer_id)
        self._validate_keys(sync_options, _SYNC_OPTION_KEYS)
        if 'schedule' not in schedule_data:
            raise pulp_exceptions.MissingValue(['schedule'])

        # build the sync call request
        sync_manager = managers_factory.repo_sync_manager()
        args = [repo_id]
        kwargs = {'sync_config_override': sync_options['override_config']}
        weight = pulp_config.config.getint('tasks', 'sync_weight')
        tags = [resource_tag(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id),
                resource_tag(dispatch_constants.RESOURCE_REPOSITORY_IMPORTER_TYPE, importer_id)]
        call_request = CallRequest(sync_manager.sync, args, kwargs, weight=weight, tags=tags, archive=True)
        call_request.reads_resource(dispatch_constants.RESOURCE_REPOSITORY_IMPORTER_TYPE, importer_id)
        call_request.updates_resource(dispatch_constants.RESOURCE_REPOSITORY_TYPE, repo_id)
        call_request.add_life_cycle_callback(dispatch_constants.CALL_ENQUEUE_LIFE_CYCLE_CALLBACK, sync_manager.prep_sync)

        # schedule the sync
        scheduler = dispatch_factory.scheduler()
        schedule_id = scheduler.add(call_request, **schedule_data)
        importer_manager = managers_factory.repo_importer_manager()
        importer_manager.add_sync_schedule(repo_id, schedule_id)
        return schedule_id

    def test_update_schedule(self):
        publish_options = {'override_config': {}}
        schedule_data = {'schedule': 'R1/P1DT'}
        schedule_id = self.schedule_manager.create_publish_schedule(self.repo_id,
                                                                    self.distributor_id,
                                                                    publish_options,
                                                                    schedule_data)
        scheduler = dispatch_factory.scheduler()
        schedule_report = scheduler.get(schedule_id)
        self.assertTrue(schedule_id == schedule_report['_id'])
        self.assertTrue(publish_options['override_config'] == schedule_report['call_request'].kwargs['publish_config_override'])
        self.assertTrue(schedule_data['schedule'] == schedule_report['schedule'])

        new_publish_options = {'override_config': {'option_1': 'new_option'}}
        new_schedule_data = {'schedule': 'R4/PT24H', 'failure_threshold': 4}
        self.schedule_manager.update_publish_schedule(self.repo_id,
                                                      self.distributor_id,
                                                      schedule_id,
                                                      new_publish_options,
                                                      new_schedule_data)
        schedule_report = scheduler.get(schedule_id)
        self.assertTrue(schedule_id == schedule_report['_id'])
        self.assertTrue(new_publish_options['override_config'] == schedule_report['call_request'].kwargs['publish_config_override'])
        self.assertTrue(new_schedule_data['schedule'] == schedule_report['schedule'])
        self.assertTrue(new_schedule_data['failure_threshold'] == schedule_report['failure_threshold'])

    def create_unit_install_schedule(self, consumer_id, units, install_options, schedule_data ):
        """
        Create a schedule for installing content units on a consumer.
        @param consumer_id: unique id for the consumer
        @param units: list of unit type and unit key dicts
        @param install_options: options to pass to the install manager
        @param schedule_data: scheduling data
        @return: schedule id
        """
        self._validate_consumer(consumer_id)
        self._validate_keys(install_options, _UNIT_INSTALL_OPTION_KEYS)
        if 'schedule' not in schedule_data:
            raise pulp_exceptions.MissingValue(['schedule'])

        manager = managers_factory.consumer_agent_manager()
        args = [consumer_id]
        kwargs = {'units': units,
                  'options': install_options.get('options', {})}
        weight = pulp_config.config.getint('tasks', 'consumer_content_weight')
        tags = [resource_tag(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id),
                action_tag('unit_install'), action_tag('scheduled_unit_install')]
        call_request = CallRequest(manager.install_content, args, kwargs, weight=weight, tags=tags, archive=True)
        call_request.reads_resource(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id)

        scheduler = dispatch_factory.scheduler()
        schedule_id = scheduler.add(call_request, **schedule_data)
        return schedule_id

    def update_sync_schedule(self, repo_id, importer_id, schedule_id, sync_options, schedule_data):
        """
        Update an existing sync schedule.
        @param repo_id:
        @param importer_id:
        @param schedule_id:
        @param sync_options:
        @param schedule_data:
        @return:
        """

        # validate the input
        self._validate_importer(repo_id, importer_id)
        schedule_updates = copy.copy(schedule_data)

        # prepare the call request if there are changes to the sync itself
        scheduler = dispatch_factory.scheduler()
        if sync_options:
            report = scheduler.get(schedule_id)
            call_request = report['call_request']
            if 'override_config' in sync_options:
                call_request.kwargs = {'sync_config_override': sync_options['override_config']}
            schedule_updates['call_request'] = call_request

        # update the scheduled sync
        scheduler.update(schedule_id, **schedule_updates)

    def _delete_all_schedules(self, management_action_name, consumer_id):
        self._validate_consumer(consumer_id)

        scheduler = dispatch_factory.scheduler()
        consumer_tag = resource_tag(dispatch_constants.RESOURCE_CONSUMER_TYPE, consumer_id)
        management_tag = action_tag(management_action_name)
        reports = scheduler.find(consumer_tag, management_tag)

        for r in reports:
            scheduler.remove(r['call_report']['schedule_id'])

    def delete_unit_install_schedule(self, consumer_id, schedule_id):
        """
        Delete an existing schedule for installing content units on a consumer.
        @param consumer_id: unique id of the consumer
        @param schedule_id: unique id of the schedule
        """
        self._validate_consumer(consumer_id)

        scheduler = dispatch_factory.scheduler()
        scheduler.remove(schedule_id)

 def GET(self, repo_id, importer_id, schedule_id):
     importer_manager = manager_factory.repo_importer_manager()
     schedule_list = importer_manager.list_sync_schedules(repo_id)
     if schedule_id not in schedule_list:
         raise exceptions.MissingResource(repo=repo_id, importer=importer_id, publish_schedule=schedule_id)
     scheduler = dispatch_factory.scheduler()
     schedule = scheduler.get(schedule_id)
     obj = serialization.dispatch.scheduled_sync_obj(schedule)
     obj.update(serialization.link.current_link_obj())
     return self.ok(obj)

    def test_delete_schedule(self):
        schedule_data = {'schedule': 'R1/P1DT'}
        schedule_id = self.schedule_manager.create_unit_uninstall_schedule(self.consumer_id, _TEST_UNITS, {}, schedule_data)

        scheduler = dispatch_factory.scheduler()
        scheduled_call = scheduler.get(schedule_id)

        self.assertFalse(scheduled_call is None)

        self.schedule_manager.delete_unit_uninstall_schedule(self.consumer_id, schedule_id)

        self.assertRaises(pulp_exceptions.MissingResource, scheduler.get, schedule_id)

def scheduler_complete_callback(call_request, call_report):
    """
    Call back for call request results and rescheduling
    """
    scheduler = dispatch_factory.scheduler()
    scheduled_call_collection = ScheduledCall.get_collection()

    schedule_id = call_report.schedule_id
    scheduled_call = scheduled_call_collection.find_one({'_id': ObjectId(schedule_id)})

    scheduler.update_last_run(scheduled_call, call_report)
    scheduler.update_next_run(scheduled_call)