Repository: fsspec/filesystem_spec

Issue:
`DirFileSystem.write_bytes` raises `NotImplementedError` 
I've got a `DirFileSystem` wrapping a `LocalFileSystem` and am seeing a `NotImplementedError` when trying to use `write_bytes`.

```python-traceback
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:
```python
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:
```

_Originally posted by @dhirschfeld in https://github.com/fsspec/filesystem_spec/issues/1414#issuecomment-1825166644_
            

Discussion:

Patch:
diff --git a/fsspec/implementations/dirfs.py b/fsspec/implementations/dirfs.py
index a3eac87..b1b3f6d 100644
--- a/fsspec/implementations/dirfs.py
+++ b/fsspec/implementations/dirfs.py
@@ -124,6 +124,12 @@ class DirFileSystem(AsyncFileSystem):
     def pipe(self, path, *args, **kwargs):
         return self.fs.pipe(self._join(path), *args, **kwargs)
 
+    async def _pipe_file(self, path, *args, **kwargs):
+        return await self.fs._pipe_file(self._join(path), *args, **kwargs)
+
+    def pipe_file(self, path, *args, **kwargs):
+        return self.fs.pipe_file(self._join(path), *args, **kwargs)
+
     async def _cat_file(self, path, *args, **kwargs):
         return await self.fs._cat_file(self._join(path), *args, **kwargs)
 


Test patch:
diff --git a/fsspec/implementations/tests/test_dirfs.py b/fsspec/implementations/tests/test_dirfs.py
index 98e7fc6..b2a7bbe 100644
--- a/fsspec/implementations/tests/test_dirfs.py
+++ b/fsspec/implementations/tests/test_dirfs.py
@@ -162,6 +162,17 @@ def test_pipe(dirfs):
     dirfs.fs.pipe.assert_called_once_with(f"{PATH}/file", *ARGS, **KWARGS)
 
 
+@pytest.mark.asyncio
+async def test_async_pipe_file(adirfs):
+    await adirfs._pipe_file("file", *ARGS, **KWARGS)
+    adirfs.fs._pipe_file.assert_called_once_with(f"{PATH}/file", *ARGS, **KWARGS)
+
+
+def test_pipe_file(dirfs):
+    dirfs.pipe_file("file", *ARGS, **KWARGS)
+    dirfs.fs.pipe_file.assert_called_once_with(f"{PATH}/file", *ARGS, **KWARGS)
+
+
 @pytest.mark.asyncio
 async def test_async_cat_file(adirfs):
     assert (