diff --git a/dvc/cli/__init__.py b/dvc/cli/__init__.py index f94000311..69d00bd10 100644 --- a/dvc/cli/__init__.py +++ b/dvc/cli/__init__.py @@ -121,7 +121,7 @@ def _log_exceptions(exc: Exception) -> Optional[int]: ) return None - from dvc.data.stage import IgnoreInCollectedDirError + from dvc_data.stage import IgnoreInCollectedDirError if isinstance(exc, IgnoreInCollectedDirError): logger.exception("") diff --git a/dvc/data/__init__.py b/dvc/data/__init__.py deleted file mode 100644 index 747232191..000000000 --- a/dvc/data/__init__.py +++ /dev/null @@ -1,33 +0,0 @@ -import logging -from typing import TYPE_CHECKING, Iterator, Union - -from .tree import Tree - -if TYPE_CHECKING: - from dvc_objects.db import ObjectDB - from dvc_objects.file import HashFile - from dvc_objects.hash_info import HashInfo - -logger = logging.getLogger(__name__) - - -def check(odb: "ObjectDB", obj: "HashFile", **kwargs): - if isinstance(obj, Tree): - for _, _, oid in obj: - odb.check(oid, **kwargs) - - odb.check(obj.hash_info, **kwargs) - - -def load(odb: "ObjectDB", hash_info: "HashInfo") -> "HashFile": - if hash_info.isdir: - return Tree.load(odb, hash_info) - return odb.get(hash_info) - - -def iterobjs( - obj: Union["Tree", "HashFile"] -) -> Iterator[Union["Tree", "HashFile"]]: - if isinstance(obj, Tree): - yield from (entry_obj for _, entry_obj in obj) - yield obj diff --git a/dvc/data/_progress.py b/dvc/data/_progress.py deleted file mode 100644 index 437aa3188..000000000 --- a/dvc/data/_progress.py +++ /dev/null @@ -1,33 +0,0 @@ -from dvc_objects._tqdm import Tqdm - - -class QueryingProgress(Tqdm): - def __init__(self, iterable=None, total=None, name=None, phase="Querying"): - msg_part = "cache in " + f"'{name}'" if name else "remote cache" - msg_fmt = "{phase} " + msg_part - - self._estimating_msg = msg_fmt.format(phase="Estimating size of") - self._listing_msg = msg_fmt.format(phase="Querying") - self.desc = desc = msg_fmt.format(phase=phase) - super().__init__( - iterable=iterable, - desc=desc, - total=total, - unit="files", - unit_scale=False, - bar_format=self.BAR_FMT_DEFAULT, - ) - - def callback(self, phase, *args): - total = args[0] if args else self.total - completed = args[1] if len(args) > 1 else self.n - desc = self.desc - if phase == "estimating": - desc = self._estimating_msg - elif phase == "querying": - desc = self._listing_msg - - if desc != self.desc: - # let Tqdm take care of when to update - self.set_description_str(desc, refresh=False) - self.update_to(completed, total) diff --git a/dvc/data/checkout.py b/dvc/data/checkout.py deleted file mode 100644 index 73bc7120c..000000000 --- a/dvc/data/checkout.py +++ /dev/null @@ -1,299 +0,0 @@ -import logging -from itertools import chain -from typing import TYPE_CHECKING, List, Optional - -from dvc_objects.fs.callbacks import Callback -from dvc_objects.fs.generic import test_links, transfer - -from .diff import ROOT -from .diff import diff as odiff -from .stage import stage - -if TYPE_CHECKING: - from dvc_objects._ignore import Ignore - -logger = logging.getLogger(__name__) - - -class PromptError(Exception): - def __init__(self, path: str) -> None: - self.path = path - super().__init__(f"unable to remove '{path}' without a confirmation.") - - -class CheckoutError(Exception): - def __init__(self, paths: List[str]) -> None: - self.paths = paths - super().__init__("Checkout failed") - - -class LinkError(Exception): - def __init__(self, path: str) -> None: - self.path = path - super().__init__("No possible cache link types for '{path}'.") - - -def _remove(fs_path, fs, in_cache, force=False, prompt=None): - if not fs.exists(fs_path): - return - - if force: - fs.remove(fs_path) - return - - if not in_cache: - msg = ( - f"file/directory '{fs_path}' is going to be removed. " - "Are you sure you want to proceed?" - ) - - if prompt is None or not prompt(msg): - raise PromptError(fs_path) - - fs.remove(fs_path) - - -def _relink(link, cache, cache_info, fs, path, in_cache, force, prompt=None): - _remove(path, fs, in_cache, force=force, prompt=prompt) - link(cache, cache_info, fs, path) - # NOTE: Depending on a file system (e.g. on NTFS), `_remove` might reset - # read-only permissions in order to delete a hardlink to protected object, - # which will also reset it for the object itself, making it unprotected, - # so we need to protect it back. - cache.protect(cache_info) - - -def _checkout_file( - link, - fs_path, - fs, - change, - cache, - force, - relink=False, - state=None, - prompt=None, -): - """The file is changed we need to checkout a new copy""" - modified = False - - cache_fs_path = cache.hash_to_path(change.new.oid.value) - if change.old.oid: - if relink: - if fs.iscopy(fs_path) and cache.cache_types[0] == "copy": - cache.unprotect(fs_path) - else: - _relink( - link, - cache, - cache_fs_path, - fs, - fs_path, - change.old.in_cache, - force=force, - prompt=prompt, - ) - else: - modified = True - _relink( - link, - cache, - cache_fs_path, - fs, - fs_path, - change.old.in_cache, - force=force, - prompt=prompt, - ) - else: - link(cache, cache_fs_path, fs, fs_path) - modified = True - - if state: - state.save(fs_path, fs, change.new.oid) - - return modified - - -def _diff( - fs_path, - fs, - obj, - cache, - relink=False, - ignore: Optional["Ignore"] = None, -): - old = None - try: - _, _, old = stage( - cache, - fs_path, - fs, - obj.hash_info.name if obj else cache.fs.PARAM_CHECKSUM, - dry_run=True, - ignore=ignore, - ) - except FileNotFoundError: - pass - - diff = odiff(old, obj, cache) - - if relink: - diff.modified.extend(diff.unchanged) - - return diff - - -class Link: - def __init__(self, links): - self._links = links - - def __call__(self, cache, from_path, to_fs, to_path, callback=None): - if to_fs.exists(to_path): - to_fs.remove(to_path) # broken symlink - - cache.makedirs(cache.fs.path.parent(to_path)) - try: - with Callback.as_tqdm_callback( - callback, - desc=cache.fs.path.name(from_path), - bytes=True, - ) as cb: - transfer( - cache.fs, - from_path, - to_fs, - to_path, - links=self._links, - callback=cb, - ) - except FileNotFoundError as exc: - raise CheckoutError([to_path]) from exc - except OSError as exc: - raise LinkError(to_path) from exc - - -def _checkout( - diff, - fs_path, - fs, - cache, - force=False, - progress_callback=None, - relink=False, - state=None, - prompt=None, -): - if not diff: - return - - links = test_links(cache.cache_types, cache.fs, cache.fs_path, fs, fs_path) - if not links: - raise LinkError(fs_path) - link = Link(links) - for change in diff.deleted: - entry_path = ( - fs.path.join(fs_path, *change.old.key) - if change.old.key != ROOT - else fs_path - ) - _remove( - entry_path, fs, change.old.in_cache, force=force, prompt=prompt - ) - - failed = [] - for change in chain(diff.added, diff.modified): - entry_path = ( - fs.path.join(fs_path, *change.new.key) - if change.new.key != ROOT - else fs_path - ) - if change.new.oid.isdir: - fs.makedirs(entry_path) - continue - - try: - _checkout_file( - link, - entry_path, - fs, - change, - cache, - force, - relink, - state=state, - prompt=prompt, - ) - if progress_callback: - progress_callback(entry_path) - except CheckoutError as exc: - failed.extend(exc.paths) - - if failed: - raise CheckoutError(failed) - - -def checkout( - fs_path, - fs, - obj, - cache, - force=False, - progress_callback=None, - relink=False, - quiet=False, - ignore: Optional["Ignore"] = None, - state=None, - prompt=None, -): - # if protocol(fs_path) not in ["local", cache.fs.protocol]: - # raise NotImplementedError - - diff = _diff( - fs_path, - fs, - obj, - cache, - relink=relink, - ignore=ignore, - ) - - failed = [] - if not obj: - if not quiet: - logger.warning( - "No file hash info found for '%s'. It won't be created.", - fs_path, - ) - failed.append(fs_path) - elif not diff: - logger.trace("Data '%s' didn't change.", fs_path) # type: ignore - - try: - _checkout( - diff, - fs_path, - fs, - cache, - force=force, - progress_callback=progress_callback, - relink=relink, - state=state, - prompt=prompt, - ) - except CheckoutError as exc: - failed.extend(exc.paths) - - if diff and state: - state.save_link(fs_path, fs) - if not failed: - state.save(fs_path, fs, obj.hash_info) - - if failed or not diff: - if progress_callback and obj: - progress_callback(fs_path, len(obj)) - if failed: - raise CheckoutError(failed) - return - - return bool(diff) and not relink diff --git a/dvc/data/db/__init__.py b/dvc/data/db/__init__.py deleted file mode 100644 index fc0b3fbdd..000000000 --- a/dvc/data/db/__init__.py +++ /dev/null @@ -1,30 +0,0 @@ -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from .index import ObjectDBIndexBase - - -def get_odb(fs, fs_path, **config): - from dvc_objects.db import ObjectDB - from dvc_objects.fs import Schemes - - from .local import LocalObjectDB - - if fs.protocol == Schemes.LOCAL: - return LocalObjectDB(fs, fs_path, **config) - - return ObjectDB(fs, fs_path, **config) - - -def get_index(odb) -> "ObjectDBIndexBase": - import hashlib - - from .index import ObjectDBIndex, ObjectDBIndexNoop - - cls = ObjectDBIndex if odb.tmp_dir else ObjectDBIndexNoop - return cls( - odb.tmp_dir, - hashlib.sha256( - odb.fs.unstrip_protocol(odb.fs_path).encode("utf-8") - ).hexdigest(), - ) diff --git a/dvc/data/db/index.py b/dvc/data/db/index.py deleted file mode 100644 index 52639f1be..000000000 --- a/dvc/data/db/index.py +++ /dev/null @@ -1,136 +0,0 @@ -import logging -import os -from abc import ABC, abstractmethod -from typing import TYPE_CHECKING, Iterable, Set - -from dvc_objects.errors import ObjectDBError - -if TYPE_CHECKING: - from dvc_objects.fs.base import AnyFSPath - -logger = logging.getLogger(__name__) - - -class ObjectDBIndexBase(ABC): - @abstractmethod - def __init__( - self, - tmp_dir: "AnyFSPath", - name: str, - ): - pass - - @abstractmethod - def __iter__(self): - pass - - @abstractmethod - def __contains__(self, hash_): - pass - - def hashes(self): - return iter(self) - - @abstractmethod - def dir_hashes(self): - pass - - @abstractmethod - def clear(self): - pass - - @abstractmethod - def update(self, dir_hashes: Iterable[str], file_hashes: Iterable[str]): - pass - - @abstractmethod - def intersection(self, hashes: Set[str]): - pass - - -class ObjectDBIndexNoop(ObjectDBIndexBase): - """No-op class for ODBs which are not indexed.""" - - def __init__( - self, - tmp_dir: "AnyFSPath", - name: str, - ): # pylint: disable=super-init-not-called - pass - - def __iter__(self): - return iter([]) - - def __contains__(self, hash_): - return False - - def dir_hashes(self): - return [] - - def clear(self): - pass - - def update(self, dir_hashes: Iterable[str], file_hashes: Iterable[str]): - pass - - def intersection(self, hashes: Set[str]): - return [] - - -class ObjectDBIndex(ObjectDBIndexBase): - """Class for indexing hashes in an ODB.""" - - INDEX_SUFFIX = ".idx" - INDEX_DIR = "index" - - def __init__( - self, - tmp_dir: "AnyFSPath", - name: str, - ): # pylint: disable=super-init-not-called - from dvc_objects.cache import Cache, Index - from dvc_objects.fs import LocalFileSystem - from dvc_objects.fs.utils import makedirs - - self.index_dir = os.path.join(tmp_dir, self.INDEX_DIR, name) - makedirs(self.index_dir, exist_ok=True) - self.fs = LocalFileSystem() - cache = Cache(self.index_dir, eviction_policy="none", type="index") - self.index = Index.fromcache(cache) - - def __iter__(self): - return iter(self.index) - - def __contains__(self, hash_): - return hash_ in self.index - - def dir_hashes(self): - """Iterate over .dir hashes stored in the index.""" - yield from (hash_ for hash_, is_dir in self.index.items() if is_dir) - - def clear(self): - """Clear this index (to force re-indexing later).""" - from dvc_objects.cache import Timeout - - try: - self.index.clear() - except Timeout as exc: - raise ObjectDBError("Failed to clear ODB index") from exc - - def update(self, dir_hashes: Iterable[str], file_hashes: Iterable[str]): - """Update this index, adding the specified hashes.""" - from dvc_objects.cache import Timeout - - try: - with self.index.transact(): - for hash_ in dir_hashes: - self.index[hash_] = True - with self.index.transact(): - for hash_ in file_hashes: - self.index[hash_] = False - except Timeout as exc: - raise ObjectDBError("Failed to update ODB index") from exc - - def intersection(self, hashes: Set[str]): - """Iterate over values from `hashes` which exist in the index.""" - yield from hashes.intersection(self.index.keys()) diff --git a/dvc/data/db/local.py b/dvc/data/db/local.py deleted file mode 100644 index 45d9712cf..000000000 --- a/dvc/data/db/local.py +++ /dev/null @@ -1,164 +0,0 @@ -import logging -import os -import stat -from functools import partial - -from funcy import cached_property -from shortuuid import uuid - -from dvc_objects.db import ObjectDB, noop, wrap_iter -from dvc_objects.errors import ObjectDBError, ObjectFormatError -from dvc_objects.fs.system import umask -from dvc_objects.fs.utils import copyfile, relpath, remove, walk_files -from dvc_objects.hash_info import HashInfo - -logger = logging.getLogger(__name__) - - -class LocalObjectDB(ObjectDB): - DEFAULT_CACHE_TYPES = ["reflink", "copy"] - CACHE_MODE = 0o444 - UNPACKED_DIR_SUFFIX = ".unpacked" - - def __init__(self, fs, fs_path, **config): - super().__init__(fs, fs_path, **config) - self.cache_dir = fs_path - - shared = config.get("shared") - if shared: - self._file_mode = 0o664 - self._dir_mode = 0o2775 - else: - self._file_mode = 0o666 & ~umask - self._dir_mode = 0o777 & ~umask - - @property - def cache_dir(self): - return self.fs_path if self.fs_path else None - - @cache_dir.setter - def cache_dir(self, value): - self.fs_path = value - - @cached_property - def cache_path(self): - return os.path.abspath(self.cache_dir) - - def move(self, from_info, to_info): - super().move(from_info, to_info) - os.chmod(to_info, self._file_mode) - - def makedirs(self, fs_path): - from dvc_objects.fs.utils import makedirs - - makedirs(fs_path, exist_ok=True, mode=self._dir_mode) - - def hash_to_path(self, hash_): - # NOTE: `self.cache_path` is already normalized so we can simply use - # `os.sep` instead of `os.path.join`. This results in this helper - # being ~5.5 times faster. - return f"{self.cache_path}{os.sep}{hash_[0:2]}{os.sep}{hash_[2:]}" - - def hashes_exist( - self, hashes, jobs=None, progress=noop - ): # pylint: disable=unused-argument - ret = [] - progress = partial(progress, "querying", len(hashes)) - - for hash_ in wrap_iter(hashes, progress): - hash_info = HashInfo(self.fs.PARAM_CHECKSUM, hash_) - try: - self.check(hash_info) - ret.append(hash_) - except (FileNotFoundError, ObjectFormatError): - pass - - return ret - - def _list_paths(self, prefix=None): - assert self.fs_path is not None - if prefix: - fs_path = self.fs.path.join(self.fs_path, prefix[:2]) - if not self.fs.exists(fs_path): - return - else: - fs_path = self.fs_path - - # NOTE: use utils.fs walk_files since fs.walk_files will not follow - # symlinks - yield from walk_files(fs_path) - - def _remove_unpacked_dir(self, hash_): - hash_fs_path = self.hash_to_path(hash_) - fs_path = self.fs.path.with_name( - hash_fs_path, - self.fs.path.name(hash_fs_path) + self.UNPACKED_DIR_SUFFIX, - ) - self.fs.remove(fs_path) - - def _unprotect_file(self, path): - if self.fs.is_symlink(path) or self.fs.is_hardlink(path): - logger.debug("Unprotecting '%s'", path) - tmp = os.path.join(os.path.dirname(path), "." + uuid()) - - # The operations order is important here - if some application - # would access the file during the process of copyfile then it - # would get only the part of file. So, at first, the file should be - # copied with the temporary name, and then original file should be - # replaced by new. - copyfile(path, tmp, name=f"Unprotecting '{relpath(path)}'") - remove(path) - os.rename(tmp, path) - - else: - logger.debug( - "Skipping copying for '%s', since it is not " - "a symlink or a hardlink.", - path, - ) - - os.chmod(path, self._file_mode) - - def _unprotect_dir(self, path): - for fname in self.fs.find(path): - self._unprotect_file(fname) - - def unprotect(self, fs_path): - if not os.path.exists(fs_path): - raise ObjectDBError( - f"can't unprotect non-existing data '{fs_path}'" - ) - - if os.path.isdir(fs_path): - self._unprotect_dir(fs_path) - else: - self._unprotect_file(fs_path) - - def protect(self, fs_path): - try: - os.chmod(fs_path, self.CACHE_MODE) - except OSError: - # NOTE: not being able to protect cache file is not fatal, it - # might happen on funky filesystems (e.g. Samba, see #5255), - # read-only filesystems or in a shared cache scenario. - logger.trace("failed to protect '%s'", fs_path, exc_info=True) - - def is_protected(self, fs_path): - try: - mode = os.stat(fs_path).st_mode - except FileNotFoundError: - return False - - return stat.S_IMODE(mode) == self.CACHE_MODE - - def set_exec(self, fs_path): - mode = os.stat(fs_path).st_mode | stat.S_IEXEC - try: - os.chmod(fs_path, mode) - except OSError: - logger.trace( - "failed to chmod '%s' '%s'", - oct(mode), - fs_path, - exc_info=True, - ) diff --git a/dvc/data/db/reference.py b/dvc/data/db/reference.py deleted file mode 100644 index 3173b81b3..000000000 --- a/dvc/data/db/reference.py +++ /dev/null @@ -1,88 +0,0 @@ -import logging -import os -from typing import TYPE_CHECKING, Dict - -from dvc_objects.db import ObjectDB -from dvc_objects.errors import ObjectFormatError -from dvc_objects.fs import Schemes - -from ..reference import ReferenceHashFile - -if TYPE_CHECKING: - from dvc_objects.fs.base import AnyFSPath, FileSystem - from dvc_objects.fs.callbacks import Callback - from dvc_objects.hash_info import HashInfo - -logger = logging.getLogger(__name__) - - -class ReferenceObjectDB(ObjectDB): - """Reference ODB. - - File objects are stored as ReferenceHashFiles which reference paths outside - of the staging ODB fs. Tree objects are stored natively. - """ - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self._fs_cache: Dict[tuple, "FileSystem"] = {} - self._obj_cache: Dict["HashInfo", "ReferenceHashFile"] = {} - - def get(self, hash_info: "HashInfo"): - if hash_info.isdir: - return super().get(hash_info) - try: - return self._obj_cache[hash_info] - except KeyError: - pass - fs_path = self.hash_to_path(hash_info.value) - try: - ref_file = ReferenceHashFile.from_bytes( - self.fs.cat_file(fs_path), - fs_cache=self._fs_cache, - ) - except OSError: - raise FileNotFoundError - try: - ref_file.check(self, check_hash=False) - except ObjectFormatError: - self.fs.remove(fs_path) - raise - self._obj_cache[hash_info] = ref_file - return ref_file - - def _add_file( - self, - from_fs: "FileSystem", - from_info: "AnyFSPath", - to_info: "AnyFSPath", - hash_info: "HashInfo", - hardlink: bool = False, - callback: "Callback" = None, - ): - if hash_info.isdir: - return super()._add_file( - from_fs, - from_info, - to_info, - hash_info, - hardlink=hardlink, - callback=callback, - ) - - self.makedirs(self.fs.path.parent(to_info)) - ref_file = ReferenceHashFile(from_info, from_fs, hash_info) - self._obj_cache[hash_info] = ref_file - try: - self.fs.pipe_file(to_info, ref_file.to_bytes()) - except OSError as exc: - 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) - else: - raise - if from_fs.protocol != Schemes.LOCAL: - self._fs_cache[ReferenceHashFile.config_tuple(from_fs)] = from_fs diff --git a/dvc/data/diff.py b/dvc/data/diff.py deleted file mode 100644 index ae205f595..000000000 --- a/dvc/data/diff.py +++ /dev/null @@ -1,133 +0,0 @@ -from dataclasses import dataclass, field -from typing import TYPE_CHECKING, List, Optional, Tuple - -if TYPE_CHECKING: - from dvc_objects.file import HashFile - from dvc_objects.hash_info import HashInfo - -ADD = "add" -MODIFY = "modify" -DELETE = "delete" -UNCHANGED = "unchanged" - - -@dataclass -class TreeEntry: - in_cache: bool - key: Tuple[str] - oid: Optional["HashInfo"] = field(default=None) - - def __bool__(self): - return bool(self.oid) - - def __eq__(self, other): - if not isinstance(other, TreeEntry): - return False - - if self.key != other.key: - return False - - return self.oid == other.oid - - -@dataclass -class Change: - old: TreeEntry - new: TreeEntry - - @property - def typ(self): - if not self.old and not self.new: - return UNCHANGED - - if self.old and not self.new: - return DELETE - - if not self.old and self.new: - return ADD - - if self.old != self.new: - return MODIFY - - return UNCHANGED - - def __bool__(self): - return self.typ != UNCHANGED - - -@dataclass -class DiffResult: - added: List[Change] = field(default_factory=list, compare=True) - modified: List[Change] = field(default_factory=list, compare=True) - deleted: List[Change] = field(default_factory=list, compare=True) - unchanged: List[Change] = field(default_factory=list, compare=True) - - def __bool__(self): - return bool(self.added or self.modified or self.deleted) - - -ROOT = ("",) - - -def diff( - old: Optional["HashFile"], new: Optional["HashFile"], cache -) -> DiffResult: - from .tree import Tree - - if old is None and new is None: - return DiffResult() - - def _get_keys(obj): - if not obj: - return [] - return [ROOT] + ( - [key for key, _, _ in obj] if isinstance(obj, Tree) else [] - ) - - old_keys = set(_get_keys(old)) - new_keys = set(_get_keys(new)) - - def _get_oid(obj, key): - if not obj or key == ROOT: - return obj.hash_info if obj else None - - entry_obj = obj.get(cache, key) - return entry_obj.hash_info if entry_obj else None - - def _in_cache(oid, cache): - from dvc_objects.errors import ObjectFormatError - - if not oid: - return False - - try: - cache.check(oid) - return True - except (FileNotFoundError, ObjectFormatError): - return False - - ret = DiffResult() - for key in old_keys | new_keys: - old_oid = _get_oid(old, key) - new_oid = _get_oid(new, key) - - change = Change( - old=TreeEntry(_in_cache(old_oid, cache), key, old_oid), - new=TreeEntry(_in_cache(new_oid, cache), key, new_oid), - ) - - if change.typ == ADD: - ret.added.append(change) - elif change.typ == MODIFY: - ret.modified.append(change) - elif change.typ == DELETE: - ret.deleted.append(change) - else: - assert change.typ == UNCHANGED - if not change.new.in_cache and not ( - change.new.oid and change.new.oid.isdir - ): - ret.modified.append(change) - else: - ret.unchanged.append(change) - return ret diff --git a/dvc/data/gc.py b/dvc/data/gc.py deleted file mode 100644 index 093b47e72..000000000 --- a/dvc/data/gc.py +++ /dev/null @@ -1,40 +0,0 @@ -def gc(odb, used, jobs=None, cache_odb=None, shallow=True): - from dvc_objects.errors import ObjectDBPermissionError - - from ._progress import QueryingProgress - from .tree import Tree - - if odb.read_only: - raise ObjectDBPermissionError("Cannot gc read-only ODB") - if not cache_odb: - cache_odb = odb - used_hashes = set() - for hash_info in used: - used_hashes.add(hash_info.value) - if hash_info.isdir and not shallow: - tree = Tree.load(cache_odb, hash_info) - used_hashes.update( - entry_obj.hash_info.value for _, entry_obj in tree - ) - - def _is_dir_hash(_hash): - from dvc_objects.hash_info import HASH_DIR_SUFFIX - - return _hash.endswith(HASH_DIR_SUFFIX) - - removed = False - # hashes must be sorted to ensure we always remove .dir files first - - hashes = QueryingProgress(odb.all(jobs), name=odb.fs_path) - for hash_ in sorted(hashes, key=_is_dir_hash, reverse=True): - if hash_ in used_hashes: - continue - fs_path = odb.hash_to_path(hash_) - if _is_dir_hash(hash_): - # backward compatibility - # pylint: disable=protected-access - odb._remove_unpacked_dir(hash_) - odb.fs.remove(fs_path) - removed = True - - return removed diff --git a/dvc/data/reference.py b/dvc/data/reference.py deleted file mode 100644 index 49f0168b4..000000000 --- a/dvc/data/reference.py +++ /dev/null @@ -1,117 +0,0 @@ -import errno -import logging -import os -import pickle -from typing import TYPE_CHECKING, Optional - -from dvc_objects.errors import ObjectFormatError -from dvc_objects.file import HashFile -from dvc_objects.fs import FS_MAP, LocalFileSystem - -if TYPE_CHECKING: - from dvc_objects.db import ObjectDB - from dvc_objects.fs.base import AnyFSPath, FileSystem - from dvc_objects.hash_info import HashInfo - -logger = logging.getLogger(__name__) - - -class ReferenceHashFile(HashFile): - PARAM_PATH = "path" - PARAM_HASH = "hash" - PARAM_CHECKSUM = "checksum" - PARAM_FS_CONFIG = "fs_config" - PARAM_FS_CLS = "fs_name" - - def __init__( - self, - fs_path: "AnyFSPath", - fs: "FileSystem", - hash_info: "HashInfo", - checksum: Optional[str] = None, - **kwargs, - ): - super().__init__(fs_path, fs, hash_info, **kwargs) - self.checksum = checksum or fs.checksum(fs_path) - - def __str__(self): - return f"ref object {self.hash_info} -> {self.fs_path}" - - def check(self, odb: "ObjectDB", check_hash: bool = True): - assert self.fs - if not self.fs.exists(self.fs_path): - raise FileNotFoundError( - errno.ENOENT, os.strerror(errno.ENOENT), self.fs_path - ) - if self.checksum != self._get_checksum(): - raise ObjectFormatError(f"{self} is changed") - if check_hash: - self._check_hash(odb) - - def _get_checksum(self) -> str: - assert self.fs - return self.fs.checksum(self.fs_path) - - def to_bytes(self): - # NOTE: dumping reference FS's this way is insecure, as the - # fully parsed remote FS config will include credentials - # - # ReferenceHashFiles should currently only be serialized in - # memory and not to disk - fs_path = self.fs_path - fs_cls = type(self.fs) - mod = None - if fs_cls not in FS_MAP.values() and fs_cls != LocalFileSystem: - mod = ".".join((fs_cls.__module__, fs_cls.__name__)) - dict_ = { - self.PARAM_PATH: fs_path, - self.PARAM_HASH: self.hash_info, - self.PARAM_CHECKSUM: self.checksum, - self.PARAM_FS_CONFIG: self.config_tuple(self.fs), - self.PARAM_FS_CLS: mod, - } - try: - return pickle.dumps(dict_) - except pickle.PickleError as exc: - raise ObjectFormatError(f"Could not pickle {self}") from exc - - @classmethod - def from_bytes(cls, data: bytes, fs_cache: Optional[dict] = None): - from dvc_objects.fs import get_fs_cls - - try: - dict_ = pickle.loads(data) - except pickle.PickleError as exc: - raise ObjectFormatError("ReferenceHashFile is corrupted") from exc - - try: - fs_path = dict_[cls.PARAM_PATH] - hash_info = dict_[cls.PARAM_HASH] - except KeyError as exc: - raise ObjectFormatError("ReferenceHashFile is corrupted") from exc - - protocol, config_pairs = dict_.get(cls.PARAM_FS_CONFIG) - fs = fs_cache.get((protocol, config_pairs)) if fs_cache else None - if not fs: - config = dict(config_pairs) - mod = dict_.get(cls.PARAM_FS_CLS, None) - fs_cls = get_fs_cls(config, cls=mod, scheme=protocol) - fs = fs_cls(**config) - return ReferenceHashFile( - fs_path, - fs, - hash_info, - checksum=dict_.get(cls.PARAM_CHECKSUM), - ) - - @staticmethod - def config_tuple(fs: "FileSystem"): - return ( - fs.protocol, - tuple( - (key, value) - for key, value in sorted( - fs.config.items(), key=lambda item: item[0] - ) - ), - ) diff --git a/dvc/data/slow_link_detection.py b/dvc/data/slow_link_detection.py deleted file mode 100644 index e69de29bb..000000000 diff --git a/dvc/data/stage.py b/dvc/data/stage.py deleted file mode 100644 index 88fbc3f4e..000000000 --- a/dvc/data/stage.py +++ /dev/null @@ -1,392 +0,0 @@ -import errno -import hashlib -import logging -import os -from concurrent.futures import ThreadPoolExecutor -from functools import partial -from typing import TYPE_CHECKING, Dict, Optional, Tuple - -from dvc_objects._tqdm import Tqdm -from dvc_objects.file import HashFile -from dvc_objects.hash import hash_file -from dvc_objects.hash_info import HashInfo -from dvc_objects.meta import Meta - -from .db.reference import ReferenceObjectDB - -if TYPE_CHECKING: - from dvc_objects._ignore import Ignore - from dvc_objects.db import ObjectDB - from dvc_objects.fs.base import AnyFSPath, FileSystem - - from .tree import Tree - - -DefaultIgnoreFile = ".dvcignore" - - -class IgnoreInCollectedDirError(Exception): - def __init__(self, ignore_file: str, ignore_dirname: str) -> None: - super().__init__( - f"{ignore_file} file should not be in collected dir path: " - f"'{ignore_dirname}'" - ) - - -logger = logging.getLogger(__name__) - - -_STAGING_MEMFS_PATH = "dvc-staging" - - -def _upload_file(from_fs_path, fs, odb, upload_odb, callback=None): - from dvc_objects.fs.callbacks import Callback - from dvc_objects.fs.utils import tmp_fname - from dvc_objects.hash import HashStreamFile - - fs_path = upload_odb.fs.path - tmp_info = fs_path.join(upload_odb.fs_path, tmp_fname()) - with fs.open(from_fs_path, mode="rb") as stream: - stream = HashStreamFile(stream) - size = fs.size(from_fs_path) - with Callback.as_tqdm_callback( - callback, - desc=fs_path.name(from_fs_path), - bytes=True, - size=size, - ) as cb: - upload_odb.fs.put_file(stream, tmp_info, size=size, callback=cb) - - hash_info = HashInfo(stream.hash_name, stream.hash_value) - odb.add(tmp_info, upload_odb.fs, hash_info) - meta = Meta(size=size) - return from_fs_path, meta, odb.get(hash_info) - - -def _stage_file(fs_path, fs, name, odb=None, upload_odb=None, dry_run=False): - state = odb.state if odb else None - meta, hash_info = hash_file(fs_path, fs, name, state=state) - if upload_odb and not dry_run: - assert odb and name == "md5" - return _upload_file(fs_path, fs, odb, upload_odb) - - if dry_run: - obj = HashFile(fs_path, fs, hash_info) - else: - odb.add(fs_path, fs, hash_info, hardlink=False) - obj = odb.get(hash_info) - - return fs_path, meta, obj - - -def _build_objects( - fs_path, - fs, - name, - ignore: "Ignore" = None, - jobs=None, - no_progress_bar=False, - **kwargs, -): - if ignore: - walk_iterator = ignore.find(fs, fs_path) - else: - walk_iterator = fs.find(fs_path) - with Tqdm( - unit="md5", - desc="Computing file/dir hashes (only done once)", - disable=no_progress_bar, - ) as pbar: - worker = pbar.wrap_fn( - partial( - _stage_file, - fs=fs, - name=name, - **kwargs, - ) - ) - with ThreadPoolExecutor( - max_workers=jobs if jobs is not None else fs.hash_jobs - ) as executor: - yield from executor.map(worker, walk_iterator) - - -def _iter_objects(fs_path, fs, name, **kwargs): - yield from _build_objects(fs_path, fs, name, **kwargs) - - -def _build_tree(fs_path, fs, name, **kwargs): - from .tree import Tree - - tree_meta = Meta(size=0, nfiles=0) - # assuring mypy that they are not None but integer - assert tree_meta.size is not None - assert tree_meta.nfiles is not None - - tree = Tree(None, None, None) - for file_fs_path, meta, obj in _iter_objects(fs_path, fs, name, **kwargs): - if fs.path.name(file_fs_path) == DefaultIgnoreFile: - raise IgnoreInCollectedDirError( - DefaultIgnoreFile, fs.path.parent(file_fs_path) - ) - - # NOTE: this is lossy transformation: - # "hey\there" -> "hey/there" - # "hey/there" -> "hey/there" - # The latter is fine filename on Windows, which - # will transform to dir/file on back transform. - # - # Yes, this is a BUG, as long as we permit "/" in - # filenames on Windows and "\" on Unix - - key = fs.path.relparts(file_fs_path, fs_path) - assert key - tree.add(key, meta, obj.hash_info) - - tree_meta.size += meta.size or 0 - tree_meta.nfiles += 1 - - return tree_meta, tree - - -def _stage_tree(fs_path, fs, fs_info, name, odb=None, **kwargs): - from .tree import Tree - - value = fs_info.get(name) - if odb and value: - hash_info = HashInfo(name, value) - try: - tree = Tree.load(odb, hash_info) - return Meta(nfiles=len(tree)), tree - except FileNotFoundError: - pass - - meta, tree = _build_tree(fs_path, fs, name, odb=odb, **kwargs) - state = odb.state if odb and odb.state else None - hash_info = None - if state: - _, hash_info = state.get( # pylint: disable=assignment-from-none - fs_path, fs - ) - tree.digest(hash_info=hash_info) - odb.add(tree.fs_path, tree.fs, tree.hash_info, hardlink=False) - raw = odb.get(tree.hash_info) - # cleanup unneeded memfs tmpfile and return tree based on the - # ODB fs/path - if odb.fs != tree.fs: - tree.fs.remove(tree.fs_path) - tree.fs = raw.fs - tree.fs_path = raw.fs_path - return meta, tree - - -_url_cache: Dict[str, str] = {} - - -def _make_staging_url( - fs: "FileSystem", odb: "ObjectDB", fs_path: Optional[str] -): - from dvc_objects.fs import Schemes - - url = f"{Schemes.MEMORY}://{_STAGING_MEMFS_PATH}" - - if fs_path is not None: - if odb.fs.protocol == Schemes.LOCAL: - fs_path = os.path.abspath(fs_path) - - if fs_path not in _url_cache: - _url_cache[fs_path] = hashlib.sha256( - fs_path.encode("utf-8") - ).hexdigest() - - url = fs.path.join(url, _url_cache[fs_path]) - - return url - - -def _get_staging(odb: "ObjectDB") -> "ReferenceObjectDB": - """Return an ODB that can be used for staging objects. - - Staging will be a reference ODB stored in the the global memfs. - """ - - from dvc_objects.fs import MemoryFileSystem - - fs = MemoryFileSystem() - fs_path = _make_staging_url(fs, odb, odb.fs_path) - state = odb.state - return ReferenceObjectDB(fs, fs_path, state=state) - - -def _load_raw_dir_obj(odb: "ObjectDB", hash_info: "HashInfo") -> "Tree": - from dvc_objects.errors import ObjectFormatError - - from .tree import Tree - - try: - tree = Tree.load(odb, hash_info.as_raw()) - tree.check(odb) - tree.hash_info = hash_info - except ObjectFormatError as exc: - raise FileNotFoundError( - errno.ENOENT, - "No such object", - odb.hash_to_path(hash_info.as_raw().value), - ) from exc - - return tree - - -def _load_from_state( - odb: "ObjectDB", - staging: "ReferenceObjectDB", - fs_path: "AnyFSPath", - fs: "FileSystem", - name: str, - dry_run: bool, -) -> Tuple["ObjectDB", "Meta", "HashFile"]: - from dvc_objects.errors import ObjectFormatError - - from . import check, load - from .tree import Tree - - state = odb.state - meta, hash_info = state.get(fs_path, fs) - if not hash_info: - raise FileNotFoundError - - for odb_ in (odb, staging): - if not odb_.exists(hash_info): - continue - - try: - obj = load(odb, hash_info) - check(odb, obj, check_hash=False) - except (ObjectFormatError, FileNotFoundError): - continue - - if isinstance(obj, Tree): - meta.nfiles = len(obj) - assert obj.hash_info.name == name - return odb_, meta, obj - - if not hash_info.isdir: - raise FileNotFoundError - - # Try loading the raw dir object saved by `stage`, see below and #7390 - tree = _load_raw_dir_obj(odb, hash_info) - meta.nfiles = len(tree) - assert tree.hash_info.name == name - - if not dry_run: - assert tree.fs - for key, _, oid in tree: - staging.add( - fs.path.join(fs_path, *key), - fs, - oid, - hardlink=False, - verify=False, - ) - - staging.add( - tree.fs_path, - tree.fs, - hash_info, - hardlink=False, - ) - - raw = staging.get(hash_info) - tree.fs = raw.fs - tree.fs_path = raw.fs_path - - logger.debug("loaded tree '%s' from raw dir obj", tree) - return staging, meta, tree - - -def _stage_external_tree_info(odb, tree, name): - # NOTE: used only for external outputs. Initial reasoning was to be - # able to validate .dir files right in the workspace (e.g. check s3 - # etag), but could be dropped for manual validation with regular md5, - # that would be universal for all clouds. - assert odb and name != "md5" - - odb.add(tree.fs_path, tree.fs, tree.hash_info) - raw = odb.get(tree.hash_info) - _, hash_info = hash_file(raw.fs_path, raw.fs, name, state=odb.state) - tree.fs_path = raw.fs_path - tree.fs = raw.fs - tree.hash_info.name = hash_info.name - tree.hash_info.value = hash_info.value - if not tree.hash_info.value.endswith(".dir"): - tree.hash_info.value += ".dir" - return tree - - -def stage( - odb: "ObjectDB", - fs_path: "AnyFSPath", - fs: "FileSystem", - name: str, - upload: bool = False, - dry_run: bool = False, - **kwargs, -) -> Tuple["ObjectDB", "Meta", "HashFile"]: - """Stage (prepare) objects from the given path for addition to an ODB. - - Returns at tuple of (staging_odb, object) where addition to the ODB can - be completed by transferring the object from staging to the dest ODB. - - If dry_run is True, object hashes will be computed and returned, but file - objects themselves will not be added to the staging ODB (i.e. the resulting - file objects cannot transferred from staging to another ODB). - - If upload is True, files will be uploaded to a temporary path on the dest - ODB filesystem, and staged objects will reference the uploaded path rather - than the original source path. - """ - assert fs_path - # assert protocol(fs_path) == fs.protocol - - details = fs.info(fs_path) - staging = _get_staging(odb) - if odb: - try: - return _load_from_state(odb, staging, fs_path, fs, name, dry_run) - except FileNotFoundError: - pass - - if details["type"] == "directory": - meta, obj = _stage_tree( - fs_path, - fs, - details, - name, - odb=staging, - upload_odb=odb if upload else None, - dry_run=dry_run, - **kwargs, - ) - logger.debug("staged tree '%s'", obj) - if name != "md5": - obj = _stage_external_tree_info(odb, obj, name) - - # In order to avoid re-building the tree when it is not committed to - # the local odb (e.g. for a status call), we save it as a raw object. - # Loading this instead of building the tree can speed up `dvc status` - # for modified directories, see #7390 - odb.add(obj.fs_path, obj.fs, obj.hash_info.as_raw()) - else: - _, meta, obj = _stage_file( - fs_path, - fs, - name, - odb=staging, - upload_odb=odb if upload else None, - dry_run=dry_run, - ) - - if odb and odb.state and obj.hash_info: - odb.state.save(fs_path, fs, obj.hash_info) - - return staging, meta, obj diff --git a/dvc/data/status.py b/dvc/data/status.py deleted file mode 100644 index 1e936949e..000000000 --- a/dvc/data/status.py +++ /dev/null @@ -1,196 +0,0 @@ -import logging -from typing import TYPE_CHECKING, Dict, Iterable, NamedTuple, Optional, Set - -from dvc_objects.fs import Schemes -from dvc_objects.hash_info import HashInfo - -from .tree import Tree - -if TYPE_CHECKING: - from dvc_objects.db import ObjectDB - from dvc_objects.file import HashFile - - from .db.index import ObjectDBIndexBase - -logger = logging.getLogger(__name__) - - -class StatusResult(NamedTuple): - exists: Set["HashInfo"] - missing: Set["HashInfo"] - - -class CompareStatusResult(NamedTuple): - ok: Set["HashInfo"] - missing: Set["HashInfo"] - new: Set["HashInfo"] - deleted: Set["HashInfo"] - - -def _indexed_dir_hashes(odb, index, dir_objs, name, cache_odb, jobs=None): - # Validate our index by verifying all indexed .dir hashes - # still exist on the remote - from ._progress import QueryingProgress - - dir_hashes = set(dir_objs.keys()) - indexed_dirs = set(index.dir_hashes()) - indexed_dir_exists = set() - if indexed_dirs: - hashes = QueryingProgress( - odb.list_hashes_exists(indexed_dirs, jobs=jobs), - total=len(indexed_dirs), - ) - indexed_dir_exists.update(hashes) - missing_dirs = indexed_dirs.difference(indexed_dir_exists) - if missing_dirs: - logger.debug( - "Remote cache missing indexed .dir hashes '%s', " - "clearing remote index", - ", ".join(missing_dirs), - ) - index.clear() - - # Check if non-indexed (new) dir hashes exist on remote - dir_exists = dir_hashes.intersection(indexed_dir_exists) - dir_missing = dir_hashes - dir_exists - dir_exists.update( - QueryingProgress( - odb.list_hashes_exists(dir_missing, jobs=jobs), - total=len(dir_missing), - ) - ) - - # If .dir hash exists in the ODB, assume directory contents - # also exists - for dir_hash in dir_exists: - tree = dir_objs.get(dir_hash) - if not tree: - try: - tree = Tree.load(cache_odb, HashInfo(name, dir_hash)) - except FileNotFoundError: - continue - file_hashes = [oid.value for _, _, oid in tree] - if dir_hash not in index: - logger.debug( - "Indexing new .dir '%s' with '%s' nested files", - dir_hash, - len(file_hashes), - ) - index.update([dir_hash], file_hashes) - yield from file_hashes - yield tree.hash_info.value - - -def status( - odb: "ObjectDB", - obj_ids: Iterable["HashInfo"], - name: Optional[str] = None, - index: Optional["ObjectDBIndexBase"] = None, - cache_odb: Optional["ObjectDB"] = None, - shallow: bool = True, - jobs: Optional[int] = None, - **kwargs, -) -> "StatusResult": - """Return status of whether or not the specified objects exist odb. - - If cache_odb is set, trees will be loaded from cache_odb instead of odb - when needed. - - Status is returned as a tuple of: - exists: objs that exist in odb - missing: objs that do not exist in ODB - """ - logger.debug("Preparing to collect status from '%s'", odb.fs_path) - if not name: - name = odb.fs.PARAM_CHECKSUM - - if cache_odb is None: - cache_odb = odb - - hash_infos: Dict[str, "HashInfo"] = {} - dir_objs: Dict[str, Optional["HashFile"]] = {} - for hash_info in obj_ids: - assert hash_info.value - if hash_info.isdir: - if shallow: - tree = None - else: - tree = Tree.load(cache_odb, hash_info) - for _, _, oid in tree: - assert oid and oid.value - hash_infos[oid.value] = oid - if index: - dir_objs[hash_info.value] = tree - hash_infos[hash_info.value] = hash_info - - if odb.fs.protocol == Schemes.MEMORY: - # assume memfs staged objects already exist - return StatusResult(set(hash_infos.values()), set()) - - hashes: Set[str] = set(hash_infos.keys()) - exists: Set[str] = set() - - logger.debug("Collecting status from '%s'", odb.fs_path) - if index and hashes: - if dir_objs: - exists = hashes.intersection( - _indexed_dir_hashes( - odb, index, dir_objs, name, cache_odb, jobs=jobs - ) - ) - hashes.difference_update(exists) - if hashes: - exists.update(index.intersection(hashes)) - hashes.difference_update(exists) - - if hashes: - from ._progress import QueryingProgress - - with QueryingProgress(phase="Checking", name=odb.fs_path) as pbar: - exists.update( - odb.hashes_exist(hashes, jobs=jobs, progress=pbar.callback) - ) - return StatusResult( - {hash_infos[hash_] for hash_ in exists}, - {hash_infos[hash_] for hash_ in (hashes - exists)}, - ) - - -def compare_status( - src: "ObjectDB", - dest: "ObjectDB", - obj_ids: Iterable["HashInfo"], - check_deleted: bool = True, - src_index: Optional["ObjectDBIndexBase"] = None, - dest_index: Optional["ObjectDBIndexBase"] = None, - jobs: Optional[int] = None, - **kwargs, -) -> "CompareStatusResult": - """Compare status for the specified objects between two ODBs. - - Status is returned as a tuple of: - ok: hashes that exist in both src and dest - missing: hashes that do not exist in neither src nor dest - new: hashes that only exist in src - deleted: hashes that only exist in dest - """ - if "cache_odb" not in kwargs: - kwargs["cache_odb"] = src - dest_exists, dest_missing = status( - dest, obj_ids, index=dest_index, jobs=jobs, **kwargs - ) - # for transfer operations we can skip src status check when all objects - # already exist in dest - if dest_missing or check_deleted: - src_exists, src_missing = status( - src, obj_ids, index=src_index, jobs=jobs, **kwargs - ) - else: - src_exists = dest_exists - src_missing = set() - return CompareStatusResult( - src_exists & dest_exists, - src_missing & dest_missing, - src_exists - dest_exists, - dest_exists - src_exists, - ) diff --git a/dvc/data/transfer.py b/dvc/data/transfer.py deleted file mode 100644 index 52db9ab16..000000000 --- a/dvc/data/transfer.py +++ /dev/null @@ -1,189 +0,0 @@ -import errno -import logging -from functools import partial, wraps -from typing import TYPE_CHECKING, Any, Callable, Iterable, Optional - -from funcy import split - -from dvc_objects._tqdm import Tqdm -from dvc_objects.executors import ThreadPoolExecutor - -if TYPE_CHECKING: - from dvc_objects.db import ObjectDB - from dvc_objects.hash_info import HashInfo - - from .db.index import ObjectDBIndexBase - from .status import CompareStatusResult - from .tree import Tree - -logger = logging.getLogger(__name__) - - -class TransferError(Exception): - def __init__(self, fails: int) -> None: - self.fails = fails - super().__init__(f"{fails} transfer failed") - - -def _log_exceptions(func): - @wraps(func) - def wrapper(fs_path, *args, **kwargs): - try: - func(fs_path, *args, **kwargs) - return 0 - except Exception as exc: # pylint: disable=broad-except - # NOTE: this means we ran out of file descriptors and there is no - # reason to try to proceed, as we will hit this error anyways. - # pylint: disable=no-member - if isinstance(exc, OSError) and exc.errno == errno.EMFILE: - raise - - logger.debug("failed to transfer '%s'", fs_path) - return 1 - - return wrapper - - -def find_tree_by_obj_id( - odbs: Iterable[Optional["ObjectDB"]], obj_id: "HashInfo" -) -> Optional["Tree"]: - from dvc_objects.errors import ObjectFormatError - - from .tree import Tree - - for odb in odbs: - if odb is not None: - try: - return Tree.load(odb, obj_id) - except (FileNotFoundError, ObjectFormatError): - pass - return None - - -def _do_transfer( - src: "ObjectDB", - dest: "ObjectDB", - obj_ids: Iterable["HashInfo"], - missing_ids: Iterable["HashInfo"], - processor: Callable, - src_index: Optional["ObjectDBIndexBase"] = None, - dest_index: Optional["ObjectDBIndexBase"] = None, - cache_odb: Optional["ObjectDB"] = None, - **kwargs: Any, -): - dir_ids, file_ids = split(lambda hash_info: hash_info.isdir, obj_ids) - total_fails = 0 - succeeded_dir_objs = [] - all_file_ids = set(file_ids) - - for dir_hash in dir_ids: - dir_obj = find_tree_by_obj_id([cache_odb, src], dir_hash) - assert dir_obj - - entry_ids = {oid for _, _, oid in dir_obj} - bound_file_ids = all_file_ids & entry_ids - all_file_ids -= entry_ids - - dir_fails = sum(processor(bound_file_ids)) - if dir_fails: - logger.debug( - "failed to upload full contents of '%s', " - "aborting .dir file upload", - dir_obj.name, - ) - logger.debug( - "failed to upload '%s' to '%s'", - src.get(dir_obj.hash_info).fs_path, - dest.get(dir_obj.hash_info).fs_path, - ) - total_fails += dir_fails + 1 - elif entry_ids.intersection(missing_ids): - # if for some reason a file contained in this dir is - # missing both locally and in the remote, we want to - # push whatever file content we have, but should not - # push .dir file - logger.debug( - "directory '%s' contains missing files," - "skipping .dir file upload", - dir_obj.name, - ) - else: - is_dir_failed = sum(processor([dir_obj.hash_info])) - total_fails += is_dir_failed - if not is_dir_failed: - succeeded_dir_objs.append(dir_obj) - - # insert the rest - total_fails += sum(processor(all_file_ids)) - if total_fails: - if src_index: - src_index.clear() - raise TransferError(total_fails) - - # index successfully pushed dirs - if dest_index: - for dir_obj in succeeded_dir_objs: - file_hashes = {oid.value for _, _, oid in dir_obj} - logger.debug( - "Indexing pushed dir '%s' with '%s' nested files", - dir_obj.hash_info, - len(file_hashes), - ) - assert dir_obj.hash_info and dir_obj.hash_info.value - dest_index.update([dir_obj.hash_info.value], file_hashes) - - -def transfer( - src: "ObjectDB", - dest: "ObjectDB", - obj_ids: Iterable["HashInfo"], - jobs: Optional[int] = None, - verify: bool = False, - hardlink: bool = False, - validate_status: Callable[["CompareStatusResult"], None] = None, - **kwargs, -) -> int: - """Transfer (copy) the specified objects from one ODB to another. - - Returns the number of successfully transferred objects - """ - from .status import compare_status - - logger.debug( - "Preparing to transfer data from '%s' to '%s'", - src.fs_path, - dest.fs_path, - ) - if src == dest: - return 0 - - status = compare_status( - src, dest, obj_ids, check_deleted=False, jobs=jobs, **kwargs - ) - - if validate_status: - validate_status(status) - - if not status.new: - return 0 - - def func(hash_info: "HashInfo") -> None: - obj = src.get(hash_info) - return dest.add( - obj.fs_path, - obj.fs, - obj.hash_info, - verify=verify, - hardlink=hardlink, - ) - - total = len(status.new) - jobs = jobs or dest.fs.jobs - with Tqdm(total=total, unit="file", desc="Transferring") as pbar: - with ThreadPoolExecutor(max_workers=jobs) as executor: - wrapped_func = pbar.wrap_fn(_log_exceptions(func)) - processor = partial(executor.imap_unordered, wrapped_func) - _do_transfer( - src, dest, status.new, status.missing, processor, **kwargs - ) - return total diff --git a/dvc/data/tree.py b/dvc/data/tree.py deleted file mode 100644 index e294eb71d..000000000 --- a/dvc/data/tree.py +++ /dev/null @@ -1,293 +0,0 @@ -import json -import logging -import posixpath -from typing import TYPE_CHECKING, Dict, Iterable, Optional, Tuple - -from funcy import cached_property - -from dvc_objects.errors import ObjectFormatError -from dvc_objects.file import HashFile -from dvc_objects.hash import hash_file - -if TYPE_CHECKING: - from dvc_objects.db import ObjectDB - from dvc_objects.hash_info import HashInfo - from dvc_objects.meta import Meta - -logger = logging.getLogger(__name__) - - -class TreeError(Exception): - pass - - -class MergeError(Exception): - pass - - -def _try_load( - odbs: Iterable["ObjectDB"], - hash_info: "HashInfo", -) -> Optional["HashFile"]: - for odb in odbs: - if not odb: - continue - - try: - return Tree.load(odb, hash_info) - except (FileNotFoundError, ObjectFormatError): - pass - - return None - - -class Tree(HashFile): - PARAM_RELPATH = "relpath" - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self._dict: Dict[Tuple[str], Tuple["Meta", "HashInfo"]] = {} - - @cached_property - def _trie(self): - from pygtrie import Trie - - return Trie(self._dict) - - def add(self, key: Tuple[str], meta: "Meta", oid: "HashInfo"): - self.__dict__.pop("trie", None) - self._dict[key] = (meta, oid) - - def digest(self, hash_info: Optional["HashInfo"] = None): - from dvc_objects.fs import MemoryFileSystem - from dvc_objects.fs.utils import tmp_fname - - memfs = MemoryFileSystem() - fs_path = "memory://{}".format(tmp_fname("")) - memfs.pipe_file(fs_path, self.as_bytes()) - self.fs = memfs - self.fs_path = fs_path - if hash_info: - self.hash_info = hash_info - else: - _, self.hash_info = hash_file(fs_path, memfs, "md5") - assert self.hash_info.value - self.hash_info.value += ".dir" - - def _load(self, key, meta, hash_info): - if hash_info and hash_info.isdir and not meta.obj: - meta.obj = _try_load([meta.odb, meta.remote], hash_info) - if meta.obj: - for ikey, value in meta.obj.iteritems(): - self._trie[key + ikey] = value - self._dict[key + ikey] = value - - def iteritems(self, prefix=None): - kwargs = {} - if prefix: - kwargs = {"prefix": prefix} - item = self._trie.longest_prefix(prefix) - if item: - key, (meta, hash_info) = item - self._load(key, meta, hash_info) - - for key, (meta, hash_info) in self._trie.iteritems(**kwargs): - self._load(key, meta, hash_info) - yield key, (meta, hash_info) - - def shortest_prefix(self, *args, **kwargs): - return self._trie.shortest_prefix(*args, **kwargs) - - def __len__(self): - return len(self._dict) - - def __iter__(self): - yield from ( - (key, value[0], value[1]) for key, value in self._dict.items() - ) - - def as_dict(self): - return self._dict.copy() - - def as_list(self): - from operator import itemgetter - - # Sorting the list by path to ensure reproducibility - return sorted( - ( - { - # NOTE: not using hash_info.to_dict() because we don't want - # size/nfiles fields at this point. - oid.name: oid.value, - self.PARAM_RELPATH: posixpath.sep.join(parts), - } - for parts, _, oid in self # noqa: B301 - ), - key=itemgetter(self.PARAM_RELPATH), - ) - - def as_bytes(self): - return json.dumps(self.as_list(), sort_keys=True).encode("utf-8") - - @classmethod - def from_list(cls, lst): - from dvc_objects.hash_info import HashInfo - - tree = cls(None, None, None) - for _entry in lst: - entry = _entry.copy() - relpath = entry.pop(cls.PARAM_RELPATH) - parts = tuple(relpath.split(posixpath.sep)) - hash_info = HashInfo.from_dict(entry) - tree.add(parts, None, hash_info) - return tree - - @classmethod - def load(cls, odb, hash_info) -> "Tree": - obj = odb.get(hash_info) - - try: - with obj.fs.open(obj.fs_path, "r") as fobj: - raw = json.load(fobj) - except ValueError as exc: - raise ObjectFormatError(f"{obj} is corrupted") from exc - - if not isinstance(raw, list): - logger.debug( - "dir cache file format error '%s' [skipping the file]", - obj.fs_path, - ) - raise ObjectFormatError(f"{obj} is corrupted") - - tree = cls.from_list(raw) - tree.fs_path = obj.fs_path - tree.fs = obj.fs - tree.hash_info = hash_info - - return tree - - def filter(self, prefix: Tuple[str]) -> Optional["Tree"]: - """Return a filtered copy of this tree that only contains entries - inside prefix. - - The returned tree will contain the original tree's hash_info and - fs_path. - - Returns an empty tree if no object exists at the specified prefix. - """ - tree = Tree(self.fs_path, self.fs, self.hash_info) - try: - for key, (meta, oid) in self._trie.items(prefix): - tree.add(key, meta, oid) - except KeyError: - pass - return tree - - def get(self, odb, prefix: Tuple[str]) -> Optional[HashFile]: - """Return object at the specified prefix in this tree. - - Returns None if no object exists at the specified prefix. - """ - _, oid = self._dict.get(prefix) or (None, None) - if oid: - return odb.get(oid) - - tree = Tree(None, None, None) - depth = len(prefix) - try: - for key, (meta, entry_oid) in self._trie.items(prefix): - tree.add(key[depth:], meta, entry_oid) - except KeyError: - return None - tree.digest() - return tree - - def ls(self, prefix=None): - kwargs = {} - if prefix: - kwargs["prefix"] = prefix - - meta, hash_info = self._trie.get(prefix, (None, None)) - if hash_info and hash_info.isdir and meta and not meta.obj: - raise TreeError - - ret = [] - - def node_factory(_, key, children, *args): - if key == prefix: - list(children) - else: - ret.append(key[-1]) - - self._trie.traverse(node_factory, **kwargs) - - return ret - - -def du(odb, tree): - try: - return sum( - odb.fs.size(odb.hash_to_path(oid.value)) for _, _, oid in tree - ) - except FileNotFoundError: - return None - - -def _diff(ancestor, other, allow_removed=False): - from dictdiffer import diff - - allowed = ["add"] - if allow_removed: - allowed.append("remove") - - result = list(diff(ancestor, other)) - for typ, _, _ in result: - if typ not in allowed: - raise MergeError( - "unable to auto-merge directories with diff that contains " - f"'{typ}'ed files" - ) - return result - - -def _merge(ancestor, our, their): - import copy - - from dictdiffer import patch - - our_diff = _diff(ancestor, our) - if not our_diff: - return copy.deepcopy(their) - - their_diff = _diff(ancestor, their) - if not their_diff: - return copy.deepcopy(our) - - # make sure there are no conflicting files - _diff(our, their, allow_removed=True) - - return patch(our_diff + their_diff, ancestor) - - -def merge(odb, ancestor_info, our_info, their_info): - from . import load - - assert our_info - assert their_info - - if ancestor_info: - ancestor = load(odb, ancestor_info) - else: - ancestor = Tree(None, None, None) - - our = load(odb, our_info) - their = load(odb, their_info) - - merged_dict = _merge(ancestor.as_dict(), our.as_dict(), their.as_dict()) - - merged = Tree(None, None, None) - for key, (meta, oid) in merged_dict.items(): - merged.add(key, meta, oid) - merged.digest() - - return merged diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 56c89babd..b5ee7bede 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -3,10 +3,10 @@ import logging from typing import TYPE_CHECKING, Iterable, Optional -from dvc.data.db import get_index +from dvc_data.db import get_index if TYPE_CHECKING: - from dvc.data.status import CompareStatusResult + from dvc_data.status import CompareStatusResult from dvc_objects.db import ObjectDB from dvc_objects.hash_info import HashInfo @@ -56,8 +56,8 @@ class DataCloud: raise NoRemoteError(error_msg) def _init_odb(self, name): - from dvc.data.db import get_odb from dvc.fs import get_cloud_fs + from dvc_data.db import get_odb cls, config, fs_path = get_cloud_fs(self.repo, name=name) config["tmp_dir"] = self.repo.index_db_dir @@ -81,8 +81,8 @@ class DataCloud: objs: Iterable["HashInfo"], **kwargs, ): - from dvc.data.transfer import TransferError, transfer from dvc.exceptions import FileTransferError + from dvc_data.transfer import TransferError, transfer try: return transfer(src_odb, dest_odb, objs, **kwargs) @@ -164,7 +164,7 @@ class DataCloud: log_missing: log warning messages if file doesn't exist neither in cache, neither in cloud. """ - from dvc.data.status import compare_status + from dvc_data.status import compare_status if not odb: odb = self.get_remote_odb(remote, "status") diff --git a/dvc/dependency/repo.py b/dvc/dependency/repo.py index 98ebd45d7..f8fda63f3 100644 --- a/dvc/dependency/repo.py +++ b/dvc/dependency/repo.py @@ -63,7 +63,7 @@ class RepoDependency(Dependency): return {self.PARAM_PATH: self.def_path, self.PARAM_REPO: self.def_repo} def download(self, to, jobs=None): - from dvc.data.checkout import checkout + from dvc_data.checkout import checkout for odb, objs in self.get_used_objs().items(): self.repo.cloud.pull(objs, jobs=jobs, odb=odb) @@ -101,10 +101,10 @@ class RepoDependency(Dependency): self, obj_only=False, **kwargs ) -> Tuple[Dict[Optional["ObjectDB"], Set["HashInfo"]], "HashFile"]: from dvc.config import NoRemoteError - from dvc.data.stage import stage - from dvc.data.tree import Tree, TreeError from dvc.exceptions import NoOutputOrStageError, PathMissingError from dvc.utils import as_posix + from dvc_data.stage import stage + from dvc_data.tree import Tree, TreeError local_odb = self.repo.odb.local locked = kwargs.pop("locked", True) @@ -153,10 +153,10 @@ class RepoDependency(Dependency): return used_obj_ids, staged_obj def _check_circular_import(self, odb, obj_ids): - from dvc.data.db.reference import ReferenceObjectDB - from dvc.data.tree import Tree from dvc.exceptions import CircularImportError from dvc.fs.dvc import DvcFileSystem + from dvc_data.db.reference import ReferenceObjectDB + from dvc_data.tree import Tree if not isinstance(odb, ReferenceObjectDB): return diff --git a/dvc/odbmgr.py b/dvc/odbmgr.py index 9cc82cc83..df4b2fb84 100644 --- a/dvc/odbmgr.py +++ b/dvc/odbmgr.py @@ -1,5 +1,5 @@ -from dvc.data.db import get_odb from dvc.fs import Schemes +from dvc_data.db import get_odb def _get_odb(repo, settings): diff --git a/dvc/output.py b/dvc/output.py index b6f44e138..4a475ec03 100644 --- a/dvc/output.py +++ b/dvc/output.py @@ -17,17 +17,17 @@ from dvc.exceptions import ( MergeError, RemoteCacheRequiredError, ) +from dvc_data import check as ocheck +from dvc_data import load as oload +from dvc_data.checkout import checkout +from dvc_data.stage import stage as ostage +from dvc_data.transfer import transfer as otransfer +from dvc_data.tree import Tree from dvc_objects.errors import ObjectFormatError from dvc_objects.hash_info import HashInfo from dvc_objects.istextfile import istextfile from dvc_objects.meta import Meta -from .data import check as ocheck -from .data import load as oload -from .data.checkout import checkout -from .data.stage import stage as ostage -from .data.transfer import transfer as otransfer -from .data.tree import Tree from .fs import ( HDFSFileSystem, LocalFileSystem, @@ -580,8 +580,8 @@ class Output: self.odb.set_exec(self.fs_path) def _checkout(self, *args, **kwargs): - from dvc.data.checkout import CheckoutError as _CheckoutError - from dvc.data.checkout import LinkError, PromptError + from dvc_data.checkout import CheckoutError as _CheckoutError + from dvc_data.checkout import LinkError, PromptError try: return checkout(*args, **kwargs) @@ -1044,8 +1044,8 @@ class Output: ) def merge(self, ancestor, other): - from dvc.data.tree import MergeError as TreeMergeError - from dvc.data.tree import du, merge + from dvc_data.tree import MergeError as TreeMergeError + from dvc_data.tree import du, merge assert other diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py index c70974550..94c865c25 100644 --- a/dvc/repo/__init__.py +++ b/dvc/repo/__init__.py @@ -414,7 +414,7 @@ class Repo: def _add_suffix(objs: Set["HashFile"], suffix: str) -> None: from itertools import chain - from dvc.data import iterobjs + from dvc_data import iterobjs for obj in chain.from_iterable(map(iterobjs, objs)): if obj.name is not None: diff --git a/dvc/repo/diff.py b/dvc/repo/diff.py index 071764758..a78ec6360 100644 --- a/dvc/repo/diff.py +++ b/dvc/repo/diff.py @@ -111,8 +111,8 @@ def _paths_checksums(repo, targets): def _output_paths(repo, targets): - from dvc.data.stage import stage as ostage from dvc.fs import LocalFileSystem + from dvc_data.stage import stage as ostage on_working_fs = isinstance(repo.fs, LocalFileSystem) diff --git a/dvc/repo/gc.py b/dvc/repo/gc.py index c3fe871bd..23ae22e73 100644 --- a/dvc/repo/gc.py +++ b/dvc/repo/gc.py @@ -44,9 +44,9 @@ def gc( from contextlib import ExitStack - from dvc.data.db import get_index - from dvc.data.gc import gc as ogc from dvc.repo import Repo + from dvc_data.db import get_index + from dvc_data.gc import gc as ogc if not repos: repos = [] diff --git a/dvc/repo/index.py b/dvc/repo/index.py index 0fde54c34..79613c999 100644 --- a/dvc/repo/index.py +++ b/dvc/repo/index.py @@ -19,13 +19,13 @@ if TYPE_CHECKING: from networkx import DiGraph from pygtrie import Trie - from dvc.data.tree import Tree from dvc.dependency import Dependency, ParamsDependency from dvc.fs import FileSystem from dvc.output import Output from dvc.repo.stage import StageLoad from dvc.stage import Stage from dvc.types import StrPath, TargetType + from dvc_data.tree import Tree from dvc_objects.db import ObjectDB from dvc_objects.hash_info import HashInfo @@ -170,7 +170,7 @@ class Index: @cached_property def tree(self) -> "Tree": from dvc.config import NoRemoteError - from dvc.data.tree import Tree + from dvc_data.tree import Tree tree = Tree(None, None, None) diff --git a/dvc/stage/cache.py b/dvc/stage/cache.py index 00b7585ff..e155275e7 100644 --- a/dvc/stage/cache.py +++ b/dvc/stage/cache.py @@ -7,9 +7,9 @@ from typing import TYPE_CHECKING, Optional from funcy import cached_property, first from dvc import fs -from dvc.data.transfer import _log_exceptions from dvc.exceptions import DvcException from dvc.utils import dict_sha256, relpath +from dvc_data.transfer import _log_exceptions if TYPE_CHECKING: from dvc_objects.db import ObjectDB diff --git a/pyproject.toml b/pyproject.toml index 8c043d226..dfe0de9f4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,7 +25,7 @@ exclude = ''' [tool.isort] profile = "black" -known_first_party = ["dvc", "dvc_objects", "tests"] +known_first_party = ["dvc", "dvc_data", "dvc_objects", "tests"] line_length = 79 [tool.pytest.ini_options] diff --git a/tests/func/data/db/test_index.py b/tests/func/data/db/test_index.py index 886920e03..d45db5973 100644 --- a/tests/func/data/db/test_index.py +++ b/tests/func/data/db/test_index.py @@ -1,9 +1,9 @@ import pytest -from dvc.data.db import get_index from dvc.exceptions import DownloadError, UploadError from dvc.fs import LocalFileSystem from dvc.utils.fs import remove +from dvc_data.db import get_index from tests.utils import clean_staging diff --git a/tests/func/test_add.py b/tests/func/test_add.py index c6860a669..3846e5fc5 100644 --- a/tests/func/test_add.py +++ b/tests/func/test_add.py @@ -11,6 +11,7 @@ import colorama import pytest import dvc as dvc_module +import dvc_data import dvc_objects from dvc.cli import main from dvc.dvcfile import DVC_FILE_SUFFIX @@ -103,7 +104,7 @@ def test_add_unsupported_file(dvc): def test_add_directory(tmp_dir, dvc): - from dvc.data import load + from dvc_data import load (stage,) = tmp_dir.dvc_gen({"dir": {"file": "file"}}) @@ -457,7 +458,7 @@ class TestAddCommit(TestDvc): def test_should_collect_dir_cache_only_once(mocker, tmp_dir, dvc): tmp_dir.gen({"data/data": "foo"}) - counter = mocker.spy(dvc_module.data.stage, "_stage_tree") + counter = mocker.spy(dvc_data.stage, "_stage_tree") ret = main(["add", "data"]) assert ret == 0 assert counter.mock.call_count == 1 @@ -881,7 +882,7 @@ def test_add_with_cache_link_error(tmp_dir, dvc, mocker, capsys): tmp_dir.gen("foo", "foo") mocker.patch( - "dvc.data.checkout.test_links", + "dvc_data.checkout.test_links", return_value=[], ) dvc.add("foo") diff --git a/tests/func/test_checkout.py b/tests/func/test_checkout.py index e9e7994ba..42417aca1 100644 --- a/tests/func/test_checkout.py +++ b/tests/func/test_checkout.py @@ -84,7 +84,7 @@ class TestCheckoutCorruptedCacheFile(TestRepro): class TestCheckoutCorruptedCacheDir(TestDvc): def test(self): - from dvc.data import load + from dvc_data import load # NOTE: using 'copy' so that cache and link don't have same inode ret = main(["config", "cache.type", "copy"]) diff --git a/tests/func/test_data_cloud.py b/tests/func/test_data_cloud.py index e71a957e7..3a827ab23 100644 --- a/tests/func/test_data_cloud.py +++ b/tests/func/test_data_cloud.py @@ -7,13 +7,13 @@ from flaky.flaky_decorator import flaky import dvc_objects from dvc.cli import main -from dvc.data.db.local import LocalObjectDB from dvc.external_repo import clean_repos from dvc.stage.exceptions import StageNotFound from dvc.testing.test_remote import ( # noqa, pylint: disable=unused-import TestRemote, ) from dvc.utils.fs import remove +from dvc_data.db.local import LocalObjectDB from dvc_objects.db import ObjectDB diff --git a/tests/func/test_external_repo.py b/tests/func/test_external_repo.py index 6207df293..15b811c0f 100644 --- a/tests/func/test_external_repo.py +++ b/tests/func/test_external_repo.py @@ -3,11 +3,11 @@ from unittest.mock import ANY from scmrepo.git import Git -from dvc.data.stage import stage -from dvc.data.transfer import transfer from dvc.external_repo import CLONES, external_repo from dvc.utils import relpath from dvc.utils.fs import makedirs, remove +from dvc_data.stage import stage +from dvc_data.transfer import transfer from tests.unit.fs.test_dvc import make_subrepo from tests.utils import clean_staging diff --git a/tests/func/test_gc.py b/tests/func/test_gc.py index a614ca9b4..ce5e40592 100644 --- a/tests/func/test_gc.py +++ b/tests/func/test_gc.py @@ -7,11 +7,11 @@ import pytest from git import Repo from dvc.cli import main -from dvc.data.db.local import LocalObjectDB from dvc.exceptions import CollectCacheError from dvc.fs import LocalFileSystem from dvc.repo import Repo as DvcRepo from dvc.utils.fs import remove +from dvc_data.db.local import LocalObjectDB from tests.basic_env import TestDir, TestDvcGit diff --git a/tests/func/test_ignore.py b/tests/func/test_ignore.py index ce59a5b97..97992c740 100644 --- a/tests/func/test_ignore.py +++ b/tests/func/test_ignore.py @@ -4,13 +4,13 @@ from pathlib import Path import pytest -from dvc.data.stage import IgnoreInCollectedDirError from dvc.ignore import DvcIgnore, DvcIgnorePatterns from dvc.output import OutputIsIgnoredError from dvc.pathspec_math import PatternInfo, merge_patterns from dvc.repo import Repo from dvc.testing.tmp_dir import TmpDir from dvc.types import List +from dvc_data.stage import IgnoreInCollectedDirError from dvc_objects.utils import get_mtime_and_size diff --git a/tests/func/test_import.py b/tests/func/test_import.py index ccdedeefd..8f79671fa 100644 --- a/tests/func/test_import.py +++ b/tests/func/test_import.py @@ -529,7 +529,7 @@ def test_import_with_no_exec(tmp_dir, dvc, erepo_dir): def test_import_with_jobs(mocker, dvc, erepo_dir): - import dvc.data.transfer as otransfer + import dvc_data.transfer as otransfer with erepo_dir.chdir(): erepo_dir.dvc_gen( diff --git a/tests/func/test_odb.py b/tests/func/test_odb.py index 5054fd6b3..f32595f02 100644 --- a/tests/func/test_odb.py +++ b/tests/func/test_odb.py @@ -43,7 +43,7 @@ def test_cache(tmp_dir, dvc): def test_cache_load_bad_dir_cache(tmp_dir, dvc): - from dvc.data import load + from dvc_data import load dir_hash = "123.dir" fname = os.fspath(dvc.odb.local.hash_to_path(dir_hash)) diff --git a/tests/func/test_remote.py b/tests/func/test_remote.py index 51ca33a50..4c1c8de20 100644 --- a/tests/func/test_remote.py +++ b/tests/func/test_remote.py @@ -139,8 +139,8 @@ def test_upper_case_remote(tmp_dir, dvc, local_cloud): def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc): - from dvc.data.stage import stage - from dvc.data.tree import Tree + from dvc_data.stage import stage + from dvc_data.tree import Tree tmp_dir.gen({"data": {"1": "1 content", "2": "2 content"}}) @@ -150,7 +150,7 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc): [{"relpath": "1", "md5": "1"}, {"relpath": "2", "md5": "2"}] ) tree.digest() - with patch("dvc.data.stage._stage_tree", return_value=(None, tree)): + with patch("dvc_data.stage._stage_tree", return_value=(None, tree)): _, _, obj = stage(dvc.odb.local, path, dvc.odb.local.fs, "md5") hash1 = obj.hash_info @@ -161,7 +161,7 @@ def test_dir_hash_should_be_key_order_agnostic(tmp_dir, dvc): [{"md5": "1", "relpath": "1"}, {"md5": "2", "relpath": "2"}] ) tree.digest() - with patch("dvc.data.stage._stage_tree", return_value=(None, tree)): + with patch("dvc_data.stage._stage_tree", return_value=(None, tree)): _, _, obj = stage(dvc.odb.local, path, dvc.odb.local.fs, "md5") hash2 = obj.hash_info diff --git a/tests/func/test_stage.py b/tests/func/test_stage.py index b661ea870..560ba0438 100644 --- a/tests/func/test_stage.py +++ b/tests/func/test_stage.py @@ -342,8 +342,8 @@ def test_stage_run_checkpoint(tmp_dir, dvc, mocker, checkpoint): def test_stage_dir_optimization( tmp_dir, dvc, mocker, dry_run, expected_staging_contents ): - from dvc.data import stage - from dvc.data.tree import Tree + from dvc_data import stage + from dvc_data.tree import Tree tmp_dir.dvc_gen( { diff --git a/tests/unit/cli/test_main.py b/tests/unit/cli/test_main.py index 38db3732b..40b753f9d 100644 --- a/tests/unit/cli/test_main.py +++ b/tests/unit/cli/test_main.py @@ -4,7 +4,7 @@ import pytest from funcy import raiser from dvc.cli import main -from dvc.data.stage import IgnoreInCollectedDirError +from dvc_data.stage import IgnoreInCollectedDirError from dvc_objects.cache import DiskError from dvc_objects.fs.base import FileSystem, RemoteMissingDepsError diff --git a/tests/unit/data/db/test_local.py b/tests/unit/data/db/test_local.py index e72ab5208..e96e64454 100644 --- a/tests/unit/data/db/test_local.py +++ b/tests/unit/data/db/test_local.py @@ -3,8 +3,8 @@ import os import pytest -from dvc.data.db.local import LocalObjectDB from dvc.fs import LocalFileSystem +from dvc_data.db.local import LocalObjectDB from dvc_objects.hash_info import HashInfo @@ -13,7 +13,7 @@ def test_status_download_optimization(mocker, dvc): And the desired files to fetch are already on the local cache, Don't check the existence of the desired files on the remote cache """ - from dvc.data.status import compare_status + from dvc_data.status import compare_status odb = LocalObjectDB(LocalFileSystem(), os.getcwd()) obj_ids = { @@ -85,9 +85,9 @@ def test_set_exec_ignore_errors(tmp_dir, dvc, mocker, err): def test_staging_file(tmp_dir, dvc): - from dvc.data import check - from dvc.data.stage import stage - from dvc.data.transfer import transfer + from dvc_data import check + from dvc_data.stage import stage + from dvc_data.transfer import transfer tmp_dir.gen("foo", "foo") fs = LocalFileSystem() @@ -113,9 +113,9 @@ def test_staging_file(tmp_dir, dvc): def test_staging_dir(tmp_dir, dvc): - from dvc.data import check - from dvc.data.stage import stage - from dvc.data.transfer import transfer + from dvc_data import check + from dvc_data.stage import stage + from dvc_data.transfer import transfer tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}}) fs = LocalFileSystem() diff --git a/tests/unit/data/test_tree.py b/tests/unit/data/test_tree.py index 2d8599f45..0c46664a1 100644 --- a/tests/unit/data/test_tree.py +++ b/tests/unit/data/test_tree.py @@ -2,7 +2,7 @@ from operator import itemgetter import pytest -from dvc.data.tree import Tree, _merge +from dvc_data.tree import Tree, _merge from dvc_objects.hash_info import HashInfo from dvc_objects.meta import Meta diff --git a/tests/unit/fs/test_data.py b/tests/unit/fs/test_data.py index 3eaec16af..8ffbcec8a 100644 --- a/tests/unit/fs/test_data.py +++ b/tests/unit/fs/test_data.py @@ -5,9 +5,9 @@ import pytest import dvc_objects from dvc.config import NoRemoteError -from dvc.data.stage import stage from dvc.fs.data import DataFileSystem from dvc.utils.fs import remove +from dvc_data.stage import stage from dvc_objects.hash_info import HashInfo diff --git a/tests/unit/fs/test_dvc.py b/tests/unit/fs/test_dvc.py index 6f8602c50..95f4de33a 100644 --- a/tests/unit/fs/test_dvc.py +++ b/tests/unit/fs/test_dvc.py @@ -5,8 +5,8 @@ from unittest import mock import pytest -from dvc.data.stage import stage from dvc.fs.dvc import DvcFileSystem +from dvc_data.stage import stage from dvc_objects.hash_info import HashInfo from tests.utils import clean_staging @@ -575,7 +575,7 @@ def test_get_hash_mixed_dir(tmp_dir, scm, dvc): def test_get_hash_dirty_file(tmp_dir, dvc): - from dvc.data import check + from dvc_data import check from dvc_objects.errors import ObjectFormatError from dvc_objects.hash import hash_file diff --git a/tests/unit/remote/test_index.py b/tests/unit/remote/test_index.py index ba57c8a75..9c6a541f4 100644 --- a/tests/unit/remote/test_index.py +++ b/tests/unit/remote/test_index.py @@ -3,7 +3,7 @@ import os import pytest from funcy import first -from dvc.data.db.index import ObjectDBIndex +from dvc_data.db.index import ObjectDBIndex @pytest.fixture diff --git a/tests/unit/remote/test_remote.py b/tests/unit/remote/test_remote.py index e9ff919f1..422a34033 100644 --- a/tests/unit/remote/test_remote.py +++ b/tests/unit/remote/test_remote.py @@ -2,8 +2,8 @@ import os import pytest -from dvc.data.db import get_index from dvc.fs import GSFileSystem, S3FileSystem, get_cloud_fs +from dvc_data.db import get_index def test_remote_with_hash_jobs(dvc): diff --git a/tests/utils/__init__.py b/tests/utils/__init__.py index 7befade7a..2575e3604 100644 --- a/tests/utils/__init__.py +++ b/tests/utils/__init__.py @@ -45,8 +45,8 @@ def dump_sv(stream, metrics, delimiter=",", header=True): def clean_staging(): - from dvc.data.stage import _STAGING_MEMFS_PATH from dvc.fs import MemoryFileSystem + from dvc_data.stage import _STAGING_MEMFS_PATH try: MemoryFileSystem().fs.rm(