Add _pipe_file and test

sshfs/spec.py

@@ -331,3 +331,16 @@ async def _cat_file(self, path, **kwargs):
         async with self._pool.get() as channel:
             async with channel.open(path, "rb") as f:
                 return await f.read()
+
+    @wrap_exceptions
+    async def _pipe_file(self, path, data, chunksize=50 * 2**20, **kwargs):
+        """Asynchronously writes the given data to a remote file in chunks."""
+        await self._makedirs(self._parent(path), exist_ok=True)
+
+        async with self._pool.get() as channel:
+            async with channel.open(path, "wb") as f:
+                for i in range(0, len(data), chunksize):
+                    chunk = data[i : i + chunksize]
+                    await f.write(chunk)
+
+        self.invalidate_cache(path)

tests/test_sshfs.py

@@ -296,7 +296,6 @@ def read_random_file(name):
             return stream.read()
 
     with futures.ThreadPoolExecutor() as executor:
-
         write_futures, _ = futures.wait(
             [executor.submit(create_random_file) for _ in range(64)],
             return_when=futures.ALL_COMPLETED,
@@ -361,3 +360,15 @@ def test_cat_file_sync(fs, remote_dir):
     assert (
         read_content == test_content
     ), "The content read from the file does not match the content written."
+
+
+def test_pipe_file(fs, remote_dir):
+    test_data = b"Test data for pipe_file" * (2**20)  # 1 MB of test data
+    test_file_path = remote_dir + "/test_pipe_file.txt"
+
+    fs.pipe_file(test_file_path, test_data)
+
+    with fs.open(test_file_path, "rb") as f:
+        assert (
+            f.read() == test_data
+        ), "The data read from the file does not match the data written."