Merge pull request #1 from ogdch/62-pagination-rdf-harvester

62 pagination rdf harvester

Author: stefina

ckanext/dcat/harvesters/base.py

@@ -96,17 +96,17 @@ def _get_content_and_type(self, url, harvest_job, page=1, content_type=None):
                 # We want to catch these ones later on
                 raise
 
-            msg = 'Could not get content. Server responded with %s %s' % (
-                error.response.status_code, error.response.reason)
+            msg = 'Could not get content from %s. Server responded with %s %s' % (
+                url, error.response.status_code, error.response.reason)
             self._save_gather_error(msg, harvest_job)
             return None, None
         except requests.exceptions.ConnectionError, error:
-            msg = '''Could not get content because a
-                                connection error occurred. %s''' % error
+            msg = '''Could not get content from %s because a
+                                connection error occurred. %s''' % (url, error)
             self._save_gather_error(msg, harvest_job)
             return None, None
         except requests.exceptions.Timeout, error:
-            msg = 'Could not get content because the connection timed out.'
+            msg = 'Could not get content from %s because the connection timed out.' % url
             self._save_gather_error(msg, harvest_job)
             return None, None
 

ckanext/dcat/harvesters/rdf.py

@@ -148,71 +148,77 @@ def gather_stage(self, harvest_job):
 
         log.debug('In DCATRDFHarvester gather_stage')
 
-        # Get file contents
-        url = harvest_job.source.url
+        rdf_format = None
+        if harvest_job.source.config:
+            rdf_format = json.loads(harvest_job.source.config).get("rdf_format")
 
-        for harvester in p.PluginImplementations(IDCATRDFHarvester):
-            url, before_download_errors = harvester.before_download(url, harvest_job)
+        # Get file contents of first page
+        next_page_url = harvest_job.source.url
 
-            for error_msg in before_download_errors:
-                self._save_gather_error(error_msg, harvest_job)
+        guids_in_source = []
+        object_ids = []
 
-            if not url:
-                return False
+        while next_page_url:
+            for harvester in p.PluginImplementations(IDCATRDFHarvester):
+                next_page_url, before_download_errors = harvester.before_download(next_page_url, harvest_job)
 
-        rdf_format = None
-        if harvest_job.source.config:
-            rdf_format = json.loads(harvest_job.source.config).get("rdf_format")
-        content, rdf_format = self._get_content_and_type(url, harvest_job, 1, content_type=rdf_format)
+                for error_msg in before_download_errors:
+                    self._save_gather_error(error_msg, harvest_job)
 
-        # TODO: store content?
-        for harvester in p.PluginImplementations(IDCATRDFHarvester):
-            content, after_download_errors = harvester.after_download(content, harvest_job)
+                if not next_page_url:
+                    return []
 
-            for error_msg in after_download_errors:
-                self._save_gather_error(error_msg, harvest_job)
+            content, rdf_format = self._get_content_and_type(next_page_url, harvest_job, 1, content_type=rdf_format)
 
-        if not content:
-            return False
+            # TODO: store content?
+            for harvester in p.PluginImplementations(IDCATRDFHarvester):
+                content, after_download_errors = harvester.after_download(content, harvest_job)
 
-        # TODO: profiles conf
-        parser = RDFParser()
+                for error_msg in after_download_errors:
+                    self._save_gather_error(error_msg, harvest_job)
 
-        try:
-            parser.parse(content, _format=rdf_format)
-        except RDFParserException, e:
-            self._save_gather_error('Error parsing the RDF file: {0}'.format(e), harvest_job)
-            return False
+            if not content:
+                return []
 
-        guids_in_source = []
-        object_ids = []
-        for dataset in parser.datasets():
-            if not dataset.get('name'):
-                dataset['name'] = self._gen_new_name(dataset['title'])
+            # TODO: profiles conf
+            parser = RDFParser()
 
-            # Unless already set by the parser, get the owner organization (if any)
-            # from the harvest source dataset
-            if not dataset.get('owner_org'):
-                source_dataset = model.Package.get(harvest_job.source.id)
-                if source_dataset.owner_org:
-                    dataset['owner_org'] = source_dataset.owner_org
+            try:
+                parser.parse(content, _format=rdf_format)
+            except RDFParserException, e:
+                self._save_gather_error('Error parsing the RDF file: {0}'.format(e), harvest_job)
+                return []
 
-            # Try to get a unique identifier for the harvested dataset
-            guid = self._get_guid(dataset)
+            for dataset in parser.datasets():
+                if not dataset.get('name'):
+                    dataset['name'] = self._gen_new_name(dataset['title'])
 
-            if not guid:
-                self._save_gather_error('Could not get a unique identifier for dataset: {0}'.format(dataset),
-                                        harvest_job)
-                continue
+                # Unless already set by the parser, get the owner organization (if any)
+                # from the harvest source dataset
+                if not dataset.get('owner_org'):
+                    source_dataset = model.Package.get(harvest_job.source.id)
+                    if source_dataset.owner_org:
+                        dataset['owner_org'] = source_dataset.owner_org
 
-            dataset['extras'].append({'key': 'guid', 'value': guid})
-            guids_in_source.append(guid)
+                # Try to get a unique identifier for the harvested dataset
+                guid = self._get_guid(dataset)
 
-            obj = HarvestObject(guid=guid, job=harvest_job,
-                                content=json.dumps(dataset))
+                if not guid:
+                    self._save_gather_error('Could not get a unique identifier for dataset: {0}'.format(dataset),
+                                            harvest_job)
+                    continue
 
-            obj.save()
-            object_ids.append(obj.id)
+                dataset['extras'].append({'key': 'guid', 'value': guid})
+                guids_in_source.append(guid)
+
+                obj = HarvestObject(guid=guid, job=harvest_job,
+                                    content=json.dumps(dataset))
+
+                obj.save()
+                object_ids.append(obj.id)
+
+            # get the next page
+            next_page_url = parser.next_page()
 
         # Check if some datasets need to be deleted
         object_ids_to_delete = self._mark_datasets_for_deletion(guids_in_source, harvest_job)

ckanext/dcat/processors.py

@@ -114,6 +114,16 @@ def _datasets(self):
         for dataset in self.g.subjects(RDF.type, DCAT.Dataset):
             yield dataset
 
+    def next_page(self):
+        '''
+        Returns the URL of the next page or None if there is no next page
+        '''
+        for pagination_node in self.g.subjects(RDF.type, HYDRA.PagedCollection):
+            for o in self.g.objects(pagination_node, HYDRA.nextPage):
+                return unicode(o)
+        return None
+
+
     def parse(self, data, _format=None):
         '''
         Parses and RDF graph serialization and into the class graph

ckanext/dcat/tests/test_base_parser.py

@@ -139,6 +139,70 @@ def test_parse_data(self):
 
         eq_(len(p.g), 2)
 
+    def test_parse_pagination_next_page(self):
+
+        data = '''<?xml version="1.0" encoding="utf-8" ?>
+        <rdf:RDF
+         xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+         xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#"
+         xmlns:hydra="http://www.w3.org/ns/hydra/core#">
+         <hydra:PagedCollection rdf:about="http://example.com/catalog.xml?page=1">
+            <hydra:totalItems rdf:datatype="http://www.w3.org/2001/XMLSchema#integer">245</hydra:totalItems>
+            <hydra:lastPage>http://example.com/catalog.xml?page=3</hydra:lastPage>
+            <hydra:itemsPerPage rdf:datatype="http://www.w3.org/2001/XMLSchema#integer">100</hydra:itemsPerPage>
+            <hydra:nextPage>http://example.com/catalog.xml?page=2</hydra:nextPage>
+            <hydra:firstPage>http://example.com/catalog.xml?page=1</hydra:firstPage>
+        </hydra:PagedCollection>
+        </rdf:RDF>
+        '''
+
+        p = RDFParser()
+
+        p.parse(data)
+
+        eq_(p.next_page(), 'http://example.com/catalog.xml?page=2')
+
+    def test_parse_without_pagination(self):
+
+        data = '''<?xml version="1.0" encoding="utf-8" ?>
+        <rdf:RDF
+         xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+         xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#">
+        <rdfs:SomeClass rdf:about="http://example.org">
+            <rdfs:label>Some label</rdfs:label>
+        </rdfs:SomeClass>
+        </rdf:RDF>
+        '''
+
+        p = RDFParser()
+
+        p.parse(data)
+
+        eq_(p.next_page(), None)
+
+    def test_parse_pagination_last_page(self):
+
+        data = '''<?xml version="1.0" encoding="utf-8" ?>
+        <rdf:RDF
+         xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
+         xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#"
+         xmlns:hydra="http://www.w3.org/ns/hydra/core#">
+         <hydra:PagedCollection rdf:about="http://example.com/catalog.xml?page=3">
+            <hydra:totalItems rdf:datatype="http://www.w3.org/2001/XMLSchema#integer">245</hydra:totalItems>
+            <hydra:lastPage>http://example.com/catalog.xml?page=3</hydra:lastPage>
+            <hydra:itemsPerPage rdf:datatype="http://www.w3.org/2001/XMLSchema#integer">100</hydra:itemsPerPage>
+            <hydra:firstPage>http://example.com/catalog.xml?page=1</hydra:firstPage>
+            <hydra:previousPage>http://example.com/catalog.xml?page=2</hydra:previousPage>
+        </hydra:PagedCollection>
+        </rdf:RDF>
+        '''
+
+        p = RDFParser()
+
+        p.parse(data)
+
+        eq_(p.next_page(), None)
+
     def test_parse_data_different_format(self):
 
         data = '''