fix(asyn): support sync style implementation in async _rm_file

Author: Sbargaoui

fsspec/asyn.py

@@ -328,6 +328,11 @@ def loop(self):
         return self._loop
 
     async def _rm_file(self, path, **kwargs):
+        if (
+            inspect.iscoroutinefunction(self._rm)
+            and type(self)._rm is not AsyncFileSystem._rm
+        ):
+            return await self._rm(path, recursive=False, batch_size=1, **kwargs)
         raise NotImplementedError
 
     async def _rm(self, path, recursive=False, batch_size=None, **kwargs):

fsspec/implementations/tests/test_local.py

@@ -929,35 +929,37 @@ def test_strip_protocol_no_authority(uri, expected, cwd, current_drive):
 
 @pytest.mark.parametrize(
     "uri, expected",
-    [
-        ("file:/path", "/path"),
-        ("file:///path", "/path"),
-        ("file:////path", "//path"),
-        ("local:/path", "/path"),
-        ("s3://bucket/key", "{cwd}/s3://bucket/key"),
-        ("/path", "/path"),
-        ("file:///", "/"),
-    ]
-    if not WIN
-    else [
-        ("file:c:/path", "c:/path"),
-        ("file:/c:/path", "c:/path"),
-        ("file:/C:/path", "C:/path"),
-        ("file://c:/path", "c:/path"),
-        ("file:///c:/path", "c:/path"),
-        ("local:/path", "{current_drive}/path"),
-        ("s3://bucket/key", "{cwd}/s3://bucket/key"),
-        ("c:/path", "c:/path"),
-        ("c:\\path", "c:/path"),
-        ("file:///", "{current_drive}/"),
-        pytest.param(
-            "file://localhost/c:/path",
-            "c:/path",
-            marks=pytest.mark.xfail(
-                reason="rfc8089 section3 'localhost uri' not supported"
+    (
+        [
+            ("file:/path", "/path"),
+            ("file:///path", "/path"),
+            ("file:////path", "//path"),
+            ("local:/path", "/path"),
+            ("s3://bucket/key", "{cwd}/s3://bucket/key"),
+            ("/path", "/path"),
+            ("file:///", "/"),
+        ]
+        if not WIN
+        else [
+            ("file:c:/path", "c:/path"),
+            ("file:/c:/path", "c:/path"),
+            ("file:/C:/path", "C:/path"),
+            ("file://c:/path", "c:/path"),
+            ("file:///c:/path", "c:/path"),
+            ("local:/path", "{current_drive}/path"),
+            ("s3://bucket/key", "{cwd}/s3://bucket/key"),
+            ("c:/path", "c:/path"),
+            ("c:\\path", "c:/path"),
+            ("file:///", "{current_drive}/"),
+            pytest.param(
+                "file://localhost/c:/path",
+                "c:/path",
+                marks=pytest.mark.xfail(
+                    reason="rfc8089 section3 'localhost uri' not supported"
+                ),
             ),
-        ),
-    ],
+        ]
+    ),
 )
 def test_strip_protocol_absolute_paths(uri, expected, current_drive, cwd):
     expected = expected.format(current_drive=current_drive, cwd=cwd)

fsspec/tests/test_async.py

@@ -205,3 +205,42 @@ async def test_async_streamed_file_read():
         == b"foo-bar" * 20
     )
     await streamed_file.close()
+
+
+def test_rm_file_with_rm_implementation():
+    class AsyncFSWithRm(fsspec.asyn.AsyncFileSystem):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            self.removed_paths = []
+
+        async def _rm(self, path, recursive=False, batch_size=None, **kwargs):
+            if isinstance(path, str):
+                path = [path]
+            for p in path:
+                self.removed_paths.append(p)
+            return None
+
+    fs = AsyncFSWithRm()
+    fs.rm_file("test/file.txt")
+    assert "test/file.txt" in fs.removed_paths
+
+
+def test_rm_file_with_rm_file_implementation():
+    class AsyncFSWithRmFile(fsspec.asyn.AsyncFileSystem):
+        def __init__(self, **kwargs):
+            super().__init__(**kwargs)
+            self.removed_paths = []
+
+        async def _rm_file(self, path, **kwargs):
+            self.removed_paths.append(path)
+            return None
+
+    fs = AsyncFSWithRmFile()
+    fs.rm_file("test/file.txt")
+    assert "test/file.txt" in fs.removed_paths
+
+
+def test_rm_file_without_implementation():
+    fs = fsspec.asyn.AsyncFileSystem()
+    with pytest.raises(NotImplementedError):
+        fs.rm_file("test/file.txt")