disable cache in listing ops

Author: jhamman

src/zarr/storage/remote.py

@@ -96,7 +96,8 @@ def __init__(
                 f"fs ({fs}) was not created with `asynchronous=True`, this may lead to surprising behavior",
                 stacklevel=2,
             )
-        if "://" in path:
+        if "://" in path and not path.startswith("http"):
+            # `not path.startswith("http")` is a special case for the http filesystem (¯\_(ツ)_/¯)
             scheme, _ = path.split("://", maxsplit=1)
             raise ValueError(f"path argument to RemoteStore must not include scheme ({scheme}://)")
 
@@ -159,21 +160,22 @@ def from_url(
         RemoteStore
         """
         opts = storage_options or {}
-        opts = {"asynchronous": True, **opts}
+        opts = {"asynchronous": True, "use_listings_cache": False, **opts}
 
         fs, path = fsspec.url_to_fs(url, **opts)
 
         # fsspec is not consistent about removing the scheme from the path, so check and strip it here
         # https://github.com/fsspec/filesystem_spec/issues/1722
-        if "://" in path:
+        if "://" in path and not path.startswith("http"):
+            # `not path.startswith("http")` is a special case for the http filesystem (¯\_(ツ)_/¯)
             _, path = path.split("://", maxsplit=1)
 
         return cls(fs=fs, path=path, mode=mode, allowed_exceptions=allowed_exceptions)
 
     async def clear(self) -> None:
         # docstring inherited
         try:
-            for subpath in await self.fs._find(self.path, withdirs=True):
+            for subpath in await self.fs._find(self.path, withdirs=True, refresh=True):
                 if subpath != self.path:
                     await self.fs._rm(subpath, recursive=True)
         except FileNotFoundError:
@@ -185,7 +187,7 @@ async def empty(self) -> bool:
         # TODO: it would be nice if we didn't have to list all keys here
         # it should be possible to stop after the first key is discovered
         try:
-            return not await self.fs._ls(self.path)
+            return not await self.fs._ls(self.path, refresh=True)
         except FileNotFoundError:
             return True
 
@@ -319,15 +321,15 @@ async def set_partial_values(
 
     async def list(self) -> AsyncGenerator[str, None]:
         # docstring inherited
-        allfiles = await self.fs._find(self.path, detail=False, withdirs=False)
+        allfiles = await self.fs._find(self.path, detail=False, withdirs=False, refresh=True)
         for onefile in (a.replace(self.path + "/", "") for a in allfiles):
             yield onefile
 
     async def list_dir(self, prefix: str) -> AsyncGenerator[str, None]:
         # docstring inherited
         prefix = f"{self.path}/{prefix.rstrip('/')}"
         try:
-            allfiles = await self.fs._ls(prefix, detail=False)
+            allfiles = await self.fs._ls(prefix, detail=False, refresh=True)
         except FileNotFoundError:
             return
         for onefile in (a.replace(prefix + "/", "") for a in allfiles):
@@ -336,5 +338,7 @@ async def list_dir(self, prefix: str) -> AsyncGenerator[str, None]:
     async def list_prefix(self, prefix: str) -> AsyncGenerator[str, None]:
         # docstring inherited
         find_str = f"{self.path}/{prefix}"
-        for onefile in await self.fs._find(find_str, detail=False, maxdepth=None, withdirs=False):
+        for onefile in await self.fs._find(
+            find_str, detail=False, maxdepth=None, withdirs=False, refresh=True
+        ):
             yield onefile.removeprefix(find_str)