mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-10 07:18:27 +01:00
dvc: remove dvc.data, fix imports
This commit is contained in:
Saugat Pachhai (सौगात)
committed by
skshetry
skshetry
parent
7bc98e3fb3
commit
9a2feaa4ed
+1
-1
@@ -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("")
|
||||
|
||||
@@ -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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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(),
|
||||
)
|
||||
@@ -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())
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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
|
||||
@@ -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]
|
||||
)
|
||||
),
|
||||
)
|
||||
@@ -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
|
||||
@@ -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,
|
||||
)
|
||||
@@ -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
|
||||
@@ -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
|
||||
+5
-5
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
+1
-1
@@ -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):
|
||||
|
||||
+10
-10
@@ -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
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
+1
-1
@@ -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)
|
||||
|
||||
|
||||
+2
-2
@@ -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 = []
|
||||
|
||||
+2
-2
@@ -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)
|
||||
|
||||
|
||||
+1
-1
@@ -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
|
||||
|
||||
+1
-1
@@ -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]
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
@@ -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"])
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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))
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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(
|
||||
{
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user