http: allow put_file()ing file-like objects

fsspec/implementations/http.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import, division, print_function
 
 import asyncio
+import io
 import logging
 import re
 import weakref
@@ -14,7 +15,7 @@
 from fsspec.callbacks import _DEFAULT_CALLBACK
 from fsspec.exceptions import FSTimeoutError
 from fsspec.spec import AbstractBufferedFile
-from fsspec.utils import DEFAULT_BLOCK_SIZE, tokenize
+from fsspec.utils import DEFAULT_BLOCK_SIZE, nullcontext, tokenize
 
 from ..caching import AllBytes
 
@@ -254,17 +255,29 @@ async def _get_file(
 
     async def _put_file(
         self,
-        rpath,
         lpath,
+        rpath,
         chunk_size=5 * 2 ** 20,
         callback=_DEFAULT_CALLBACK,
         method="post",
         **kwargs,
     ):
         async def gen_chunks():
-            with open(rpath, "rb") as f:
-                callback.set_size(f.seek(0, 2))
-                f.seek(0)
+            # Support passing arbitrary file-like objects
+            # and use them instead of streams.
+            if isinstance(lpath, io.IOBase):
+                context = nullcontext(lpath)
+                use_seek = False  # might not support seeking
+            else:
+                context = open(lpath, "rb")
+                use_seek = True
+
+            with context as f:
+                if use_seek:
+                    callback.set_size(f.seek(0, 2))
+                    f.seek(0)
+                else:
+                    callback.set_size(getattr(f, "size", None))
 
                 chunk = f.read(64 * 1024)
                 while chunk:
@@ -283,8 +296,8 @@ async def gen_chunks():
             )
 
         meth = getattr(session, method)
-        async with meth(lpath, data=gen_chunks(), **kw) as resp:
-            self._raise_not_found_for_status(resp, lpath)
+        async with meth(rpath, data=gen_chunks(), **kw) as resp:
+            self._raise_not_found_for_status(resp, rpath)
 
     async def _exists(self, path, **kwargs):
         kw = self.kwargs.copy()

fsspec/implementations/tests/test_http.py

@@ -1,5 +1,6 @@
 import asyncio
 import contextlib
+import io
 import os
 import sys
 import threading
@@ -465,6 +466,14 @@ def test_put_file(server, tmp_path, method, reset_files):
     fs.get_file(server + "/hey", dwl_file)
     assert dwl_file.read_bytes() == data
 
+    src_file.write_bytes(b"xxx")
+    with open(src_file, "rb") as stream:
+        fs.put_file(stream, server + "/hey_2", method=method)
+    assert fs.cat(server + "/hey_2") == b"xxx"
+
+    fs.put_file(io.BytesIO(b"yyy"), server + "/hey_3", method=method)
+    assert fs.cat(server + "/hey_3") == b"yyy"
+
 
 @pytest.mark.xfail(
     condition=sys.flags.optimize > 1, reason="no docstrings when optimised"

fsspec/utils.py

@@ -4,6 +4,7 @@
 import pathlib
 import re
 import sys
+from contextlib import contextmanager
 from functools import partial
 from hashlib import md5
 from urllib.parse import urlsplit
@@ -466,3 +467,8 @@ def wrapper(cls):
         return cls
 
     return wrapper
+
+
+@contextmanager
+def nullcontext(obj):
+    yield obj