DirFileSystem breaks open_async for any underling filesystem

[vovochka404]

DirFileSystem can wrap any async filesystem. But cause open_async method is not redirected to wrapped fs, it will always raise NotImplementedError.

As mentioned in #1411, cause all unimplemented methods in base abstract classes got no abstractmethod decorator, it's hard to see that u've missed implementation for some method.

[martindurant]

open_async is experimental anyway, and there is no requirement that filesystems implement it. This is the problem with abstractmethod - in this case, NotImplemented is the right answer, since it's just waiting for someone to implement it.

[vovochka404]

But DirFileSystem is a wrapper. It should pass original call (with modified path) to wrapped file system. NotImpementedError must be thrown by wrapped (underling) fs, not by dirfs.

[martindurant]

I agree that it should be implemented, yes - it just hasn't happened yet

[dhirschfeld]

I've got a DirFileSystem wrapping a LocalFileSystem and am seeing a NotImplementedError when trying to use write_bytes.

Posting here in case the underlying issue is the same. If not it can be moved to a new issue...

fs.write_bytes(filepath, blob)
Traceback (most recent call last):
  File "C:\python\envs\dev-py310\lib\site-packages\IPython\core\interactiveshell.py", line 3548, in run_code
    exec(code_obj, self.user_global_ns, self.user_ns)
  File "<ipython-input-3-cd365f2247d4>", line 1, in <module>
    fs.write_bytes(filepath, blob)
  File "C:\python\envs\dev-py310\lib\site-packages\fsspec\spec.py", line 1502, in write_bytes
    self.pipe_file(path, value, **kwargs)
  File "C:\python\envs\dev-py310\lib\site-packages\fsspec\asyn.py", line 118, in wrapper
    return sync(self.loop, func, *args, **kwargs)
  File "C:\python\envs\dev-py310\lib\site-packages\fsspec\asyn.py", line 103, in sync
    raise return_result
  File "C:\python\envs\dev-py310\lib\site-packages\fsspec\asyn.py", line 56, in _runner
    result[0] = await coro
  File "C:\python\envs\dev-py310\lib\site-packages\fsspec\asyn.py", line 393, in _pipe_file
    raise NotImplementedError
NotImplementedError

Repro:

In [3]: from fsspec.implementations.local import LocalFileSystem
   ...: from fsspec.implementations.dirfs import DirFileSystem

In [5]: fs = DirFileSystem(path='C:/temp', fs=LocalFileSystem())

In [6]: fs.write_bytes('/deleteme.bin', b"123")
---------------------------------------------------------------------------
NotImplementedError                       Traceback (most recent call last)
Cell In[6], line 1
----> 1 fs.write_bytes('/deleteme.bin', b"123")

File C:\python\envs\dev-py310\lib\site-packages\fsspec\spec.py:1502, in AbstractFileSystem.write_bytes(self, path, value, **kwargs)
   1500 def write_bytes(self, path, value, **kwargs):
   1501     """Alias of `AbstractFileSystem.pipe_file`."""
-> 1502     self.pipe_file(path, value, **kwargs)

File C:\python\envs\dev-py310\lib\site-packages\fsspec\asyn.py:118, in sync_wrapper.<locals>.wrapper(*args, **kwargs)
    115 @functools.wraps(func)
    116 def wrapper(*args, **kwargs):
    117     self = obj or args[0]
--> 118     return sync(self.loop, func, *args, **kwargs)

File C:\python\envs\dev-py310\lib\site-packages\fsspec\asyn.py:103, in sync(loop, func, timeout, *args, **kwargs)
    101     raise FSTimeoutError from return_result
    102 elif isinstance(return_result, BaseException):
--> 103     raise return_result
    104 else:
    105     return return_result

File C:\python\envs\dev-py310\lib\site-packages\fsspec\asyn.py:56, in _runner(event, coro, result, timeout)
     54     coro = asyncio.wait_for(coro, timeout=timeout)
     55 try:
---> 56     result[0] = await coro
     57 except Exception as ex:
     58     result[0] = ex

File C:\python\envs\dev-py310\lib\site-packages\fsspec\asyn.py:393, in AsyncFileSystem._pipe_file(self, path, value, **kwargs)
    392 async def _pipe_file(self, path, value, **kwargs):
--> 393     raise NotImplementedError

NotImplementedError: