MAST: get_product_urls for Observations

CHANGES.rst

@@ -190,6 +190,7 @@ mast
 - Added ``pass_id`` as an alias for the ``pass`` column in query functions for the Roman mission to avoid conflicts with
   the reserved Python keyword. [#3588]
 - Update the cutout format request parameter in ``Zcut.download_cutouts`` to reflect a recent service change. [#3608]
+- Introduces ``Observations.get_product_urls``, which returns a list of URLs for products without downloading them. [#3639]
 
 
 jplspec

astroquery/mast/observations.py

@@ -935,6 +935,69 @@ def _download_curl_script(self, products, out_dir, verbose=True):
                           'Message': [msg]})
         return manifest
 
+    def _build_products_list(self, products, mrp_only=None, **filters):
+        """
+        Builds a list of data products.
+
+        Parameters
+        ----------
+        products : str, list, `~astropy.table.Table`
+            Either a single or list of obsids (as can be given to `get_product_list`),
+            a Table of products (as is returned by `get_product_list`), or a single or list of
+            URIs.
+        mrp_only : bool, optional
+            Default False. When set to true only "Minimum Recommended Products" will be returned.
+        **filters :
+            Filters to be applied.  Valid filters are all products fields returned by
+            ``get_metadata("products")`` and 'extension' which is the desired file extension.
+            The Column Name (or 'extension') is the keyword, with the argument being one or
+            more acceptable values for that parameter.
+            Filter behavior is AND between the filters and OR within a filter set.
+            For example: productType="SCIENCE",extension=["fits","jpg"]
+
+        Returns
+        -------
+        response : (Table, list[str])
+            Products table and products URIs list associated with the requested products.
+        """
+        # If the products list is a row we need to cast it as a table
+        if isinstance(products, Row):
+            products = Table(products, masked=True)
+
+        # If the products list is not already a table of products we need to
+        # get the products and filter them appropriately
+        if not isinstance(products, Table):
+
+            # make every str a list
+            if isinstance(products, str):
+                products = [products]
+
+            # every mast URI stays a list
+            if all(str(item).startswith("mast:") for item in products):
+                if mrp_only or filters:
+                    warnings.warn(
+                        "Filtering is not supported when providing MAST URIs. "
+                        "To apply filters, provide a product table, row, or obsid.",
+                        InputWarning,
+                    )
+            else:
+                # collect list of products
+                product_lists = []
+                for oid in products:
+                    product_lists.append(self.get_product_list(oid))
+
+                products = vstack(product_lists)
+
+        # for non-URI lists, filter the products
+        if isinstance(products, Table):
+            # apply filters
+            products = self.filter_products(products, mrp_only=mrp_only, **filters)
+
+        # remove duplicate products
+        products = utils.remove_duplicate_products(products, 'dataURI')
+
+        return products
+
     def download_products(self, products, *, download_dir=None, flat=False,
                           cache=True, curl_flag=False, mrp_only=False, cloud_only=False, verbose=True,
                           **filters):
@@ -987,29 +1050,8 @@ def download_products(self, products, *, download_dir=None, flat=False,
         # Ensure cloud access is enabled
         self._ensure_cloud_access()
 
-        # If the products list is a row we need to cast it as a table
-        if isinstance(products, Row):
-            products = Table(products, masked=True)
-
-        # If the products list is not already a table of products we need to
-        # get the products and filter them appropriately
-        if not isinstance(products, Table):
-
-            if isinstance(products, str):
-                products = [products]
-
-            # collect list of products
-            product_lists = []
-            for oid in products:
-                product_lists.append(self.get_product_list(oid))
-
-            products = vstack(product_lists)
-
-        # apply filters
-        products = self.filter_products(products, mrp_only=mrp_only, **filters)
-
-        # remove duplicate products
-        products = utils.remove_duplicate_products(products, 'dataURI')
+        # Get the filtered table of products
+        products = self._build_products_list(products, mrp_only=mrp_only, **filters)
 
         if not len(products):
             warnings.warn("No products to download.", NoResultsWarning)
@@ -1039,6 +1081,95 @@ def download_products(self, products, *, download_dir=None, flat=False,
 
         return manifest
 
+    def get_product_urls(self, products, *, mrp_only=False, cloud_only=False, include_bucket=True,
+                         full_url=False, verbose=True, **filters):
+        """
+        Get URLs for data products.
+        If cloud access is enabled, cloud locations will be returned. By default these are S3
+        URIs. If ``full_url`` is True, downloadable URLs are returned instead.
+
+        Parameters
+        ----------
+        products : str, list, `~astropy.table.Table`
+            Either a single or list of obsids (as can be given to `get_product_list`),
+            a Table of products (as is returned by `get_product_list`), or a single or list of
+            URIs.
+        mrp_only : bool, optional
+            Default False. When set to True only "Minimum Recommended Products" will be returned.
+        cloud_only : bool, optional
+            Default False. If set to True and cloud data access is enabled (see `enable_cloud_dataset`)
+            files that are not found in the cloud will be skipped rather than returned in the list
+            as is the default behavior. If cloud access is not enabled this argument has no effect.
+        include_bucket : bool
+            Default True. When False, returns the path of the file relative to the
+            top level cloud storage location.
+            Must be set to False when using the full_url argument.
+        full_url : bool
+            Default False. Return an HTTP fetchable url instead of a cloud uri.
+            Must set include_bucket to False to use this option.
+        verbose : bool, optional
+            Default True. Whether to show download progress in the console.
+        **filters :
+            Filters to be applied.  Valid filters are all products fields returned by
+            ``get_metadata("products")`` and 'extension' which is the desired file extension.
+            The Column Name (or 'extension') is the keyword, with the argument being one or
+            more acceptable values for that parameter.
+            Filter behavior is AND between the filters and OR within a filter set.
+            For example: productType="SCIENCE",extension=["fits","jpg"]
+
+        Returns
+        -------
+        response : list[str]
+            URLs or cloud URIs associated with the requested products.
+        """
+        # Ensure cloud access is enabled
+        cloud_enabled = self._ensure_cloud_access()
+
+        # Check cloud_enabled for cloud_only queries
+        if cloud_only and not cloud_enabled:
+            warnings.warn("`cloud_only` is True but cloud data access is not enabled. "
+                          "Falling back to MAST urls.", InputWarning)
+
+        if full_url and include_bucket:
+            raise InvalidQueryError(
+                "`include_bucket` must be False when `full_url=True`."
+            )
+
+        # Get the filtered products table or URI list
+        products = self._build_products_list(products, mrp_only=mrp_only, **filters)
+
+        if not len(products):
+            warnings.warn("No products to return urls for.", NoResultsWarning)
+            return
+
+        if isinstance(products, Table):
+            # parse out the uris and get cloud uris if cloud enabled
+            uri_list = [url for url in products['dataURI']]
+        else:
+            uri_list = products
+
+        base_url = self._portal_api_connection.MAST_DOWNLOAD_URL
+        url_list = [f"{base_url}?uri={quote(uri, safe=':/')}" for uri in uri_list]
+
+        if cloud_enabled:
+            cloud_uris = self._cloud_connection.get_cloud_uri_list(
+                uri_list,
+                include_bucket=include_bucket,
+                full_url=full_url,
+                verbose=verbose
+            )
+
+            # remove None values if cloud only, otherwise zip the two lists
+            if cloud_only:
+                url_list = [uri for uri in cloud_uris if uri is not None]
+            else:
+                url_list = [
+                    cloud_uri if cloud_uri is not None
+                    else url for cloud_uri, url in zip(cloud_uris, url_list)
+                ]
+
+        return url_list
+
     def list_cloud_datasets(self):
         """
         Returns a list of datasets that support cloud data access. Datasets are the prefixes

astroquery/mast/tests/test_mast.py

@@ -1373,6 +1373,185 @@ def test_observations_disable_cloud_dataset(patch_boto3):
     assert Observations._cloud_enabled_explicitly is False
 
 
+def test_observations_build_products_list(patch_boto3):
+    obsid = '2003738726'
+    data_uri = 'mast:HST/product/u9o40504m_c3m.fits'
+
+    # checking invalid arg combos
+    with pytest.warns(InputWarning, match="Filtering is not supported"):
+        products = Observations._build_products_list([data_uri], extension="png")
+    assert isinstance(products, list)
+    assert len(products) == 1
+
+    with pytest.warns(InputWarning, match="Filtering is not supported"):
+        products = Observations._build_products_list([data_uri], mrp=True)
+    assert isinstance(products, list)
+    assert len(products) == 1
+
+    # obsid input
+    products = Observations._build_products_list(obsid)
+    assert isinstance(products, Table)
+    assert all(products["obsID"] == obsid)
+
+    # list of obsid input
+    products = Observations._build_products_list([obsid])
+    assert isinstance(products, Table)
+    assert all(products["obsID"] == obsid)
+
+    # row input
+    product = Table()
+    product['dataURI'] = [data_uri]
+    products = Observations._build_products_list(product[0])
+    assert isinstance(products, Table)
+    assert all(product["dataURI"] == data_uri)
+
+    # table input
+    product = Table()
+    product['dataURI'] = [data_uri]
+    products = Observations._build_products_list(product)
+    assert isinstance(products, Table)
+    assert all(products["dataURI"] == data_uri)
+
+    # URI input
+    products = Observations._build_products_list(data_uri)
+    assert isinstance(products, list)
+    assert len(products) == 1
+    assert products[0] == data_uri
+
+    # URI list input
+    products = Observations._build_products_list([data_uri])
+    assert isinstance(products, list)
+    assert len(products) == 1
+    assert products[0] == data_uri
+
+
+def test_observations_get_product_urls(patch_boto3):
+    obsid = '2003738726'
+    data_uri = 'mast:HST/product/u9o40504m_c3m.fits'
+    expected = 's3://stpubdata/hst/public/u9o4/u9o40504m/u9o40504m_c3m.fits'
+
+    # row input
+    product = Table()
+    product['dataURI'] = [data_uri]
+    url_list = Observations.get_product_urls(product[0])
+    assert isinstance(url_list, list)
+    assert len(url_list) == 1
+    assert url_list[0] == expected
+
+    # table input
+    product = Table()
+    product['dataURI'] = [data_uri]
+    url_list = Observations.get_product_urls(product)
+    assert isinstance(url_list, list)
+    assert len(url_list) == 1
+    assert url_list[0] == expected
+
+    # URI input
+    url_list = Observations.get_product_urls(data_uri)
+    assert isinstance(url_list, list)
+    assert len(url_list) == 1
+    assert url_list[0] == expected
+
+    # URI list input
+    url_list = Observations.get_product_urls([data_uri])
+    assert isinstance(url_list, list)
+    assert len(url_list) == 1
+    assert url_list[0] == expected
+
+    # disable cloud
+    Observations.disable_cloud_dataset()
+
+    # obsid input
+    url_list = Observations.get_product_urls(obsid)
+    assert isinstance(url_list, list)
+
+    # list of obsid input
+    url_list = Observations.get_product_urls([obsid])
+    assert isinstance(url_list, list)
+
+
+def test_observations_get_product_urls_invalid(patch_boto3):
+    obsid = '2003738726'
+    data_uri = 'mast:HST/product/u9o40504m_c3m.fits'
+    expected = 's3://stpubdata/hst/public/u9o4/u9o40504m/u9o40504m_c3m.fits'
+
+    # explicitly enable cloud
+    Observations.enable_cloud_dataset()
+
+    # check invalid arg combo
+    with pytest.raises(InvalidQueryError,
+                       match="`include_bucket` must be False"):
+        Observations.get_product_urls(
+            obsid,
+            include_bucket=True,
+            full_url=True
+        )
+
+    with pytest.warns(InputWarning, match="Filtering is not supported"):
+        url_list = Observations.get_product_urls([data_uri], extension="png")
+    assert isinstance(url_list, list)
+    assert len(url_list) == 1
+    assert url_list[0] == expected
+
+    with pytest.warns(InputWarning, match="Filtering is not supported"):
+        url_list = Observations.get_product_urls([data_uri], mrp=True)
+    assert isinstance(url_list, list)
+    assert len(url_list) == 1
+    assert url_list[0] == expected
+
+    # check warning for cloud_only but cloud_disabled
+    Observations.disable_cloud_dataset()
+
+    with pytest.warns(InputWarning, match="cloud data access is not enabled"):
+        result = Observations.get_product_urls(data_uri, cloud_only=True)
+    assert isinstance(result, list)
+    assert len(result) > 0
+
+
+def test_observations_get_product_urls_cloud_fallback(patch_boto3):
+    Observations.enable_cloud_dataset()
+    products = Table({
+        "dataURI": [
+            "mast:JWST/product/test1.fits",
+            "mast:HST/product/test2.fits",
+        ]
+    })
+    test_s3_uri = "s3://bucket/test1.fits"
+
+    # check fallback to mast urls
+    with patch.object(
+        Observations._cloud_connection,
+        "get_cloud_uri_list",
+        return_value=[
+            "s3://bucket/test1.fits",
+            None,
+        ],
+    ):
+        result = Observations.get_product_urls(products)
+        assert len(result) == 2
+        assert result[0] == test_s3_uri
+        assert result[1].startswith(
+            Observations._portal_api_connection.MAST_DOWNLOAD_URL
+        )
+
+        # checking cloud_only
+        result = Observations.get_product_urls(products, cloud_only=True)
+        assert len(result) == 1
+        assert result[0] == test_s3_uri
+
+        # check removal of duplicates
+        products = Table({
+            "dataURI": [
+                "mast:JWST/product/test1.fits",
+                "mast:HST/product/test2.fits",
+                "mast:JWST/product/test1.fits",
+            ]
+        })
+        result = Observations.get_product_urls(products)
+        assert len(result) == 2
+        assert result[0] == test_s3_uri
+
+
 ######################
 # CatalogClass tests #
 ######################