tree: make get_hash return type and hash pair (#4397)

Currently we kinda assume that whatever is returned by `get_file_hash`
is of type self.PARAM_CHECKSUM, which is not actually true. E.g. for
http it might return `etag` or `md5`, but we don't distinguish between
those and call both `etag`. This is becoming more relevant for dir
hashes that are computed a few different ways (e.g. in-memory md5 or
upload to remote and get etag for the dir file).

Prerequisite for #4144 and #3069
This commit is contained in:
Ruslan Kuprieiev
2020-08-15 20:35:39 +03:00
committed by GitHub
parent b773ba73cf
commit 023dec4aa6
17 changed files with 51 additions and 34 deletions
+5 -3
View File
@@ -127,7 +127,9 @@ class CloudCache:
logger.debug("cache for '%s'('%s') has changed.", path_info, hash_)
return True
actual = self.tree.get_hash(path_info)
typ, actual = self.tree.get_hash(path_info)
assert typ == self.tree.PARAM_CHECKSUM
if hash_ != actual:
logger.debug(
"hash value '%s' for '%s' has changed (actual '%s').",
@@ -312,7 +314,7 @@ class CloudCache:
)
return False
actual = self.tree.get_hash(cache_info)
_, actual = self.tree.get_hash(cache_info)
logger.debug(
"cache '%s' expected '%s' actual '%s'", cache_info, hash_, actual,
@@ -358,7 +360,7 @@ class CloudCache:
return self.changed_cache_file(hash_)
def already_cached(self, path_info):
current = self.tree.get_hash(path_info)
_, current = self.tree.get_hash(path_info)
if not current:
return False
+3 -1
View File
@@ -91,7 +91,9 @@ class LocalCache(CloudCache):
def already_cached(self, path_info):
assert path_info.scheme in ["", "local"]
current_md5 = self.tree.get_hash(path_info)
typ, current_md5 = self.tree.get_hash(path_info)
assert typ == "md5"
if not current_md5:
return False
+3 -1
View File
@@ -64,7 +64,9 @@ class RepoDependency(LocalDependency):
# We are polluting our repo cache with some dir listing here
if tree.isdir(path):
return self.repo.cache.local.tree.get_hash(path, tree=tree)
return self.repo.cache.local.tree.get_hash(
path, tree=tree
)[1]
return tree.get_file_hash(path)
def workspace_status(self):
+1 -1
View File
@@ -184,7 +184,7 @@ class BaseOutput:
self.info[self.tree.PARAM_CHECKSUM] = checksum
def get_checksum(self):
return self.tree.get_hash(self.path_info)
return self.tree.get_hash(self.path_info)[1]
@property
def is_dir_checksum(self):
+1 -1
View File
@@ -37,7 +37,7 @@ def diff(self, a_rev="HEAD", b_rev=None):
def _to_checksum(output):
if on_working_tree:
return self.cache.local.tree.get_hash(output.path_info)
return self.cache.local.tree.get_hash(output.path_info)[1]
return output.checksum
def _exists(output):
+6 -3
View File
@@ -238,8 +238,11 @@ class DvcTree(BaseTree): # pylint:disable=abstract-method
raise OutputNotFoundError
out = outs[0]
if out.is_dir_checksum:
return self._get_granular_checksum(path_info, out)
return out.checksum
return (
out.tree.PARAM_CHECKSUM,
self._get_granular_checksum(path_info, out),
)
return out.tree.PARAM_CHECKSUM, out.checksum
class RepoTree(BaseTree): # pylint:disable=abstract-method
@@ -504,7 +507,7 @@ class RepoTree(BaseTree): # pylint:disable=abstract-method
return dvc_tree.get_file_hash(path_info)
except OutputNotFoundError:
pass
return file_md5(path_info, self)[0]
return self.PARAM_CHECKSUM, file_md5(path_info, self)[0]
def copytree(self, top, dest):
top = PathInfo(top)
+1 -1
View File
@@ -153,7 +153,7 @@ class AzureTree(BaseTree):
).delete_blob()
def get_file_hash(self, path_info):
return self.get_etag(path_info)
return self.PARAM_CHECKSUM, self.get_etag(path_info)
def _upload(
self, from_file, to_info, name=None, no_progress_bar=False, **_kwargs
+15 -13
View File
@@ -242,7 +242,7 @@ class BaseTree:
)
if not self.exists(path_info):
return None
return self.PARAM_CHECKSUM, None
# pylint: disable=assignment-from-none
hash_ = self.state.get(path_info)
@@ -260,17 +260,17 @@ class BaseTree:
hash_ = None
if hash_:
return hash_
return self.PARAM_CHECKSUM, hash_
if self.isdir(path_info):
hash_ = self.get_dir_hash(path_info, **kwargs)
typ, hash_ = self.get_dir_hash(path_info, **kwargs)
else:
hash_ = self.get_file_hash(path_info)
typ, hash_ = self.get_file_hash(path_info)
if hash_ and self.exists(path_info):
self.state.save(path_info, hash_)
return hash_
return typ, hash_
def get_file_hash(self, path_info):
raise NotImplementedError
@@ -294,7 +294,8 @@ class BaseTree:
return "".join(parts)
def save_info(self, path_info, **kwargs):
return {self.PARAM_CHECKSUM: self.get_hash(path_info, **kwargs)}
typ, hash_ = self.get_hash(path_info, **kwargs)
return {typ: hash_}
def _calculate_hashes(self, file_infos):
file_infos = list(file_infos)
@@ -305,9 +306,10 @@ class BaseTree:
) as pbar:
worker = pbar.wrap_fn(self.get_file_hash)
with ThreadPoolExecutor(max_workers=self.hash_jobs) as executor:
tasks = executor.map(worker, file_infos)
hashes = dict(zip(file_infos, tasks))
return hashes
hashes = (
value for typ, value in executor.map(worker, file_infos)
)
return dict(zip(file_infos, hashes))
def _collect_dir(self, path_info, **kwargs):
file_infos = set()
@@ -344,7 +346,7 @@ class BaseTree:
return sorted(result, key=itemgetter(self.PARAM_RELPATH))
def _save_dir_info(self, dir_info):
hash_, tmp_info = self._get_dir_info_hash(dir_info)
typ, hash_, tmp_info = self._get_dir_info_hash(dir_info)
new_info = self.cache.tree.hash_to_path_info(hash_)
if self.cache.changed_cache_file(hash_):
self.cache.tree.makedirs(new_info.parent)
@@ -354,7 +356,7 @@ class BaseTree:
self.state.save(new_info, hash_)
return hash_
return typ, hash_
def _get_dir_info_hash(self, dir_info):
tmp = tempfile.NamedTemporaryFile(delete=False).name
@@ -366,8 +368,8 @@ class BaseTree:
to_info = tree.path_info / tmp_fname("")
tree.upload(from_info, to_info, no_progress_bar=True)
hash_ = tree.get_file_hash(to_info) + self.CHECKSUM_DIR_SUFFIX
return hash_, to_info
typ, hash_ = tree.get_file_hash(to_info)
return typ, hash_ + self.CHECKSUM_DIR_SUFFIX, to_info
def upload(self, from_info, to_info, name=None, no_progress_bar=False):
if not hasattr(self, "_upload"):
+5 -2
View File
@@ -190,11 +190,14 @@ class GSTree(BaseTree):
path = path_info.path
blob = self.gs.bucket(bucket).get_blob(path)
if not blob:
return None
return self.PARAM_CHECKSUM, None
b64_md5 = blob.md5_hash
md5 = base64.b64decode(b64_md5)
return codecs.getencoder("hex")(md5)[0].decode("utf-8")
return (
self.PARAM_CHECKSUM,
codecs.getencoder("hex")(md5)[0].decode("utf-8"),
)
def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
bucket = self.gs.bucket(to_info.bucket)
+1 -1
View File
@@ -167,7 +167,7 @@ class HDFSTree(BaseTree):
stdout = self.hadoop_fs(
f"checksum {path_info.url}", user=path_info.user
)
return self._group(regex, stdout, "checksum")
return self.PARAM_CHECKSUM, self._group(regex, stdout, "checksum")
def _upload(self, from_file, to_info, **_kwargs):
with self.hdfs(to_info) as hdfs:
+1 -1
View File
@@ -136,7 +136,7 @@ class HTTPTree(BaseTree): # pylint:disable=abstract-method
"Content-MD5 header for '{url}'".format(url=url)
)
return etag
return self.PARAM_CHECKSUM, etag
def _download(self, from_info, to_file, name=None, no_progress_bar=False):
response = self.request("GET", from_info.url, stream=True)
+1 -1
View File
@@ -309,7 +309,7 @@ class LocalTree(BaseTree):
return stat.S_IMODE(mode) == self.CACHE_MODE
def get_file_hash(self, path_info):
return file_md5(path_info)[0]
return self.PARAM_CHECKSUM, file_md5(path_info)[0]
@staticmethod
def getsize(path_info):
+4 -1
View File
@@ -318,7 +318,10 @@ class S3Tree(BaseTree):
raise ETagMismatchError(etag, cached_etag)
def get_file_hash(self, path_info):
return self.get_etag(self.s3, path_info.bucket, path_info.path)
return (
self.PARAM_CHECKSUM,
self.get_etag(self.s3, path_info.bucket, path_info.path),
)
def _upload(self, from_file, to_info, name=None, no_progress_bar=False):
total = os.path.getsize(from_file)
+1 -1
View File
@@ -238,7 +238,7 @@ class SSHTree(BaseTree):
raise NotImplementedError
with self.ssh(path_info) as ssh:
return ssh.md5(path_info.path)
return self.PARAM_CHECKSUM, ssh.md5(path_info.path)
def getsize(self, path_info):
with self.ssh(path_info) as ssh:
+1 -1
View File
@@ -142,7 +142,7 @@ class WebDAVTree(BaseTree): # pylint:disable=abstract-method
"Content-MD5 header for '{url}'".format(url=path_info.url)
)
return etag
return self.PARAM_CHECKSUM, etag
# Checks whether path points to directory
def isdir(self, path_info):
+1 -1
View File
@@ -211,7 +211,7 @@ def test_repotree_cache_save(tmp_dir, dvc, scm, erepo_dir, local_cloud):
# into dvc.cache, not fetched or streamed from a remote
tree = RepoTree(erepo_dir.dvc, stream=True)
expected = [
tree.get_file_hash(PathInfo(erepo_dir / path))
tree.get_file_hash(PathInfo(erepo_dir / path))[1]
for path in ("dir/bar", "dir/subdir/foo")
]
+1 -1
View File
@@ -36,7 +36,7 @@ def test_get_file_hash(tmp_dir, azure):
to_info = azure
tree.upload(PathInfo("foo"), to_info)
assert tree.exists(to_info)
hash_ = tree.get_file_hash(to_info)
_, hash_ = tree.get_file_hash(to_info)
assert hash_
assert isinstance(hash_, str)
assert hash_.strip("'").strip('"') == hash_