Make fsspec_open_kwargs, query_string_secrets, & is_opendap attributes of FilePattern

pangeo_forge_recipes/recipes/xarray_zarr.py

@@ -156,6 +156,7 @@ def cache_input(
     delete_input_encoding: bool,
     process_input: Optional[Callable[[xr.Dataset, str], xr.Dataset]],
     metadata_cache: Optional[MetadataTarget],
+    query_string_secrets: Optional[str],
     is_opendap=bool,
 ) -> None:
     if cache_inputs:
@@ -165,7 +166,7 @@ def cache_input(
             raise ValueError("input_cache is not set.")
         logger.info(f"Caching input '{input_key!s}'")
         fname = file_pattern[input_key]
-        input_cache.cache_file(fname, **fsspec_open_kwargs)
+        input_cache.cache_file(fname, query_string_secrets, **fsspec_open_kwargs)
 
     if cache_metadata:
         return cache_input_metadata(
@@ -703,6 +704,7 @@ class XarrayZarrRecipe(BaseRecipe):
     lock_timeout: Optional[int] = None
     subset_inputs: SubsetSpec = field(default_factory=dict)
     is_opendap: bool = False
+    query_string_secrets: Optional[str] = None
 
     # internal attributes not meant to be seen or accessed by user
     _concat_dim: str = field(default_factory=str, repr=False, init=False)
@@ -846,6 +848,7 @@ def cache_input(self) -> Callable[[Hashable], None]:
             process_input=self.process_input,
             metadata_cache=self.metadata_cache,
             is_opendap=self.is_opendap,
+            query_string_secrets=self.query_string_secrets,
         )
 
     @property

pangeo_forge_recipes/storage.py

@@ -147,7 +147,7 @@ def _full_path(self, path: str) -> str:
 class CacheFSSpecTarget(FlatFSSpecTarget):
     """Alias for FlatFSSpecTarget"""
 
-    def cache_file(self, fname: str, **open_kwargs) -> None:
+    def cache_file(self, fname: str, secrets: Optional[str], **open_kwargs) -> None:
         # check and see if the file already exists in the cache
         logger.info(f"Caching file '{fname}'")
         if self.exists(fname):
@@ -158,7 +158,8 @@ def cache_file(self, fname: str, **open_kwargs) -> None:
                 logger.info(f"File '{fname}' is already cached")
                 return
 
-        input_opener = fsspec.open(fname, mode="rb", **open_kwargs)
+        input_fname = fname if not secrets else _add_query_string_secrets(fname, secrets)
+        input_opener = fsspec.open(input_fname, mode="rb", **open_kwargs)
         target_opener = self.open(fname, mode="wb")
         logger.info(f"Coping remote file '{fname}' to cache")
         _copy_btw_filesystems(input_opener, target_opener)
@@ -240,3 +241,7 @@ def _slugify(value: str) -> str:
     value = unicodedata.normalize("NFKD", value).encode("ascii", "ignore").decode("ascii")
     value = re.sub(r"[^.\w\s-]+", "_", value.lower())
     return re.sub(r"[-\s]+", "-", value).strip("-_")
+
+
+def _add_query_string_secrets(fname: str, secrets: str) -> str:
+    return fname + secrets

tests/conftest.py

@@ -160,6 +160,13 @@ def teardown():
     return all_urls, items_per_file
 
 
+@pytest.fixture(scope="session")
+def netcdf_http_paths_with_secrets(netcdf_http_paths, fake_secrets):
+    all_urls, items_per_file = netcdf_http_paths
+    all_urls = [url + fake_secrets for url in all_urls]
+    return all_urls, items_per_file
+
+
 @pytest.fixture()
 def tmp_target(tmpdir_factory):
     fs = fsspec.get_filesystem_class("file")()
@@ -175,6 +182,11 @@ def tmp_cache(tmpdir_factory):
     return cache
 
 
+@pytest.fixture(scope="session")
+def fake_secrets():
+    return "?a-pretend-api-token"
+
+
 @pytest.fixture()
 def tmp_metadata_target(tmpdir_factory):
     path = str(tmpdir_factory.mktemp("cache"))

tests/test_storage.py

@@ -40,25 +40,41 @@ def test_metadata_target(tmp_metadata_target):
 
 
 @pytest.mark.parametrize(
-    "file_paths", [lazy_fixture("netcdf_local_paths"), lazy_fixture("netcdf_http_paths")]
+    "file_paths",
+    [
+        lazy_fixture("netcdf_local_paths"),
+        lazy_fixture("netcdf_http_paths"),
+        lazy_fixture("netcdf_http_paths_with_secrets"),
+    ],
 )
 @pytest.mark.parametrize("copy_to_local", [False, True])
 @pytest.mark.parametrize("use_cache, cache_first", [(False, False), (True, False), (True, True)])
 @pytest.mark.parametrize("use_dask", [True, False])
 @pytest.mark.parametrize("use_xarray", [True, False])
+@pytest.mark.parametrize("use_query_string_secrets", [True, False])
 def test_file_opener(
-    file_paths, tmp_cache, copy_to_local, use_cache, cache_first, dask_cluster, use_dask, use_xarray
+    file_paths,
+    tmp_cache,
+    fake_secrets,
+    copy_to_local,
+    use_cache,
+    cache_first,
+    dask_cluster,
+    use_dask,
+    use_xarray,
+    use_query_string_secrets,
 ):
     all_paths, _ = file_paths
     path = str(all_paths[0])
     cache = tmp_cache if use_cache else None
+    secrets = fake_secrets if use_query_string_secrets and file_paths[0][-1] == "?" else None
 
     def do_actual_test():
         if cache_first:
-            cache.cache_file(path)
+            cache.cache_file(path, secrets)
             assert cache.exists(path)
             details = cache.fs.ls(cache.root_path, detail=True)
-            cache.cache_file(path)
+            cache.cache_file(path, secrets)
             # check that nothing happened
             assert cache.fs.ls(cache.root_path, detail=True) == details