added column guesser based on utype and ucd, changed parse_json to static function

Author: d-giles

pyvo/dal/adhoc.py

@@ -6,6 +6,7 @@
 import warnings
 import copy
 import requests
+import json
 
 from .query import DALResults, DALQuery, DALService, Record
 from .exceptions import DALServiceError
@@ -396,51 +397,6 @@ def iter_datalinks(self, preserve_order=False):
             yield from self._iter_datalinks_from_dlblock(
                 preserve_order=preserve_order)
 
-    def iter_parse_json_params(
-        self,
-        json_key: str,
-        colname: str="cloud_access",
-        verbose: bool=False,
-        **match_params
-    ):
-        """
-        Iterate over all Records in a DalResult and return parsed json parameters.
-
-        Parameters
-        ----------
-        json_key : str
-            The primary key by which to filter JSON results.
-        colname : str, optional
-            The column containing JSON to be parsed, by default "cloud_access".
-        verbose : bool, optional
-            Whether to print progress and errors, by default False.
-        **match_params : str, optional
-            Further parameters on which to match beyond `json_key`.
-
-        Returns
-        -------
-        astropy.Table
-            A table containing the JSON parameters separated into columns, each
-            row corresponding to a matching JSON entry for each DataLinkRecord
-            for each row of the original DalResult.
-
-        """
-        for irow, record in enumerate(self):
-            access_points = record.parse_json_params(
-                colname=colname,
-                json_key=json_key,
-                verbose=verbose,
-                **match_params
-                )
-            access_points.add_column([irow]*len(access_points), name="record_row", index=0)
-            if irow == 0:
-                new_table = access_points
-            else:
-                for row in access_points.iterrows():
-                    new_table.add_row(row)
-
-        return new_table
-
     def iter_get_cloud_params(
         self,
         provider: str,
@@ -475,9 +431,10 @@ def iter_get_cloud_params(
 
             for jrow, row in enumerate(products):
                 # if no colname column, there is nothing to do    
-                try:
+                jsontxt = row._guess_cloud_column(colname=colname)
+                if jsontxt:
                     access_points = row.parse_json_params(
-                        colname=colname,
+                        json_txt=jsontxt,
                         json_key=provider,
                         verbose=verbose,
                         **match_params
@@ -488,15 +445,14 @@ def iter_get_cloud_params(
                     else:
                         for row in access_points.iterrows():
                             new_dl_table.add_row(row)
-                except KeyError:
+                else:
                     # no json column, continue
                     if verbose:
-                        print(f'No column {colname} found for row {irow}, datalink {jrow}')
+                        print(f'No column {colname} found for Results row {irow}, datalink row {jrow}')
                     new_dl_table = TableElement(VOTableFile()).to_table()
-                    continue
 
             # do the json parsing
-            cloud_params = access_points
+            cloud_params = new_dl_table
             cloud_params.add_column([irow]*len(cloud_params), name="record_row", index=0)
             if irow == 0:
                 new_table = cloud_params
@@ -554,21 +510,21 @@ def getdataset(self, timeout=None):
             # this should go to Record.getdataset()
             return super().getdataset(timeout=timeout)
 
+    @staticmethod
     def parse_json_params(
-        self,
+        json_txt: str,
         json_key: str,
-        colname: str="cloud_access",
         verbose: bool=False,
         **match_params
         ):
         """Parse information stored as JSON by key
         
         Parameters
         ----------
+        json_txt : str
+            Text interpreted as JSON
         json_key : str
             The primary key by which to filter JSON results
-        colname : str, optional
-            The column containing JSON to be parsed, by default "cloud_access"
         verbose : bool, optional
             Whether to print progress and errors, by default False
         **match_params : str, optional
@@ -581,43 +537,52 @@ def parse_json_params(
             row representing a matching JSON entry.
 
         """
-        import json
 
         # init results table (avoiding adding import of astropy.table.Table)
         new_table = TableElement(VOTableFile()).to_table()
-        
-        if verbose:
-            print(f'searching for and processing json column {colname}')
 
-        try:
-            jsontxt  = self[colname]
-            jsonDict = json.loads(jsontxt)
-            if json_key not in jsonDict and verbose:
-                print(f'No key "{json_key}" found for record'
-                        'in column "{colname}"')
-            else:
-                p_params = jsonDict[json_key]
-                checks = []
-                for k, value in match_params.items():
-                    checks.append(p_params.getitem(k, value) == value)
-
-                if all(checks):
-                    if not isinstance(p_params, list):
-                        p_params = [p_params]
-                    colnames = list(p_params[0].keys())
-                    colvals = [[] for _ in colnames]
-                    for ppar in p_params:
-                        for idx, val in enumerate(ppar.values()):
-                            colvals[idx].append(val)
-                    new_table.add_columns(cols=colvals, names=colnames)
-
-        except KeyError:
-            # no json column, return empty list
-            if verbose:
-                print(f'No column {colname} found for record.')
+        jsonDict = json.loads(json_txt)
+        if json_key not in jsonDict and verbose:
+            print(f'No key "{json_key}" found in json_txt given.')
+        else:
+            p_params = jsonDict[json_key]
+            checks = []
+            for k, value in match_params.items():
+                checks.append(p_params.getitem(k, value) == value)
+
+            if all(checks):
+                if not isinstance(p_params, list):
+                    p_params = [p_params]
+                colnames = list(p_params[0].keys())
+                colvals = [[] for _ in colnames]
+                for ppar in p_params:
+                    for idx, val in enumerate(ppar.values()):
+                        colvals[idx].append(val)
+                new_table.add_columns(cols=colvals, names=colnames)
 
         return new_table
 
+    def _guess_cloud_column(self, colname="cloud_access"):
+        """returns a guess for a URI to a data product in row.
+
+        This tries a few heuristics based on how cloud access or records might
+        be marked up.  This will return None if row does not look as if
+        it contained a cloud access column.
+        """
+        if hasattr(self, colname):
+            return getattr(self, colname)
+
+        if colname in self:
+            return self[colname]
+
+        cloud_access = self.getbyutype("adhoc:cloudstorage")
+        if cloud_access:
+            return cloud_access
+
+        cloud_access = self.getbyucd("meta.ref.cloudstorage") 
+        if cloud_access:
+            return cloud_access
+
     def get_cloud_params(
         self,
         provider: str,
@@ -651,10 +616,11 @@ def get_cloud_params(
 
         for irow, row in enumerate(products):
             # if no colname column, there is nothing to do    
-            try:
+            cloud_json = row._guess_cloud_column(colname=colname)
+            if cloud_json:
                 access_points = row.parse_json_params(
-                    colname=colname,
-                    key=provider,
+                    json_txt=cloud_json,
+                    json_key=provider,
                     verbose=verbose,
                     **match_params
                     )
@@ -664,12 +630,12 @@ def get_cloud_params(
                 else:
                     for row in access_points.iterrows():
                         new_table.add_row(row)
-            except KeyError:
-                # no json column, continue
+            else:
+                # no json column, return None
                 if verbose:
                     print(f'No column {colname} found for row {irow}')
-                new_table = TableElement(VOTableFile()).to_table()
-                continue
+                new_table = None
+                break
 
         return new_table
 

pyvo/dal/tests/test_datalink.py

@@ -77,18 +77,38 @@ def callback(request, context):
         yield matcher
 
 @pytest.fixture()
-def datalink_cloud(mocker):
+def datalink_cloud1(mocker):
     def callback(request, context): 
         dl_base = parse('pyvo/dal/tests/data/datalink/datalink.xml')
         dl_base_table = dl_base.get_first_table().to_table()
         cloud_access_str = '{"aws": {"bucket_name": "test", "key":"path/to/cloudfile.fits", "region": "us-west-2"}}'
         dl_base_table.add_column([cloud_access_str]*4, name='cloud_access')
         out = BytesIO()
-        writeto(from_table(dl_base_table), out)
+        votable = from_table(dl_base_table)
+        votable.get_first_table().get_field_by_id("cloud_access").utype = "adhoc:cloudstorage"
+        writeto(votable, out)
         return out.getvalue()
 
     with mocker.register_uri(
-        'GET', 'http://example.com/datalink-cloud.xml', content=callback
+        'GET', 'http://example.com/datalink-cloud1.xml', content=callback
+    ) as matcher:
+        yield matcher
+
+@pytest.fixture()
+def datalink_cloud2(mocker):
+    def callback(request, context): 
+        dl_base = parse('pyvo/dal/tests/data/datalink/datalink.xml')
+        dl_base_table = dl_base.get_first_table().to_table()
+        cloud_access_str = '{"aws": {"bucket_name": "test", "key":"path/to/cloudfile.fits", "region": "us-west-2"}}'
+        dl_base_table.add_column([cloud_access_str]*4, name='cloud_access')
+        out = BytesIO()
+        votable = from_table(dl_base_table)
+        votable.get_first_table().get_field_by_id("cloud_access").ucd = "meta.ref.cloudstorage"
+        writeto(votable, out)
+        return out.getvalue()
+
+    with mocker.register_uri(
+        'GET', 'http://example.com/datalink-cloud2.xml', content=callback
     ) as matcher:
         yield matcher
 
@@ -355,7 +375,7 @@ def test_no_datalink():
         result.getdatalink()
 
 @pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.E02")
-@pytest.mark.usefixtures('datalink_cloud')
+@pytest.mark.usefixtures('datalink_cloud1', 'datalink_cloud2')
 class TestJsonColumns:
     """Tests for producing datalinks from tables containing links to
     datalink documents.
@@ -366,41 +386,39 @@ class TestJsonColumns:
                 "ucd": "meta.ref.url"},
             {"name": "access_format", "datatype": "char", "arraysize": "*",
                 "utype": "meta.code.mime"},
-            {"name": "cloud_access", "datatype": "char", "arraysize": "*"},],
-            [("http://example.com/datalink-cloud.xml",
+            {"name": "cloud_access", "datatype": "char", "arraysize": "*",
+                "utype": "adhoc:cloudstorage", "ucd": "meta.ref.cloudstorage"},],
+            [("http://example.com/datalink-cloud1.xml",
                 "application/x-votable+xml;content=datalink",
                 '{"aws": {"bucket_name": "test", "key":"path/to/file1.fits", "region": "us-west-2"}}',),
-                ("http://example.com/datalink-cloud.xml",
+                ("http://example.com/datalink-cloud2.xml",
                 "application/x-votable+xml;content=datalink",
                 '{"aws": {"bucket_name": "test", "key":"path/to/file2.fits", "region": "us-west-2"}}',)],
             resultsClass=SIA2Results
         )
     def test_record_w_json(self):
 
-
-        parsed_json_matches = self.res[0].parse_json_params("cloud_access", "aws")
+        jsontxt = '{"aws": {"bucket_name": "test", "key":"path/to/file1.fits", "region": "us-west-2"}}'
+        parsed_json_matches = self.res[0].parse_json_params(json_txt=jsontxt, json_key="aws")
         assert parsed_json_matches[0]["bucket_name"] == "test"
         assert parsed_json_matches[0]["key"] == "path/to/file1.fits"
         assert parsed_json_matches[0]["region"] == "us-west-2"
 
-    def test_iter_json(self):
-
-        parsed_json_matches = self.res.iter_parse_json_params("cloud_access", "aws")
-        assert parsed_json_matches[0]["record_row"] == 0
-        assert parsed_json_matches[0]["bucket_name"] == "test"
-        assert parsed_json_matches[0]["key"] == "path/to/file1.fits"
-        assert parsed_json_matches[0]["region"] == "us-west-2"
-        assert parsed_json_matches[1]["record_row"] == 1
-        assert parsed_json_matches[1]["key"] == "path/to/file2.fits"
-
     def test_datalink_json(self):
-        parsed_cloud_params = self.res[0].get_cloud_params("cloud_access", "aws")
+        parsed_cloud_params = self.res[0].get_cloud_params(provider="aws", colname="cloud_access")
         assert parsed_cloud_params[0]["bucket_name"] == "test"
         assert parsed_cloud_params[0]["key"] == "path/to/cloudfile.fits"
         assert parsed_cloud_params[0]["region"] == "us-west-2"
 
+        parsed2 = self.res[0].get_cloud_params(provider="aws", colname="bad_col_name")
+        assert parsed_cloud_params == parsed2
+
+        parsed3 = self.res[1].get_cloud_params(provider="aws", colname="bad_col_name")
+        parsed4 = self.res[1].get_cloud_params(provider="aws", colname="cloud_access")
+        assert parsed3 == parsed4
+
     def test_iter_datalink_json(self):
-        parsed_json_matches = self.res.iter_get_cloud_params("cloud_access", "aws")
+        parsed_json_matches = self.res.iter_get_cloud_params(provider="aws", colname="cloud_access")
         assert parsed_json_matches[0]["record_row"] == 0
         assert parsed_json_matches[0]["datalink_row"] == 0
         assert parsed_json_matches[0]["bucket_name"] == "test"