From c944d4af5fe71cf96f2e8ef32199793105aeee58 Mon Sep 17 00:00:00 2001 From: Saugat Pachhai Date: Mon, 6 Sep 2021 22:02:00 +0545 Subject: [PATCH] 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 | 1 + dvc/fs/utils.py | 55 ++++++++++++++++++++++++++++++++++++++++++ dvc/objects/db/base.py | 40 +++++------------------------- 3 files changed, 62 insertions(+), 34 deletions(-) create mode 100644 dvc/fs/utils.py diff --git a/dvc/fs/__init__.py b/dvc/fs/__init__.py index 230a78733..84de0aabf 100644 --- a/dvc/fs/__init__.py +++ b/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 diff --git a/dvc/fs/utils.py b/dvc/fs/utils.py new file mode 100644 index 000000000..c585e6fa4 --- /dev/null +++ b/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 diff --git a/dvc/objects/db/base.py b/dvc/objects/db/base.py index ef3993393..4be316903 100644 --- a/dvc/objects/db/base.py +++ b/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: