Repository: fsspec/filesystem_spec
Pull request #1745 (2024-11-05 18:59:10)

Add async wrapper for sync FS
This PR adds support for arbitrarily wrapping synchronous filesystems to make them compatible with interfaces which might expect `AsynchronousFileSystem` instances. Both instance and class-level wrapping behavior is provided and tested. This is a feature that will be of use downstream to enable kerchunk/zarr-python3 compatibility and is related to the efforts [here](https://github.com/fsspec/kerchunk/issues/514)



martindurant: cf https://github.com/fsspec/kerchunk/pull/524
moradology: Hey @martindurant - would you prefer this (or some version thereof) implemented over in kerchunk rather than here?
martindurant: The only reason to have it in kerchunk, is to indicate a level of experimentation and allow for releasing whenever is convenient. I suppose ice/zarr has no problem depending on it. I don't think it really matters.
moradology: OK, digging into things a little bit, I think there's a decent argument for pushing this behavior upstream to `fsspec` itself for the sake of not requiring a `kerchunk` dependency in `zarr-python`. In particular, this method will need to have access to the wrapper: https://github.com/zarr-developers/zarr-python/blob/main/src/zarr/storage/remote.py#L159-L168

The alternative might be to provide it as a util in `zarr-python` itself, as `kerchunk` also depends on `zarr-python`.
![image](https://github.com/user-attachments/assets/a849d7eb-aa66-4f4c-8497-60b406b1c722)

martindurant: It can stay here, perhaps it'll come in useful for some other uses besides zarr.
moradology: I still need to get the docs updated here. Other than that, I think I think this should be good. I've been chasing other test failures for zarr3 integration, but will plan to get docs updated + CI fixed ready for potential merger tomorrow
martindurant: OK, thank you. Let me know.
moradology: KK, I think this is ready. Wasn't *entirely* sure how best to handle `fsid`, so that might deserve some critical attention but otherwise things look OK
martindurant: > how best to handle fsid

I think what you've done is fine.
martindurant: I did the lint, but please check [this](https://github.com/fsspec/filesystem_spec/pull/1745/commits/34074c17c677d66f49157c097b1c14a7786ecd13#diff-234fdfcf7b8bd75c3ec9902d38b77327ededfb9b0dc01aed46ffc12af6df62ffL25) - is there a chance that `self` gets lost?
moradology: Just looked things over again. It appears to me that line was kept in error and should be ditched

Patch:
diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index 653eb80..f9e5ed9 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -14,7 +14,6 @@ jobs:
       fail-fast: false
       matrix:
         PY:
-          - "3.8"
           - "3.9"
           - "3.10"
           - "3.11"
diff --git a/docs/source/async.rst b/docs/source/async.rst
index 58334b3..dc44df3 100644
--- a/docs/source/async.rst
+++ b/docs/source/async.rst
@@ -152,3 +152,37 @@ available as the attribute ``.loop``.
 
     <script data-goatcounter="https://fsspec.goatcounter.com/count"
         async src="//gc.zgo.at/count.js"></script>
+
+AsyncFileSystemWrapper
+----------------------
+
+The `AsyncFileSystemWrapper` class is an experimental feature that allows you to convert
+a synchronous filesystem into an asynchronous one. This is useful for quickly integrating
+synchronous filesystems into workflows that may expect `AsyncFileSystem` instances.
+
+Basic Usage
+~~~~~~~~~~~
+
+To use `AsyncFileSystemWrapper`, wrap any synchronous filesystem to work in an asynchronous context.
+In this example, the synchronous `LocalFileSystem` is wrapped, creating an `AsyncFileSystem` instance
+backed by the normal, synchronous methods of `LocalFileSystem`:
+
+.. code-block:: python
+
+    import asyncio
+    import fsspec
+    from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
+
+    async def async_copy_file():
+        sync_fs = fsspec.filesystem('file')  # by-default synchronous, local filesystem
+        async_fs = AsyncFileSystemWrapper(sync_fs)
+        return await async_fs._copy('/source/file.txt', '/destination/file.txt')
+
+    asyncio.run(async_copy_file())
+
+Limitations
+-----------
+
+This is experimental. Users should not expect this wrapper to magically make things faster.
+It is primarily provided to allow usage of synchronous filesystems with interfaces that expect
+`AsyncFileSystem` instances.
diff --git a/fsspec/implementations/asyn_wrapper.py b/fsspec/implementations/asyn_wrapper.py
new file mode 100644
index 0000000..9ba7811
--- /dev/null
+++ b/fsspec/implementations/asyn_wrapper.py
@@ -0,0 +1,96 @@
+import asyncio
+import inspect
+import functools
+from fsspec.asyn import AsyncFileSystem
+
+
+def async_wrapper(func, obj=None):
+    """
+    Wraps a synchronous function to make it awaitable.
+
+    Parameters
+    ----------
+    func : callable
+        The synchronous function to wrap.
+    obj : object, optional
+        The instance to bind the function to, if applicable.
+
+    Returns
+    -------
+    coroutine
+        An awaitable version of the function.
+    """
+
+    @functools.wraps(func)
+    async def wrapper(*args, **kwargs):
+        return await asyncio.to_thread(func, *args, **kwargs)
+
+    return wrapper
+
+
+class AsyncFileSystemWrapper(AsyncFileSystem):
+    """
+    A wrapper class to convert a synchronous filesystem into an asynchronous one.
+
+    This class takes an existing synchronous filesystem implementation and wraps all
+    its methods to provide an asynchronous interface.
+
+    Parameters
+    ----------
+    sync_fs : AbstractFileSystem
+        The synchronous filesystem instance to wrap.
+    """
+
+    def __init__(self, sync_fs, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.asynchronous = True
+        self.fs = sync_fs
+        self._wrap_all_sync_methods()
+
+    @property
+    def fsid(self):
+        return f"async_{self.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):
+            if method_name.startswith("_"):
+                continue
+
+            attr = inspect.getattr_static(self.fs, method_name)
+            if isinstance(attr, property):
+                continue
+
+            method = getattr(self.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)
+
+    @classmethod
+    def wrap_class(cls, sync_fs_class):
+        """
+        Create a new class that can be used to instantiate an AsyncFileSystemWrapper
+        with lazy instantiation of the underlying synchronous filesystem.
+
+        Parameters
+        ----------
+        sync_fs_class : type
+            The class of the synchronous filesystem to wrap.
+
+        Returns
+        -------
+        type
+            A new class that wraps the provided synchronous filesystem class.
+        """
+
+        class GeneratedAsyncFileSystemWrapper(cls):
+            def __init__(self, *args, **kwargs):
+                sync_fs = sync_fs_class(*args, **kwargs)
+                super().__init__(sync_fs)
+
+        GeneratedAsyncFileSystemWrapper.__name__ = (
+            f"Async{sync_fs_class.__name__}Wrapper"
+        )
+        return GeneratedAsyncFileSystemWrapper


Test patch:
diff --git a/fsspec/implementations/tests/test_asyn_wrapper.py b/fsspec/implementations/tests/test_asyn_wrapper.py
new file mode 100644
index 0000000..d292020
--- /dev/null
+++ b/fsspec/implementations/tests/test_asyn_wrapper.py
@@ -0,0 +1,142 @@
+import asyncio
+import pytest
+import os
+
+import fsspec
+from fsspec.implementations.asyn_wrapper import AsyncFileSystemWrapper
+from fsspec.implementations.local import LocalFileSystem
+from .test_local import csv_files, filetexts
+
+
+def test_is_async():
+    fs = fsspec.filesystem("file")
+    async_fs = AsyncFileSystemWrapper(fs)
+    assert async_fs.async_impl
+
+
+def test_class_wrapper():
+    fs_cls = LocalFileSystem
+    async_fs_cls = AsyncFileSystemWrapper.wrap_class(fs_cls)
+    assert async_fs_cls.__name__ == "AsyncLocalFileSystemWrapper"
+    async_fs = async_fs_cls()
+    assert async_fs.async_impl
+
+
+@pytest.mark.asyncio
+async def test_cats():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        result = await async_fs._cat(".test.fakedata.1.csv")
+        assert result == b"a,b\n1,2\n"
+
+        out = set(
+            (
+                await async_fs._cat([".test.fakedata.1.csv", ".test.fakedata.2.csv"])
+            ).values()
+        )
+        assert out == {b"a,b\n1,2\n", b"a,b\n3,4\n"}
+
+        result = await async_fs._cat(".test.fakedata.1.csv", None, None)
+        assert result == b"a,b\n1,2\n"
+
+        result = await async_fs._cat(".test.fakedata.1.csv", start=1, end=6)
+        assert result == b"a,b\n1,2\n"[1:6]
+
+        result = await async_fs._cat(".test.fakedata.1.csv", start=-1)
+        assert result == b"a,b\n1,2\n"[-1:]
+
+        result = await async_fs._cat(".test.fakedata.1.csv", start=1, end=-2)
+        assert result == b"a,b\n1,2\n"[1:-2]
+
+        # test synchronous API is available as expected
+        result = async_fs.cat(".test.fakedata.1.csv", start=1, end=-2)
+        assert result == b"a,b\n1,2\n"[1:-2]
+
+        out = set(
+            (
+                await async_fs._cat(
+                    [".test.fakedata.1.csv", ".test.fakedata.2.csv"], start=1, end=-1
+                )
+            ).values()
+        )
+        assert out == {b"a,b\n1,2\n"[1:-1], b"a,b\n3,4\n"[1:-1]}
+
+
+@pytest.mark.asyncio
+async def test_basic_crud_operations():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        await async_fs._touch(".test.fakedata.3.csv")
+        assert await async_fs._exists(".test.fakedata.3.csv")
+
+        data = await async_fs._cat(".test.fakedata.1.csv")
+        assert data == b"a,b\n1,2\n"
+
+        await async_fs._pipe(".test.fakedata.1.csv", b"a,b\n5,6\n")
+        data = await async_fs._cat(".test.fakedata.1.csv")
+        assert data == b"a,b\n5,6\n"
+
+        await async_fs._rm(".test.fakedata.1.csv")
+        assert not await async_fs._exists(".test.fakedata.1.csv")
+
+
+@pytest.mark.asyncio
+async def test_error_handling():
+    fs = fsspec.filesystem("file")
+    async_fs = AsyncFileSystemWrapper(fs)
+
+    with pytest.raises(FileNotFoundError):
+        await async_fs._cat(".test.non_existent.csv")
+
+    with pytest.raises(FileNotFoundError):
+        await async_fs._rm(".test.non_existent.csv")
+
+
+@pytest.mark.asyncio
+async def test_concurrent_operations():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        async def read_file(file_path):
+            return await async_fs._cat(file_path)
+
+        results = await asyncio.gather(
+            read_file(".test.fakedata.1.csv"),
+            read_file(".test.fakedata.2.csv"),
+            read_file(".test.fakedata.1.csv"),
+        )
+
+        assert results == [b"a,b\n1,2\n", b"a,b\n3,4\n", b"a,b\n1,2\n"]
+
+
+@pytest.mark.asyncio
+async def test_directory_operations():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        await async_fs._makedirs("new_directory")
+        assert await async_fs._isdir("new_directory")
+
+        files = await async_fs._ls(".")
+        filenames = [os.path.basename(file) for file in files]
+
+        assert ".test.fakedata.1.csv" in filenames
+        assert ".test.fakedata.2.csv" in filenames
+        assert "new_directory" in filenames
+
+
+@pytest.mark.asyncio
+async def test_batch_operations():
+    with filetexts(csv_files, mode="b"):
+        fs = fsspec.filesystem("file")
+        async_fs = AsyncFileSystemWrapper(fs)
+
+        await async_fs._rm([".test.fakedata.1.csv", ".test.fakedata.2.csv"])
+        assert not await async_fs._exists(".test.fakedata.1.csv")
+        assert not await async_fs._exists(".test.fakedata.2.csv")