extract transfer related logic from odb to fs.utils.transfer (#6539)

* extract transfer related logic from odb to fs.utils.transfer

* don't makedirs on dirinfo

As local odb needs to set a mode for dir, we cannot create a directory
in the transfer

dvc/fs/__init__.py

@@ -2,6 +2,7 @@ import posixpath
 from urllib.parse import urlparse
 
 from ..scheme import Schemes
+from . import utils  # noqa: F401
 from .azure import AzureFileSystem
 from .gdrive import GDriveFileSystem
 from .gs import GSFileSystem

dvc/fs/utils.py

@@ -0,0 +1,55 @@
+import logging
+import os
+from typing import TYPE_CHECKING
+
+from .local import LocalFileSystem
+
+if TYPE_CHECKING:
+    from dvc.types import AnyPath, DvcPath
+
+    from .base import BaseFileSystem
+
+logger = logging.getLogger(__name__)
+
+
+def transfer(
+    from_fs: "BaseFileSystem",
+    from_info: "AnyPath",
+    to_fs: "BaseFileSystem",
+    to_info: "DvcPath",
+    move: bool = False,
+) -> None:
+    use_move = isinstance(from_fs, type(to_fs)) and move
+    try:
+        if use_move:
+            return to_fs.move(from_info, to_info)
+
+        if isinstance(from_fs, LocalFileSystem):
+            if not isinstance(from_info, from_fs.PATH_CLS):
+                from_info = from_fs.PATH_CLS(from_info)
+            return to_fs.upload(from_info, to_info)
+
+        if isinstance(to_fs, LocalFileSystem):
+            return from_fs.download_file(from_info, to_info)
+
+        with from_fs.open(from_info, mode="rb") as fobj:
+            return to_fs.upload_fobj(fobj, to_info)
+    except OSError as exc:
+        # If the target file already exists, we are going to simply
+        # ignore the exception (#4992).
+        #
+        # On Windows, it is not always guaranteed that you'll get
+        # FileExistsError (it might be PermissionError or a bare OSError)
+        # but all of those exceptions raised from the original
+        # FileExistsError so we have a separate check for that.
+        if isinstance(exc, FileExistsError) or (
+            os.name == "nt"
+            and exc.__context__
+            and isinstance(exc.__context__, FileExistsError)
+        ):
+            logger.debug("'%s' file already exists, skipping", to_info)
+            if use_move:
+                from_fs.remove(from_info)
+            return None
+
+        raise

dvc/objects/db/base.py

@@ -1,12 +1,10 @@
 import itertools
 import logging
-import os
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import suppress
 from copy import copy
 from typing import TYPE_CHECKING, Optional
 
-from dvc.fs.local import LocalFileSystem
 from dvc.objects.errors import ObjectDBPermissionError, ObjectFormatError
 from dvc.objects.file import HashFile
 from dvc.progress import Tqdm
@@ -85,38 +83,12 @@ class ObjectDB:
         _hash_info: "HashInfo",
         move: bool = False,
     ):
+        from dvc import fs
+
         self.makedirs(to_info.parent)
-        use_move = isinstance(from_fs, type(self.fs)) and move
-        try:
-            if use_move:
-                self.fs.move(from_info, to_info)
-            elif isinstance(from_fs, LocalFileSystem):
-                if not isinstance(from_info, from_fs.PATH_CLS):
-                    from_info = from_fs.PATH_CLS(from_info)
-                self.fs.upload(from_info, to_info)
-            elif isinstance(self.fs, LocalFileSystem):
-                from_fs.download_file(from_info, to_info)
-            else:
-                with from_fs.open(from_info, mode="rb") as fobj:
-                    self.fs.upload_fobj(fobj, to_info)
-        except OSError as exc:
-            # If the target file already exists, we are going to simply
-            # ignore the exception (#4992).
-            #
-            # On Windows, it is not always guaranteed that you'll get
-            # FileExistsError (it might be PermissionError or a bare OSError)
-            # but all of those exceptions raised from the original
-            # FileExistsError so we have a separate check for that.
-            if isinstance(exc, FileExistsError) or (
-                os.name == "nt"
-                and exc.__context__
-                and isinstance(exc.__context__, FileExistsError)
-            ):
-                logger.debug("'%s' file already exists, skipping", to_info)
-                if use_move:
-                    from_fs.remove(from_info)
-            else:
-                raise
+        return fs.utils.transfer(
+            from_fs, from_info, self.fs, to_info, move=move
+        )
 
     def add(
         self,
@@ -139,7 +111,7 @@ class ObjectDB:
             pass
 
         cache_info = self.hash_to_path_info(hash_info.value)
-        self._add_file(fs, path_info, cache_info, hash_info, move)
+        self._add_file(fs, path_info, cache_info, hash_info, move=move)
 
         try:
             if verify: