Ensure reference FS wraps any sync filesystems (#1755)

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

Author: moradology

fsspec/implementations/asyn_wrapper.py

@@ -45,26 +45,27 @@ class AsyncFileSystemWrapper(AsyncFileSystem):
     def __init__(self, sync_fs, *args, **kwargs):
         super().__init__(*args, **kwargs)
         self.asynchronous = True
-        self.fs = sync_fs
+        self.sync_fs = sync_fs
+        self.protocol = self.sync_fs.protocol
         self._wrap_all_sync_methods()
 
     @property
     def fsid(self):
-        return f"async_{self.fs.fsid}"
+        return f"async_{self.sync_fs.fsid}"
 
     def _wrap_all_sync_methods(self):
         """
         Wrap all synchronous methods of the underlying filesystem with asynchronous versions.
         """
-        for method_name in dir(self.fs):
+        for method_name in dir(self.sync_fs):
             if method_name.startswith("_"):
                 continue
 
-            attr = inspect.getattr_static(self.fs, method_name)
+            attr = inspect.getattr_static(self.sync_fs, method_name)
             if isinstance(attr, property):
                 continue
 
-            method = getattr(self.fs, method_name)
+            method = getattr(self.sync_fs, method_name)
             if callable(method) and not asyncio.iscoroutinefunction(method):
                 async_method = async_wrapper(method, obj=self)
                 setattr(self, f"_{method_name}", async_method)

fsspec/implementations/reference.py

@@ -20,6 +20,7 @@
 from fsspec.asyn import AsyncFileSystem
 from fsspec.callbacks import DEFAULT_CALLBACK
 from fsspec.core import filesystem, open, split_protocol
+from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
 from fsspec.utils import isfilelike, merge_offset_ranges, other_paths
 
 logger = logging.getLogger("fsspec.reference")
@@ -757,6 +758,10 @@ def __init__(
             self.fss[remote_protocol] = fs
 
         self.fss[None] = fs or filesystem("file")  # default one
+        # Wrap any non-async filesystems to ensure async methods are available below
+        for k, f in self.fss.items():
+            if not f.async_impl:
+                self.fss[k] = AsyncFileSystemWrapper(f)
 
     def _cat_common(self, path, start=None, end=None):
         path = self._strip_protocol(path)

fsspec/implementations/tests/test_reference.py

@@ -458,7 +458,8 @@ def test_fss_has_defaults(m):
     assert fs.fss["memory"].protocol == "memory"
 
     fs = fsspec.filesystem("reference", fs=m, fo={})
-    assert fs.fss[None] is m
+    # Default behavior here wraps synchronous filesystems to enable the async API
+    assert fs.fss[None].sync_fs is m
 
     fs = fsspec.filesystem("reference", fs={"memory": m}, fo={})
     assert fs.fss["memory"] is m