fs: temporary implement get_file

src/scmrepo/fs.py

@@ -3,7 +3,9 @@
 import posixpath
 from typing import TYPE_CHECKING, Any, BinaryIO, Callable, Dict, Optional, Tuple
 
+from fsspec.callbacks import _DEFAULT_CALLBACK
 from fsspec.spec import AbstractFileSystem
+from fsspec.utils import isfilelike
 
 if TYPE_CHECKING:
     from io import BytesIO
@@ -242,3 +244,32 @@ def ls(self, path, detail=True, **kwargs):
             return paths
 
         return [self.info(_path) for _path in paths]
+
+    def get_file(
+        self, rpath, lpath, callback=_DEFAULT_CALLBACK, outfile=None, **kwargs
+    ):
+        # NOTE: temporary workaround while waiting for
+        # https://github.com/fsspec/filesystem_spec/pull/1191
+
+        if isfilelike(lpath):
+            outfile = lpath
+        elif self.isdir(rpath):
+            os.makedirs(lpath, exist_ok=True)
+            return None
+
+        with self.open(rpath, "rb", **kwargs) as f1:
+            if outfile is None:
+                outfile = open(lpath, "wb")
+
+            try:
+                callback.set_size(getattr(f1, "size", None))
+                data = True
+                while data:
+                    data = f1.read(self.blocksize)
+                    segment_len = outfile.write(data)
+                    if segment_len is None:
+                        segment_len = len(data)
+                    callback.relative_update(segment_len)
+            finally:
+                if not isfilelike(lpath):
+                    outfile.close()