Wrap sync fs for xarray.to_zarr (#2533)

Co-authored-by: Martin Durant <[email protected]>

changes/2533.bigfix.rst

@@ -0,0 +1 @@
+Wrap sync fsspec filesystems with AsyncFileSystemWrapper in xarray.to_zarr

src/zarr/storage/_fsspec.py

@@ -172,6 +172,17 @@ def from_url(
         opts = {"asynchronous": True, **opts}
 
         fs, path = url_to_fs(url, **opts)
+        if not fs.async_impl:
+            try:
+                from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
+
+                fs = AsyncFileSystemWrapper(fs)
+            except ImportError as e:
+                raise ImportError(
+                    f"The filesystem for URL '{url}' is synchronous, and the required "
+                    "AsyncFileSystemWrapper is not available. Upgrade fsspec to version "
+                    "2024.12.0 or later to enable this functionality."
+                ) from e
 
         # 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

tests/test_store/test_fsspec.py

@@ -6,6 +6,7 @@
 
 import pytest
 from botocore.session import Session
+from packaging.version import parse as parse_version
 
 import zarr.api.asynchronous
 from zarr.abc.store import OffsetByteRequest
@@ -215,3 +216,31 @@ async def test_empty_nonexistent_path(self, store_kwargs) -> None:
         store_kwargs["path"] += "/abc"
         store = await self.store_cls.open(**store_kwargs)
         assert await store.is_empty("")
+
+
+@pytest.mark.skipif(
+    parse_version(fsspec.__version__) < parse_version("2024.12.0"),
+    reason="No AsyncFileSystemWrapper",
+)
+def test_wrap_sync_filesystem():
+    """The local fs is not async so we should expect it to be wrapped automatically"""
+    from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
+
+    store = FsspecStore.from_url("local://test/path")
+
+    assert isinstance(store.fs, AsyncFileSystemWrapper)
+    assert store.fs.async_impl
+
+
+@pytest.mark.skipif(
+    parse_version(fsspec.__version__) < parse_version("2024.12.0"),
+    reason="No AsyncFileSystemWrapper",
+)
+def test_no_wrap_async_filesystem():
+    """An async fs should not be wrapped automatically; fsspec's https filesystem is such an fs"""
+    from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
+
+    store = FsspecStore.from_url("https://test/path")
+
+    assert not isinstance(store.fs, AsyncFileSystemWrapper)
+    assert store.fs.async_impl