Make asynchronous optional in async_wrapper (#1794)

Author: martindurant

fsspec/implementations/asyn_wrapper.py

@@ -2,7 +2,7 @@
 import functools
 import inspect
 
-from fsspec.asyn import AsyncFileSystem
+from fsspec.asyn import AsyncFileSystem, running_async
 
 
 def async_wrapper(func, obj=None):
@@ -42,10 +42,14 @@ class AsyncFileSystemWrapper(AsyncFileSystem):
         The synchronous filesystem instance to wrap.
     """
 
-    def __init__(self, sync_fs, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.asynchronous = True
-        self.sync_fs = sync_fs
+    protocol = "async_wrapper"
+    cachable = False
+
+    def __init__(self, fs, *args, asynchronous=None, **kwargs):
+        if asynchronous is None:
+            asynchronous = running_async()
+        super().__init__(*args, asynchronous=asynchronous, **kwargs)
+        self.sync_fs = fs
         self.protocol = self.sync_fs.protocol
         self._wrap_all_sync_methods()
 

fsspec/implementations/dirfs.py

@@ -36,8 +36,6 @@ def __init__(
         super().__init__(**storage_options)
         if fs is None:
             fs = filesystem(protocol=target_protocol, **(target_options or {}))
-        if (path is not None) ^ (fo is not None) is False:
-            raise ValueError("Provide path or fo, not both")
         path = path or fo
 
         if self.asynchronous and not fs.async_impl:

fsspec/implementations/tests/test_asyn_wrapper.py

@@ -10,10 +10,14 @@
 from .test_local import csv_files, filetexts
 
 
-def test_is_async():
+@pytest.mark.asyncio
+async def test_is_async_default():
     fs = fsspec.filesystem("file")
     async_fs = AsyncFileSystemWrapper(fs)
     assert async_fs.async_impl
+    assert async_fs.asynchronous
+    async_fs = AsyncFileSystemWrapper(fs, asynchronous=False)
+    assert not async_fs.asynchronous
 
 
 def test_class_wrapper():
@@ -53,6 +57,7 @@ async def test_cats():
         assert result == b"a,b\n1,2\n"[1:-2]
 
         # test synchronous API is available as expected
+        async_fs = AsyncFileSystemWrapper(fs, asynchronous=False)
         result = async_fs.cat(".test.fakedata.1.csv", start=1, end=-2)
         assert result == b"a,b\n1,2\n"[1:-2]
 
@@ -142,3 +147,17 @@ async def test_batch_operations():
         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")
+
+
+def test_open(tmpdir):
+    fn = f"{tmpdir}/afile"
+    with open(fn, "wb") as f:
+        f.write(b"hello")
+    of = fsspec.open(
+        "dir://afile::async_wrapper::file",
+        mode="rb",
+        async_wrapper={"asynchronous": False},
+        dir={"path": str(tmpdir)},
+    )
+    with of as f:
+        assert f.read() == b"hello"

fsspec/implementations/tests/test_reference.py

@@ -515,10 +515,10 @@ def test_fss_has_defaults(m):
     assert fs.fss[None] is m
 
     fs = fsspec.filesystem("reference", fo={"key": ["memory://a"]})
-    assert fs.fss[None] is fs.fss["memory"]
+    assert fs.fss[None] == fs.fss["memory"]
 
     fs = fsspec.filesystem("reference", fo={"key": ["memory://a"], "blah": ["path"]})
-    assert fs.fss[None] is fs.fss["memory"]
+    assert fs.fss[None] == fs.fss["memory"]
 
 
 def test_merging(m):

fsspec/registry.py

@@ -72,6 +72,9 @@ def register_implementation(name, cls, clobber=False, errtxt=None):
         "class": "fsspec.implementations.arrow.HadoopFileSystem",
         "err": "pyarrow and local java libraries required for HDFS",
     },
+    "async_wrapper": {
+        "class": "morefs.asyn_wrapper.AsyncWrapperFileSystem",
+    },
     "asynclocal": {
         "class": "morefs.asyn_local.AsyncLocalFileSystem",
         "err": "Install 'morefs[asynclocalfs]' to use AsyncLocalFileSystem",