harvester/ext/importer/ckan_sync_client.py
def sync(self, source_name, data):
        """
        Synchronize data from a source into Ckan.

        - datasets are matched by _harvest_source
        - groups and organizations are matched by name

        :param source_name:
            String identifying the source of the data. Used to build
            ids that will be used in further synchronizations.
        :param data:
            Data to be synchronized. Should be a dict (or dict-like)
            with top level keys coresponding to the object type,
            mapping to dictionaries of ``{'id': <object>}``.
        """

        groups = dict(
            (key, CkanGroup(val))
            for key, val in data['group'].iteritems())

        organizations = dict(
            (key, CkanOrganization(val))
            for key, val in data['organization'].iteritems())

        # Upsert groups and organizations
        groups_map = self._upsert_groups(groups)
        orgs_map = self._upsert_organizations(organizations)

        # Create list of datasets to be synced
        logger.info('Creating list of datasets to be synchronized')
        source_datasets = {}
        for source_id, dataset_dict in data['dataset'].iteritems():
            _dataset_dict = copy.deepcopy(dataset_dict)

            # We need to make sure "source" datasets
            # don't have (otherwise misleading) ids
            _dataset_dict.pop('id', None)

            # We need to update groups and organizations,
            # to map their name from the source into a
            # ckan id
            _dataset_dict['groups'] = [
                groups_map.to_ckan(grp_id)
                for grp_id in _dataset_dict['groups']
            ]
            _dataset_dict['owner_org'] = \
                orgs_map.to_ckan(_dataset_dict['owner_org'])

            dataset = CkanDataset(_dataset_dict)

            # We also want to add the "source id", used for further
            # synchronizations to find stuff
            dataset.extras[HARVEST_SOURCE_ID_FIELD] = \
                self._join_source_id(source_name, source_id)

            source_datasets[source_id] = dataset

        # Retrieve list of datasets from Ckan
        logger.info('Retrieving current status from Ckan')
        ckan_datasets = self._find_datasets_by_source(source_name)

        # Compare collections to find differences
        differences = self._compare_collections(
            ckan_datasets, source_datasets)

        # ------------------------------------------------------------
        # We now need to create/update/delete datasets.

        # todo: we need to make sure dataset names are not
        # already used by another dataset. The only
        # way is to randomize resource names and hope
        # a 409 response indicates duplicate name..

        # _progress_total = sum(len(differences[x])
        #                       for x in ('left', 'right', 'differing'))
        # _progress_next = itertools.count(1).next
        # report_progress(0, _progress_total)

        _prog_tot_add = len(differences['right'])
        _prog_next_add = itertools.count(1).next
        _prog_tot_remove = len(differences['left'])
        _prog_next_remove = itertools.count(1).next
        _prog_tot_update = len(differences['differing'])
        _prog_next_update = itertools.count(1).next

        # Create progress bars early..
        report_progress(('datasets', 'delete'), 0, _prog_tot_remove)
        report_progress(('datasets', 'create'), 0, _prog_tot_add)
        report_progress(('datasets', 'update'), 0, _prog_tot_update)

        # We delete first, in order to (possibly) deallocate
        # some already-used names..
        for source_id in differences['left']:
            ckan_id = ckan_datasets[source_id].id
            logger.info('Deleting dataset {0}'.format(ckan_id))
            self._client.delete_dataset(ckan_id)
            report_progress(('datasets', 'delete'),
                            _prog_next_remove(), _prog_tot_remove)

        def force_dataset_operation(operation, dataset, retry=5):
            # Maximum dataset name length is 100 characters
            # We trim it down to 80 just to be safe.

            # Note: we generally want to preserve the original name
            #       and there should *never* be problems with that
            #       when updating..

            _orig_name = dataset.name[:80]
            dataset.name = _orig_name

            while True:
                try:
                    result = operation(dataset)
                except HTTPError, e:
                    if e.status_code != 409:
                        raise
                    retry -= 1
                    if retry < 0:
                        raise
                    dataset.name = '{0}-{1:06d}'.format(
                        _orig_name,
                        random.randint(0, 999999))
                    logger.debug('Got 409: trying to rename dataset to {0}'
                                 .format(dataset.name))
                else:
                    return result

        # Create missing datasets
        for source_id in differences['right']:
            logger.info('Creating dataset {0}'.format(source_id))
            dataset = source_datasets[source_id]
            force_dataset_operation(self._client.create_dataset, dataset)
            report_progress(('datasets', 'create'),
                            _prog_next_add(), _prog_tot_add)

        # Update outdated datasets
        for source_id in differences['differing']:
            logger.info('Updating dataset {0}'.format(source_id))
            # dataset = source_datasets[source_id]
            old_dataset = ckan_datasets[source_id]
            new_dataset = source_datasets[source_id]
            dataset = self._merge_datasets(old_dataset, new_dataset)
            dataset.id = old_dataset.id  # Mandatory!
            self._client.update_dataset(dataset)  # should never fail!
            report_progress(('datasets', 'update'),
                            _prog_next_update(), _prog_tot_update)

        Synchronize data from a source into Ckan.

        - datasets are matched by _harvest_source
        - groups and organizations are matched by name

        :param source_name:
            String identifying the source of the data. Used to build
            ids that will be used in further synchronizations.
        :param data:
            Data to be synchronized. Should be a dict (or dict-like)
            with top level keys coresponding to the object type,
            mapping to dictionaries of ``{'id': <object>}``.
        
Synchronize data from a source into Ckan.
datasets are matched by _harvest_source
groups and organizations are matched by name
Synchronize data from a source into Ckan . datasets are matched by _harvest_source groups and organizations are matched by name
def sync(self, source_name, data):
        groups = dict(
            (key, CkanGroup(val))
            for key, val in data['group'].iteritems())
        organizations = dict(
            (key, CkanOrganization(val))
            for key, val in data['organization'].iteritems())
        groups_map = self._upsert_groups(groups)
        orgs_map = self._upsert_organizations(organizations)
        logger.info('Creating list of datasets to be synchronized')
        source_datasets = {}
        for source_id, dataset_dict in data['dataset'].iteritems():
            _dataset_dict = copy.deepcopy(dataset_dict)
            _dataset_dict.pop('id', None)
            _dataset_dict['groups'] = [
                groups_map.to_ckan(grp_id)
                for grp_id in _dataset_dict['groups']
            ]
            _dataset_dict['owner_org'] = \
                orgs_map.to_ckan(_dataset_dict['owner_org'])
            dataset = CkanDataset(_dataset_dict)
            dataset.extras[HARVEST_SOURCE_ID_FIELD] = \
                self._join_source_id(source_name, source_id)
            source_datasets[source_id] = dataset
        logger.info('Retrieving current status from Ckan')
        ckan_datasets = self._find_datasets_by_source(source_name)
        differences = self._compare_collections(
            ckan_datasets, source_datasets)
        _prog_tot_add = len(differences['right'])
        _prog_next_add = itertools.count(1).next
        _prog_tot_remove = len(differences['left'])
        _prog_next_remove = itertools.count(1).next
        _prog_tot_update = len(differences['differing'])
        _prog_next_update = itertools.count(1).next
        report_progress(('datasets', 'delete'), 0, _prog_tot_remove)
        report_progress(('datasets', 'create'), 0, _prog_tot_add)
        report_progress(('datasets', 'update'), 0, _prog_tot_update)
        for source_id in differences['left']:
            ckan_id = ckan_datasets[source_id].id
            logger.info('Deleting dataset {0}'.format(ckan_id))
            self._client.delete_dataset(ckan_id)
            report_progress(('datasets', 'delete'),
                            _prog_next_remove(), _prog_tot_remove)
        def force_dataset_operation(operation, dataset, retry=5):
            _orig_name = dataset.name[:80]
            dataset.name = _orig_name
            while True:
                try:
                    result = operation(dataset)
                except HTTPError, e:
                    if e.status_code != 409:
                        raise
                    retry -= 1
                    if retry < 0:
                        raise
                    dataset.name = '{0}-{1:06d}'.format(
                        _orig_name,
                        random.randint(0, 999999))
                    logger.debug('Got 409: trying to rename dataset to {0}'
                                 .format(dataset.name))
                else:
                    return result
        for source_id in differences['right']:
            logger.info('Creating dataset {0}'.format(source_id))
            dataset = source_datasets[source_id]
            force_dataset_operation(self._client.create_dataset, dataset)
            report_progress(('datasets', 'create'),
                            _prog_next_add(), _prog_tot_add)
        for source_id in differences['differing']:
            logger.info('Updating dataset {0}'.format(source_id))
            old_dataset = ckan_datasets[source_id]
            new_dataset = source_datasets[source_id]
            dataset = self._merge_datasets(old_dataset, new_dataset)
            dataset.id = old_dataset.id  
            self._client.update_dataset(dataset)  
            report_progress(('datasets', 'update'),
                            _prog_next_update(), _prog_tot_update)
def sync ( self , source_name , data ) : groups = dict ( ( key , CkanGroup ( val ) ) for key , val in data [ 'group' ] . iteritems ( ) ) organizations = dict ( ( key , CkanOrganization ( val ) ) for key , val in data [ 'organization' ] . iteritems ( ) ) groups_map = self . _upsert_groups ( groups ) orgs_map = self . _upsert_organizations ( organizations ) logger . info ( 'Creating list of datasets to be synchronized' ) source_datasets = { } for source_id , dataset_dict in data [ 'dataset' ] . iteritems ( ) : _dataset_dict = copy . deepcopy ( dataset_dict ) _dataset_dict . pop ( 'id' , None ) _dataset_dict [ 'groups' ] = [ groups_map . to_ckan ( grp_id ) for grp_id in _dataset_dict [ 'groups' ] ] _dataset_dict [ 'owner_org' ] = orgs_map . to_ckan ( _dataset_dict [ 'owner_org' ] ) dataset = CkanDataset ( _dataset_dict ) dataset . extras [ HARVEST_SOURCE_ID_FIELD ] = self . _join_source_id ( source_name , source_id ) source_datasets [ source_id ] = dataset logger . info ( 'Retrieving current status from Ckan' ) ckan_datasets = self . _find_datasets_by_source ( source_name ) differences = self . _compare_collections ( ckan_datasets , source_datasets ) _prog_tot_add = len ( differences [ 'right' ] ) _prog_next_add = itertools . count ( 1 ) . next _prog_tot_remove = len ( differences [ 'left' ] ) _prog_next_remove = itertools . count ( 1 ) . next _prog_tot_update = len ( differences [ 'differing' ] ) _prog_next_update = itertools . count ( 1 ) . next report_progress ( ( 'datasets' , 'delete' ) , 0 , _prog_tot_remove ) report_progress ( ( 'datasets' , 'create' ) , 0 , _prog_tot_add ) report_progress ( ( 'datasets' , 'update' ) , 0 , _prog_tot_update ) for source_id in differences [ 'left' ] : ckan_id = ckan_datasets [ source_id ] . id logger . info ( 'Deleting dataset {0}' . format ( ckan_id ) ) self . _client . delete_dataset ( ckan_id ) report_progress ( ( 'datasets' , 'delete' ) , _prog_next_remove ( ) , _prog_tot_remove ) def force_dataset_operation ( operation , dataset , retry = 5 ) : _orig_name = dataset . name [ : 80 ] dataset . name = _orig_name while True : try : result = operation ( dataset ) except HTTPError , e : if e . status_code != 409 : raise retry -= 1 if retry < 0 : raise dataset . name = '{0}-{1:06d}' . format ( _orig_name , random . randint ( 0 , 999999 ) ) logger . debug ( 'Got 409: trying to rename dataset to {0}' . format ( dataset . name ) ) else : return result for source_id in differences [ 'right' ] : logger . info ( 'Creating dataset {0}' . format ( source_id ) ) dataset = source_datasets [ source_id ] force_dataset_operation ( self . _client . create_dataset , dataset ) report_progress ( ( 'datasets' , 'create' ) , _prog_next_add ( ) , _prog_tot_add ) for source_id in differences [ 'differing' ] : logger . info ( 'Updating dataset {0}' . format ( source_id ) ) old_dataset = ckan_datasets [ source_id ] new_dataset = source_datasets [ source_id ] dataset = self . _merge_datasets ( old_dataset , new_dataset ) dataset . id = old_dataset . id self . _client . update_dataset ( dataset ) report_progress ( ( 'datasets' , 'update' ) , _prog_next_update ( ) , _prog_tot_update )
Synchronize data from a source into Ckan.
Synchronize data from a source into Ckan .
"""
        Synchronize data from a source into Ckan.

        - datasets are matched by _harvest_source
        - groups and organizations are matched by name

        :param source_name:
            String identifying the source of the data. Used to build
            ids that will be used in further synchronizations.
        :param data:
            Data to be synchronized. Should be a dict (or dict-like)
            with top level keys coresponding to the object type,
            mapping to dictionaries of ``{'id': <object>}``.
        """ # Upsert groups and organizations # Create list of datasets to be synced # We need to make sure "source" datasets # don't have (otherwise misleading) ids # We need to update groups and organizations, # to map their name from the source into a # ckan id # We also want to add the "source id", used for further # synchronizations to find stuff # Retrieve list of datasets from Ckan # Compare collections to find differences # ------------------------------------------------------------ # We now need to create/update/delete datasets. # todo: we need to make sure dataset names are not # already used by another dataset. The only # way is to randomize resource names and hope # a 409 response indicates duplicate name.. # _progress_total = sum(len(differences[x]) #                       for x in ('left', 'right', 'differing')) # _progress_next = itertools.count(1).next # report_progress(0, _progress_total) # Create progress bars early.. # We delete first, in order to (possibly) deallocate # some already-used names.. # Maximum dataset name length is 100 characters # We trim it down to 80 just to be safe. # Note: we generally want to preserve the original name #       and there should *never* be problems with that #       when updating.. # Create missing datasets # Update outdated datasets # dataset = source_datasets[source_id] # Mandatory! # should never fail!
self
source_name
String identifying the source of the data. Used to build
ids that will be used in further synchronizations.
String identifying the source of the data . Used to build ids that will be used in further synchronizations .
data
Data to be synchronized. Should be a dict (or dict-like)
with top level keys coresponding to the object type,
mapping to dictionaries of ``{'id': }``.
Data to be synchronized . Should be a dict ( or dict - like ) with top level keys coresponding to the object type mapping to dictionaries of ` ` { ' id ' : } ` ` .