Add _pipe_file and test to http

fsspec/implementations/http.py

@@ -522,6 +522,34 @@ async def _isdir(self, path):
         except (FileNotFoundError, ValueError):
             return False
 
+    async def _pipe_file(self, path, value, mode="overwrite", **kwargs):
+        """
+        Write bytes to a remote file over HTTP.
+
+        Parameters
+        ----------
+        path : str
+            Target URL where the data should be written
+        value : bytes
+            Data to be written
+        mode : str
+            How to write to the file - 'overwrite' or 'append'
+        **kwargs : dict
+            Additional parameters to pass to the HTTP request
+        """
+        url = self._strip_protocol(path)
+        headers = kwargs.pop("headers", {})
+        headers["Content-Length"] = str(len(value))
+
+        if not hasattr(self, "session"):
+            import aiohttp
+
+            self.session = aiohttp.ClientSession()
+
+        async with self.session.put(url, data=value, headers=headers, **kwargs) as r:
+            r.raise_for_status()
+        return True
+
 
 class HTTPFile(AbstractBufferedFile):
     """

fsspec/implementations/tests/test_http.py

@@ -581,3 +581,42 @@ async def test_async_walk(server):
             pass
 
     await fs._session.close()
+
+
+def test_pipe_file(server, tmpdir, reset_files):
+    """Test that the pipe_file method works correctly."""
+    import io
+    import fsspec
+
+    # Create test data
+    test_content = b"This is test data to pipe to a file"
+
+    # Initialize filesystem
+    fs = fsspec.filesystem("http", headers={"accept_put": "true"})
+
+    # Test that the file doesn't exist yet
+    with pytest.raises(FileNotFoundError):
+        fs.info(server.address + "/piped_file")
+
+    # Pipe data to the file
+    fs.pipe_file(server.address + "/piped_file", test_content)
+
+    # Verify the file exists now
+    assert fs.exists(server.address + "/piped_file")
+
+    # Verify content
+    assert fs.cat(server.address + "/piped_file") == test_content
+
+    # Test with different modes and headers
+    fs.pipe_file(
+        server.address + "/piped_file2",
+        test_content,
+        mode="overwrite",
+        headers={"Content-Type": "text/plain"},
+    )
+    assert fs.cat(server.address + "/piped_file2") == test_content
+
+    # Test with byte-like object
+    bytesio = io.BytesIO(b"BytesIO content")
+    fs.pipe_file(server.address + "/piped_bytes", bytesio.getvalue())
+    assert fs.cat(server.address + "/piped_bytes") == b"BytesIO content"