Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-13 15:28:27 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'master' into analytics-transparency-improvements

Browse Source
This commit is contained in:
Ivan Shcheklein
2018-12-21 18:14:12 -08:00
parent 8e73b97374 716f63cba8
commit 16ec387a7e
24 changed files with 552 additions and 955 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.dvc/.gitignore
+8 -10
View File
@@ -1,10 +1,8 @@
link.state
link.state.lock
lock
config.local
updater
cache
state
state.lock
state-journal
state-wal
/state
/lock
/config.local
/updater
/updater.lock
/state-journal
/state-wal
/cache
dvc/dependency/__init__.py
+3 -3
View File
@@ -39,11 +39,11 @@ DEP_MAP = {
'https': DependencyHTTP,
}
# We are skipping RemoteHTTP.PARAM_ETAG because is the same as RemoteS3
# We are skipping RemoteHTTP.PARAM_CHECKSUM because is the same as RemoteS3
SCHEMA = {
DependencyBase.PARAM_PATH: str,
schema.Optional(RemoteLOCAL.PARAM_MD5): schema.Or(str, None),
schema.Optional(RemoteS3.PARAM_ETAG): schema.Or(str, None),
schema.Optional(RemoteLOCAL.PARAM_CHECKSUM): schema.Or(str, None),
schema.Optional(RemoteS3.PARAM_CHECKSUM): schema.Or(str, None),
schema.Optional(RemoteHDFS.PARAM_CHECKSUM): schema.Or(str, None),
}
dvc/dependency/base.py
+1 -1
View File
@@ -60,7 +60,7 @@ class DependencyBase(object):
@property
def exists(self):
return self.remote.exists([self.path_info])[0]
return self.remote.exists(self.path_info)
def changed(self):
raise NotImplementedError
dvc/dependency/gs.py
+3 -3
View File
@@ -17,9 +17,9 @@ class DependencyGS(DependencyS3):
self.info = info
self.remote = remote if remote else RemoteGS(stage.project, {})
bucket = remote.bucket if remote else urlparse(path).netloc
key = urlparse(path).path.lstrip('/')
path = urlparse(path).path.lstrip('/')
if remote:
key = posixpath.join(remote.prefix, key)
path = posixpath.join(remote.prefix, path)
self.path_info = {'scheme': 'gs',
'bucket': bucket,
'key': key}
'path': path}
dvc/dependency/hdfs.py
+1 -1
View File
@@ -21,7 +21,7 @@ class DependencyHDFS(DependencyBase):
user = remote.user if remote else self.group('user')
self.path_info = {'scheme': 'hdfs',
'user': user,
'url': path}
'path': path}
def changed(self):
if not self.exists:
dvc/dependency/http.py
+1 -1
View File
@@ -21,7 +21,7 @@ class DependencyHTTP(DependencyBase):
self.path_info = {
'scheme': urlparse(path).scheme,
'url': path,
'path': path,
}
def save(self):
dvc/dependency/local.py
+1 -1
View File
@@ -30,7 +30,7 @@ class DependencyLOCAL(DependencyBase):
path = os.path.join(remote.prefix, urlparse(path).path.lstrip('/'))
if not os.path.isabs(path):
path = self.remote.ospath(path)
path = self.remote.to_ospath(path)
path = os.path.join(stage.cwd, path)
self.path = os.path.abspath(os.path.normpath(path))
dvc/dependency/s3.py
+3 -3
View File
@@ -18,12 +18,12 @@ class DependencyS3(DependencyBase):
self.remote = remote or RemoteS3(stage.project, {})
bucket = remote.bucket if remote else urlparse(path).netloc
key = urlparse(path).path.lstrip('/')
path = urlparse(path).path.lstrip('/')
if remote:
key = posixpath.join(remote.prefix, key)
path = posixpath.join(remote.prefix, path)
self.path_info = {'scheme': 's3',
'bucket': bucket,
'key': key}
'path': path}
def changed(self):
if not self.exists:
dvc/output/local.py
+1 -1
View File
@@ -34,7 +34,7 @@ class OutputLOCAL(DependencyLOCAL):
@property
def md5(self):
return self.info.get(self.project.cache.local.PARAM_MD5, None)
return self.info.get(self.project.cache.local.PARAM_CHECKSUM, None)
@property
def cache(self):
dvc/project.py
+13 -10
View File
@@ -602,7 +602,7 @@ class Project(object):
info = out.dumpd()
ret = [info]
r = out.remote
md5 = info[r.PARAM_MD5]
md5 = info[r.PARAM_CHECKSUM]
if self.cache.local.changed_cache_file(md5):
try:
@@ -649,7 +649,7 @@ class Project(object):
if out.path_info['scheme'] != 'local':
return ret
md5 = info[out.remote.PARAM_MD5]
md5 = info[out.remote.PARAM_CHECKSUM]
cache = self.cache.local.get(md5)
if not out.remote.is_dir_cache(cache):
return ret
@@ -896,12 +896,15 @@ class Project(object):
remote=remote,
jobs=jobs)['local']
status = {}
for md5, ret in self.cloud.status(used,
jobs,
remote=remote,
show_checksums=show_checksums):
if ret == cloud.STATUS_OK:
ret = {}
status_info = self.cloud.status(used,
jobs,
remote=remote,
show_checksums=show_checksums)
for md5, info in status_info.items():
name = info['name']
status = info['status']
if status == cloud.STATUS_OK:
continue
prefix_map = {
@@ -909,9 +912,9 @@ class Project(object):
cloud.STATUS_NEW: 'new',
}
status[md5] = prefix_map[ret]
ret[name] = prefix_map[status]
return status
return ret
def status(self,
target=None,
dvc/remote/azure.py
+26 -124
View File
@@ -1,7 +1,6 @@
from __future__ import absolute_import
import os
import re
import posixpath
try:
from azure.storage.blob import BlockBlobService
@@ -49,7 +48,7 @@ class RemoteAzure(RemoteBase):
or match.group('container_name') # backward compatibility
or os.getenv('AZURE_STORAGE_CONTAINER_NAME'))
self.prefix = urlparse(self.url).path.lstrip('/') or '/'
self.prefix = urlparse(self.url).path.lstrip('/')
self.connection_string = (
config.get(Config.SECTION_AZURE_CONNECTION_STRING)
@@ -64,6 +63,8 @@ class RemoteAzure(RemoteBase):
self.__blob_service = None
self.path_info = {'scheme': self.scheme, 'bucket': self.bucket}
@property
def blob_service(self):
if self.__blob_service is None:
@@ -72,104 +73,33 @@ class RemoteAzure(RemoteBase):
self.__blob_service.create_container(self.bucket)
return self.__blob_service
# FIXME: temporarily disabled because of the lack of test for external azure
# dependencies/outputs/cache.
#
# def _get_etag(self, bucket, key):
# try:
# blob = self.blob_service.get_blob_properties(bucket, key)
# return blob.properties.etag
# except Exception:
# return None
#
# def save_info(self, path_info):
# if path_info['scheme'] != self.scheme:
# raise NotImplementedError
#
# return {self.PARAM_ETAG: self._get_etag(
# path_info['bucket'], path_info['key'])}
#
# def save(self, path_info):
# if path_info['scheme'] != self.scheme:
# raise NotImplementedError
#
# etag = self._get_etag(path_info['bucket'], path_info['key'])
# dest_key = '{}/{}'.format(etag[0:2], etag[2:])
#
# self._copy_blob(to_bucket=self.bucket,
# to_key=dest_key,
# from_bucket=path_info['bucket'],
# from_key=path_info['key'])
#
# return {self.PARAM_ETAG: etag}
#
# def changed(self, path_info, checksum_info):
# return checksum_info != self.save_info(path_info)
#
# def _copy_blob(self, to_bucket, to_key, from_bucket, from_key):
# source = self.blob_service.make_blob_url(from_bucket, from_key)
#
# copy = self.blob_service.copy_blob(to_bucket, to_key, source)
#
# if self.COPY_POLL_SECONDS <= 0:
# return
#
# while copy.status != 'success':
# time.sleep(self.COPY_POLL_SECONDS)
# copy = self.blob_service.get_blob_properties(
# to_bucket, to_key).properties.copy
#
# def checkout(self, path_info, checksum_info):
# if path_info['scheme'] != self.scheme:
# raise NotImplementedError
#
# etag = checksum_info.get(self.PARAM_ETAG, None)
# if not etag:
# return
#
# self._copy_blob(to_bucket=path_info['bucket'],
# to_key=path_info['key'],
# from_bucket=self.bucket,
# from_key='{}/{}'.format(etag[0:2], etag[2:]))
#
def remove(self, path_info):
if path_info['scheme'] != self.scheme:
raise NotImplementedError
logger.debug('Removing azure://{}/{}'.format(path_info['bucket'],
path_info['key']))
path_info['path']))
self.blob_service.delete_blob(path_info['bucket'], path_info['key'])
self.blob_service.delete_blob(path_info['bucket'], path_info['path'])
def md5s_to_path_infos(self, md5s):
return [{
'scheme': self.scheme,
'bucket': self.bucket,
'key': '{}/{}/{}'.format(self.prefix, md5[0:2], md5[2:])
} for md5 in md5s]
def _list_paths(self, bucket, prefix):
blob_service = self.blob_service
next_marker = None
while True:
blobs = blob_service.list_blobs(bucket,
prefix=prefix,
marker=next_marker)
def _all_keys(self, bucket, prefix):
return {blob.name
for blob in self.blob_service.list_blobs(bucket,
prefix=prefix)}
for blob in blobs:
yield blob.name
def exists(self, path_infos):
ret = []
if not blobs.next_marker:
break
if len(path_infos) == 0:
return ret
next_marker = blobs.next_marker
bucket = path_infos[0]['bucket']
key = path_infos[0]['key']
for path_info in path_infos:
assert path_info['scheme'] == self.scheme
assert path_info['bucket'] == bucket
key = path_info['key']
ret.append(key in self._all_keys(bucket, key))
return ret
def list_cache_paths(self):
return self._list_paths(self.bucket, self.prefix)
def upload(self, from_infos, to_infos, names=None):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -182,10 +112,10 @@ class RemoteAzure(RemoteBase):
raise NotImplementedError
bucket = to_info['bucket']
key = to_info['key']
path = to_info['path']
logger.debug("Uploading '{}' to '{}/{}'".format(
from_info['path'], bucket, key))
from_info['path'], bucket, path))
if not name:
name = os.path.basename(from_info['path'])
@@ -194,7 +124,7 @@ class RemoteAzure(RemoteBase):
try:
self.blob_service.create_blob_from_path(
bucket, key, from_info['path'], progress_callback=cb)
bucket, path, from_info['path'], progress_callback=cb)
except Exception as ex:
msg = "Failed to upload '{}'".format(from_info['path'])
logger.warn(msg, ex)
@@ -216,10 +146,10 @@ class RemoteAzure(RemoteBase):
raise NotImplementedError
bucket = from_info['bucket']
key = from_info['key']
path = from_info['path']
logger.debug("Downloading '{}/{}' to '{}'".format(
bucket, key, to_info['path']))
bucket, path, to_info['path']))
tmp_file = self.tmp_file(to_info['path'])
if not name:
@@ -231,40 +161,12 @@ class RemoteAzure(RemoteBase):
try:
self.blob_service.get_blob_to_path(
bucket, key, tmp_file, progress_callback=cb)
bucket, path, tmp_file, progress_callback=cb)
except Exception as exc:
msg = "Failed to download '{}/{}'".format(bucket, key)
msg = "Failed to download '{}/{}'".format(bucket, path)
logger.warn(msg, exc)
else:
os.rename(tmp_file, to_info['path'])
if not no_progress_bar:
progress.finish_target(name)
def _path_to_etag(self, path):
relpath = posixpath.relpath(path, self.prefix)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all(self):
keys = self._all_keys(self.bucket, self.prefix)
return [self._path_to_etag(key) for key in keys]
def gc(self, cinfos):
from dvc.remote.local import RemoteLOCAL
used = [info[self.PARAM_ETAG] for info in cinfos['azure']]
used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']]
removed = False
for etag in self._all():
if etag in used:
continue
path_info = {'scheme': self.scheme,
'key': posixpath.join(self.prefix,
etag[0:2],
etag[2:]),
'bucket': self.bucket}
self.remove(path_info)
removed = True
return removed
dvc/remote/base.py
+147 -6
View File
@@ -1,8 +1,10 @@
import os
import re
import errno
import posixpath
from multiprocessing import cpu_count
import dvc.prompt as prompt
from dvc.config import Config
from dvc.logger import logger
from dvc.exceptions import DvcException
@@ -104,9 +106,6 @@ class RemoteBase(object):
def save(self, path_info):
raise NotImplementedError
def checkout(self, path_info, checksum_info, force=False):
raise NotImplementedError
def download(self, from_infos, to_infos, no_progress_bar=False, name=None):
raise NotImplementedError
@@ -119,6 +118,9 @@ class RemoteBase(object):
def move(self, path_info):
raise NotImplementedError
def copy(self, from_info, to_info):
raise NotImplementedError
def _makedirs(self, fname):
dname = os.path.dirname(fname)
@@ -128,9 +130,6 @@ class RemoteBase(object):
if e.errno != errno.EEXIST:
raise
def md5s_to_path_infos(self, md5s):
raise NotImplementedError
def exists(self, path_infos):
raise NotImplementedError
@@ -147,3 +146,145 @@ class RemoteBase(object):
assert len(names) == len(to_infos)
return names
@property
def ospath(self):
return posixpath
def checksum_to_path(self, checksum):
return self.ospath.join(self.prefix, checksum[0:2], checksum[2:])
def path_to_checksum(self, path):
relpath = self.ospath.relpath(path, self.prefix)
return self.ospath.dirname(relpath) + self.ospath.basename(relpath)
def checksum_to_path_info(self, checksum):
path_info = self.path_info.copy()
path_info['path'] = self.checksum_to_path(checksum)
return path_info
def md5s_to_path_infos(self, md5s):
return [
self.checksum_to_path_info(md5)
for md5 in md5s
]
def list_cache_paths(self):
raise NotImplementedError
def all(self):
# NOTE: The list might be way too big(e.g. 100M entries, md5 for each
# is 32 bytes, so ~3200Mb list) and we don't really need all of it at
# the same time, so it makes sense to use a generator to gradually
# iterate over it, without keeping all of it in memory.
return (
self.path_to_checksum(path)
for path in self.list_cache_paths()
)
def gc(self, cinfos):
from dvc.remote.local import RemoteLOCAL
used = [info[RemoteLOCAL.PARAM_CHECKSUM] for info in cinfos['local']]
if self.scheme != '':
used += [info[self.PARAM_CHECKSUM] for info in cinfos[self.scheme]]
removed = False
for checksum in self.all():
if checksum in used:
continue
path_info = self.checksum_to_path_info(checksum)
self.remove(path_info)
removed = True
return removed
def changed_cache(self, checksum):
cache = self.checksum_to_path_info(checksum)
expected = {self.PARAM_CHECKSUM: checksum}
actual = self.save_info(cache)
logger.debug("Cache '{}' actual '{}'.".format(str(expected),
str(actual)))
if expected != actual:
if self.exists(cache):
msg = 'Corrupted cache file {}'
logger.warn(msg.format(str(cache)))
self.remove(cache)
return True
return False
def cache_exists(self, checksums):
# NOTE: The reason for such an odd logic is that most of the remotes
# take much shorter time to just retrieve everything they have under
# a certain prefix(e.g. s3, gs, ssh, hdfs). Other remotes that can
# check if particular file exists much quicker, use their own
# implementation of cache_exists(see http, local).
#
# Result of all() might be way too big, so we should walk through
# it in one pass.
return list(filter(lambda checksum: checksum in checksums, self.all()))
def already_cached(self, path_info):
current = self.save_info(path_info)[self.PARAM_CHECKSUM]
if not current:
return False
return not self.changed_cache(current)
def safe_remove(self, path_info, force=False):
if not self.exists(path_info):
return
if not force and not self.already_cached(path_info):
msg = (
'File "{}" is going to be removed. '
'Are you sure you want to proceed?'
.format(str(path_info))
)
if not prompt.confirm(msg):
raise DvcException("Unable to remove {} without a confirmation"
" from the user. Use '-f' to force."
.format(str(path_info)))
self.remove(path_info)
def do_checkout(self, path_info, checksum, force=False):
if self.exists(path_info):
msg = "Data '{}' exists. Removing before checkout."
logger.warn(msg.format(str(path_info)))
self.safe_remove(path_info, force=force)
from_info = self.checksum_to_path_info(checksum)
self.copy(from_info, path_info)
def checkout(self, path_info, checksum_info, force=False):
scheme = path_info['scheme']
if scheme not in ['', 'local'] and scheme != self.scheme:
raise NotImplementedError
checksum = checksum_info.get(self.PARAM_CHECKSUM)
if not checksum:
msg = "No checksum info for '{}'."
logger.info(msg.format(str(path_info)))
return
if not self.changed(path_info, checksum_info):
msg = "Data '{}' didn't change."
logger.info(msg.format(str(path_info)))
return
if self.changed_cache(checksum):
msg = "Cache '{}' not found. File '{}' won't be created."
logger.warn(msg.format(checksum, str(path_info)))
self.safe_remove(path_info, force=force)
return
msg = "Checking out '{}' with cache '{}'."
logger.info(msg.format(str(path_info), checksum))
self.do_checkout(path_info, checksum, force=force)
dvc/remote/gs.py
+42 -137
View File
@@ -1,5 +1,4 @@
import os
import posixpath
try:
from google.cloud import storage
@@ -13,7 +12,6 @@ except ImportError:
from dvc.logger import logger
from dvc.remote.base import RemoteBase
from dvc.remote.local import RemoteLOCAL
from dvc.config import Config
from dvc.progress import progress
from dvc.exceptions import DvcException
@@ -23,7 +21,7 @@ class RemoteGS(RemoteBase):
scheme = 'gs'
REGEX = r'^gs://(?P<path>.*)$'
REQUIRES = {'google.cloud.storage': storage}
PARAM_MD5 = 'md5'
PARAM_CHECKSUM = 'md5'
def __init__(self, project, config):
self.project = project
@@ -33,6 +31,12 @@ class RemoteGS(RemoteBase):
self.url = config.get(Config.SECTION_REMOTE_URL, storagepath)
self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None)
parsed = urlparse(self.url)
self.bucket = parsed.netloc
self.prefix = parsed.path.lstrip('/')
self.path_info = {'scheme': 'gs', 'bucket': self.bucket}
@staticmethod
def compat_config(config):
ret = config.copy()
@@ -40,23 +44,15 @@ class RemoteGS(RemoteBase):
ret[Config.SECTION_REMOTE_URL] = url
return ret
@property
def bucket(self):
return urlparse(self.url).netloc
@property
def prefix(self):
return urlparse(self.url).path.lstrip('/')
@property
def gs(self):
return storage.Client()
def get_md5(self, bucket, key):
def get_md5(self, bucket, path):
import base64
import codecs
blob = self.gs.bucket(bucket).get_blob(key)
blob = self.gs.bucket(bucket).get_blob(path)
if not blob:
return None
@@ -68,33 +64,14 @@ class RemoteGS(RemoteBase):
if path_info['scheme'] != 'gs':
raise NotImplementedError
return {self.PARAM_MD5: self.get_md5(path_info['bucket'],
path_info['key'])}
@staticmethod
def to_string(path_info):
return "{}://{}/{}".format(path_info['scheme'],
path_info['bucket'],
path_info['key'])
def changed_cache(self, md5):
key = posixpath.join(self.prefix, md5[0:2], md5[2:])
cache = {'scheme': 'gs', 'bucket': self.bucket, 'key': key}
if {self.PARAM_MD5: md5} != self.save_info(cache):
if self.exists([cache])[0]:
msg = 'Corrupted cache file {}'
logger.warn(msg.format(self.to_string(cache)))
self.remove(cache)
return True
return False
return {self.PARAM_CHECKSUM: self.get_md5(path_info['bucket'],
path_info['path'])}
def changed(self, path_info, checksum_info):
if not self.exists([path_info])[0]:
if not self.exists(path_info):
return True
md5 = checksum_info.get(self.PARAM_MD5, None)
md5 = checksum_info.get(self.PARAM_CHECKSUM, None)
if md5 is None:
return True
@@ -103,103 +80,57 @@ class RemoteGS(RemoteBase):
return checksum_info != self.save_info(path_info)
def _copy(self, from_info, to_info, gs=None):
def copy(self, from_info, to_info, gs=None):
gs = gs if gs else self.gs
blob = gs.bucket(from_info['bucket']).get_blob(from_info['key'])
blob = gs.bucket(from_info['bucket']).get_blob(from_info['path'])
if not blob:
msg = '{} doesn\'t exist in the cloud'
raise DvcException(msg.format(from_info['key']))
raise DvcException(msg.format(from_info['path']))
bucket = self.gs.bucket(to_info['bucket'])
bucket.copy_blob(blob,
self.gs.bucket(to_info['bucket']),
new_name=to_info['key'])
new_name=to_info['path'])
def save(self, path_info):
if path_info['scheme'] != 'gs':
raise NotImplementedError
md5 = self.get_md5(path_info['bucket'], path_info['key'])
key = posixpath.join(self.prefix, md5[0:2], md5[2:])
to_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key}
md5 = self.get_md5(path_info['bucket'], path_info['path'])
path = self.checksum_to_path(md5)
to_info = {'scheme': 'gs', 'bucket': self.bucket, 'path': path}
self._copy(path_info, to_info)
self.copy(path_info, to_info)
return {self.PARAM_MD5: md5}
def checkout(self, path_info, checksum_info):
if path_info['scheme'] != 'gs':
raise NotImplementedError
md5 = checksum_info.get(self.PARAM_MD5, None)
if not md5:
return
if not self.changed(path_info, checksum_info):
msg = "Data '{}' didn't change."
logger.info(msg.format(self.to_string(path_info)))
return
if self.changed_cache(md5):
msg = "Cache '{}' not found. File '{}' won't be created."
logger.warn(msg.format(md5, self.to_string(path_info)))
return
if self.exists([path_info])[0]:
msg = "Data '{}' exists. Removing before checkout."
logger.warn(msg.format(self.to_string(path_info)))
self.remove(path_info)
return
msg = "Checking out '{}' with cache '{}'."
logger.info(msg.format(self.to_string(path_info), md5))
key = posixpath.join(self.prefix, md5[0:2], md5[2:])
from_info = {'scheme': 'gs', 'bucket': self.bucket, 'key': key}
self._copy(from_info, path_info)
return {self.PARAM_CHECKSUM: md5}
def remove(self, path_info):
if path_info['scheme'] != 'gs':
raise NotImplementedError
logger.debug("Removing gs://{}/{}".format(path_info['bucket'],
path_info['key']))
path_info['path']))
blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['key'])
blob = self.gs.bucket(path_info['bucket']).get_blob(path_info['path'])
if not blob:
return
blob.delete()
def md5s_to_path_infos(self, md5s):
return [{'scheme': 'gs',
'bucket': self.bucket,
'key': posixpath.join(self.prefix,
md5[0:2], md5[2:])} for md5 in md5s]
def _list_paths(self, bucket, prefix):
for blob in self.gs.bucket(bucket).list_blobs(prefix=prefix):
yield blob.name
def exists(self, path_infos):
ret = []
def list_cache_paths(self):
return self._list_paths(self.bucket, self.prefix)
if len(path_infos) == 0:
return ret
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info['scheme'] == 'gs'
gs = self.gs
bucket_name = path_infos[0]['bucket']
bucket = gs.bucket(bucket_name)
for path_info in path_infos:
assert self.scheme == path_info['scheme']
assert bucket_name == path_info['bucket']
exists = False
key = path_info['key']
keys = [blob.name for blob in bucket.list_blobs(prefix=key)]
if key in keys:
exists = True
ret.append(exists)
return ret
paths = self._list_paths(path_info['bucket'], path_info['path'])
return any(path_info['path'] == path for path in paths)
def upload(self, from_infos, to_infos, names=None):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -215,7 +146,7 @@ class RemoteGS(RemoteBase):
logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'],
to_info['bucket'],
to_info['key']))
to_info['path']))
if not name:
name = os.path.basename(from_info['path'])
@@ -224,13 +155,13 @@ class RemoteGS(RemoteBase):
try:
bucket = gs.bucket(to_info['bucket'])
blob = bucket.blob(to_info['key'])
blob = bucket.blob(to_info['path'])
blob.upload_from_filename(from_info['path'])
except Exception as exc:
msg = "Failed to upload '{}' to '{}/{}'"
logger.warn(msg.format(from_info['path'],
to_info['bucket'],
to_info['key']), exc)
to_info['path']), exc)
continue
progress.finish_target(name)
@@ -249,14 +180,14 @@ class RemoteGS(RemoteBase):
raise NotImplementedError
if to_info['scheme'] == 'gs':
self._copy(from_info, to_info, gs=gs)
self.copy(from_info, to_info, gs=gs)
continue
if to_info['scheme'] != 'local':
raise NotImplementedError
msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'],
from_info['key'],
from_info['path'],
to_info['path'])
logger.debug(msg)
@@ -266,19 +197,19 @@ class RemoteGS(RemoteBase):
if not no_progress_bar:
# percent_cb is not available for download_to_filename, so
# lets at least update progress at keypoints(start, finish)
# lets at least update progress at pathpoints(start, finish)
progress.update_target(name, 0, None)
self._makedirs(to_info['path'])
try:
bucket = gs.bucket(from_info['bucket'])
blob = bucket.get_blob(from_info['key'])
blob = bucket.get_blob(from_info['path'])
blob.download_to_filename(tmp_file)
except Exception as exc:
msg = "Failed to download '{}/{}' to '{}'"
logger.warn(msg.format(from_info['bucket'],
from_info['key'],
from_info['path'],
to_info['path']), exc)
continue
@@ -286,29 +217,3 @@ class RemoteGS(RemoteBase):
if not no_progress_bar:
progress.finish_target(name)
def _path_to_md5(self, path):
relpath = posixpath.relpath(path, self.prefix)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all_md5s(self):
blobs = self.gs.bucket(self.bucket).list_blobs(prefix=self.prefix)
blobs = list(blobs)
return [self._path_to_md5(blob.name) for blob in blobs]
def gc(self, cinfos):
used = [info[self.PARAM_MD5] for info in cinfos['gs']]
used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']]
removed = False
for md5 in self._all_md5s():
if md5 in used:
continue
path_info = {'scheme': 'gs',
'bucket': self.bucket,
'key': posixpath.join(self.prefix,
md5[0:2], md5[2:])}
self.remove(path_info)
removed = True
return removed
dvc/remote/hdfs.py
+37 -116
View File
@@ -6,7 +6,6 @@ from subprocess import Popen, PIPE
from dvc.config import Config
from dvc.remote.base import RemoteBase, RemoteBaseCmdError
from dvc.remote.local import RemoteLOCAL
from dvc.logger import logger
from dvc.utils import fix_env
@@ -23,11 +22,14 @@ class RemoteHDFS(RemoteBase):
def __init__(self, project, config):
self.project = project
self.url = config.get(Config.SECTION_REMOTE_URL, '/')
self.prefix = self.url
self.user = self.group('user')
if not self.user:
self.user = config.get(Config.SECTION_REMOTE_USER,
getpass.getuser())
self.path_info = {'scheme': 'hdfs', 'user': self.user}
def hadoop_fs(self, cmd, user=None):
cmd = 'hadoop fs -' + cmd
if user:
@@ -59,50 +61,31 @@ class RemoteHDFS(RemoteBase):
def checksum(self, path_info):
regex = r'.*\t.*\t(?P<checksum>.*)'
stdout = self.hadoop_fs('checksum {}'.format(path_info['url']),
stdout = self.hadoop_fs('checksum {}'.format(path_info['path']),
user=path_info['user'])
return self._group(regex, stdout, 'checksum')
def cp(self, from_info, to_info):
self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])),
user=to_info['user'])
self.hadoop_fs('cp -f {} {}'.format(from_info['url'], to_info['url']),
def copy(self, from_info, to_info):
dname = posixpath.dirname(to_info['path'])
self.hadoop_fs('mkdir -p {}'.format(dname), user=to_info['user'])
self.hadoop_fs('cp -f {} {}'.format(from_info['path'],
to_info['path']),
user=to_info['user'])
def rm(self, path_info):
self.hadoop_fs('rm {}'.format(path_info['url']),
self.hadoop_fs('rm {}'.format(path_info['path']),
user=path_info['user'])
def save_info(self, path_info):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
assert path_info.get('url')
assert path_info.get('path')
return {self.PARAM_CHECKSUM: self.checksum(path_info)}
@staticmethod
def to_string(path_info):
return "{}://{}".format(path_info['scheme'],
path_info['url'])
def changed_cache(self, checksum):
cache = {}
cache['scheme'] = 'hdfs'
cache['user'] = self.user
cache['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:])
if {self.PARAM_CHECKSUM: checksum} != self.save_info(cache):
if self.exists([cache])[0]:
msg = 'Corrupted cache file {}'
logger.warn(msg.format(self.to_string(cache)))
self.remove(cache)
return True
return False
def changed(self, path_info, checksum_info):
if not self.exists([path_info])[0]:
if not self.exists(path_info):
return True
checksum = checksum_info.get(self.PARAM_CHECKSUM, None)
@@ -118,82 +101,35 @@ class RemoteHDFS(RemoteBase):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
assert path_info.get('url')
assert path_info.get('path')
checksum = self.checksum(path_info)
dest = path_info.copy()
dest['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:])
dest['path'] = self.checksum_to_path(checksum)
self.cp(path_info, dest)
self.copy(path_info, dest)
return {self.PARAM_CHECKSUM: checksum}
def checkout(self, path_info, checksum_info):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
assert path_info.get('url')
checksum = checksum_info.get(self.PARAM_CHECKSUM, None)
if not checksum:
return
if not self.changed(path_info, checksum_info):
msg = "Data '{}' didn't change."
logger.info(msg.format(self.to_string(path_info)))
return
if self.changed_cache(checksum):
msg = "Cache '{}' not found. File '{}' won't be created."
logger.warn(msg.format(checksum, self.to_string(path_info)))
return
if self.exists([path_info])[0]:
msg = "Data '{}' exists. Removing before checkout."
logger.warn(msg.format(self.to_string(path_info)))
self.remove(path_info)
return
msg = "Checking out '{}' with cache '{}'."
logger.info(msg.format(self.to_string(path_info), checksum))
src = path_info.copy()
src['url'] = posixpath.join(self.url, checksum[0:2], checksum[2:])
self.cp(src, path_info)
def remove(self, path_info):
if path_info['scheme'] != 'hdfs':
raise NotImplementedError
assert path_info.get('url')
assert path_info.get('path')
logger.debug('Removing {}'.format(path_info['url']))
logger.debug('Removing {}'.format(path_info['path']))
self.rm(path_info)
def md5s_to_path_infos(self, md5s):
return [{'scheme': 'hdfs',
'user': self.user,
'url': posixpath.join(self.url,
md5[0:2], md5[2:])} for md5 in md5s]
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info['scheme'] == 'hdfs'
def exists(self, path_infos):
ret = []
if len(path_infos) == 0:
return ret
for path_info in path_infos:
assert path_info['scheme'] == self.scheme
try:
self.hadoop_fs('test -e {}'.format(path_info['url']))
exists = True
except RemoteHDFSCmdError:
exists = False
ret.append(exists)
return ret
try:
self.hadoop_fs('test -e {}'.format(path_info['path']))
return True
except RemoteHDFSCmdError:
return False
def upload(self, from_infos, to_infos, names=None):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -205,11 +141,11 @@ class RemoteHDFS(RemoteBase):
if from_info['scheme'] != 'local':
raise NotImplementedError
cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['url']))
cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['path']))
self.hadoop_fs(cmd, user=to_info['user'])
cmd = 'copyFromLocal {} {}'.format(from_info['path'],
to_info['url'])
to_info['path'])
self.hadoop_fs(cmd, user=to_info['user'])
def download(self,
@@ -224,7 +160,7 @@ class RemoteHDFS(RemoteBase):
raise NotImplementedError
if to_info['scheme'] == 'hdfs':
self.cp(from_info, to_info)
self.copy(from_info, to_info)
continue
if to_info['scheme'] != 'local':
@@ -234,36 +170,21 @@ class RemoteHDFS(RemoteBase):
if not os.path.exists(dname):
os.makedirs(dname)
cmd = 'copyToLocal {} {}'.format(from_info['url'], to_info['path'])
cmd = 'copyToLocal {} {}'.format(from_info['path'],
to_info['path'])
self.hadoop_fs(cmd, user=from_info['user'])
def _path_to_checksum(self, path):
relpath = posixpath.relpath(path, self.url)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def list_cache_paths(self):
try:
self.hadoop_fs('test -e {}'.format(self.prefix))
except RemoteHDFSCmdError:
return []
def _all_checksums(self):
stdout = self.hadoop_fs('ls -R {}'.format(self.url))
stdout = self.hadoop_fs('ls -R {}'.format(self.prefix))
lines = stdout.split('\n')
flist = []
for line in lines:
if not line.startswith('-'):
continue
flist.append(line.split()[-1])
return [self._path_to_checksum(path) for path in flist]
def gc(self, cinfos):
used = [info[self.PARAM_CHECKSUM] for info in cinfos['hdfs']]
used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']]
removed = False
for checksum in self._all_checksums():
if checksum in used:
continue
path_info = {'scheme': 'hdfs',
'user': self.user,
'url': posixpath.join(self.url,
checksum[0:2], checksum[2:])}
self.remove(path_info)
removed = True
return removed
return flist
dvc/remote/http.py
+21 -20
View File
@@ -1,7 +1,6 @@
import os
import threading
import requests
import posixpath
from dvc.logger import logger
from dvc.progress import progress
@@ -34,6 +33,7 @@ class RemoteHTTP(RemoteBase):
self.project = project
self.cache_dir = config.get(Config.SECTION_REMOTE_URL)
self.url = self.cache_dir
self.path_info = {'scheme': 'http'}
@property
def prefix(self):
@@ -53,7 +53,7 @@ class RemoteHTTP(RemoteBase):
if to_info['scheme'] != 'local':
raise NotImplementedError
msg = "Downloading '{}' to '{}'".format(from_info['url'],
msg = "Downloading '{}' to '{}'".format(from_info['path'],
to_info['path'])
logger.debug(msg)
@@ -63,7 +63,7 @@ class RemoteHTTP(RemoteBase):
self._makedirs(to_info['path'])
total = self._content_length(from_info['url'])
total = self._content_length(from_info['path'])
if no_progress_bar or not total:
cb = None
@@ -71,9 +71,9 @@ class RemoteHTTP(RemoteBase):
cb = ProgressBarCallback(name, total)
try:
self._download_to(from_info['url'], tmp_file, callback=cb)
self._download_to(from_info['path'], tmp_file, callback=cb)
except Exception as exc:
msg = "Failed to download '{}'".format(from_info['url'])
msg = "Failed to download '{}'".format(from_info['path'])
logger.warn(msg, exc)
continue
@@ -82,26 +82,24 @@ class RemoteHTTP(RemoteBase):
if not no_progress_bar:
progress.finish_target(name)
def exists(self, path_infos):
return [
bool(self._request('HEAD', path_info.get('url')))
for path_info in path_infos
]
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info['scheme'] in ['http', 'https']
return bool(self._request('HEAD', path_info.get('path')))
def cache_exists(self, md5s):
assert isinstance(md5s, list)
def func(md5):
return bool(self._request('HEAD', self.checksum_to_path(md5)))
return list(filter(func, md5s))
def save_info(self, path_info):
if path_info['scheme'] not in ['http', 'https']:
raise NotImplementedError
return {self.PARAM_ETAG: self._etag(path_info['url'])}
def md5s_to_path_infos(self, md5s):
return [
{
'scheme': 'http',
'url': posixpath.join(self.prefix, md5[0:2], md5[2:]),
}
for md5 in md5s
]
return {self.PARAM_ETAG: self._etag(path_info['path'])}
def _content_length(self, url):
return self._request('HEAD', url).headers.get('Content-Length')
@@ -141,3 +139,6 @@ class RemoteHTTP(RemoteBase):
return requests.request(method, url, **kwargs)
except requests.exceptions.RequestException:
raise DvcException('Could not perform a {} request'.format(method))
def gc(self):
raise NotImplementedError
dvc/remote/local.py
+150 -242
View File
@@ -7,9 +7,8 @@ import shutil
import posixpath
from operator import itemgetter
import dvc.prompt as prompt
from dvc.system import System
from dvc.remote.base import RemoteBase, STATUS_MAP
from dvc.remote.base import RemoteBase, STATUS_MAP, STATUS_NEW, STATUS_DELETED
from dvc.logger import logger
from dvc.utils import remove, move, copyfile, dict_md5, to_chunks
from dvc.utils import LARGE_DIR_SIZE
@@ -22,7 +21,7 @@ from concurrent.futures import ThreadPoolExecutor
class RemoteLOCAL(RemoteBase):
scheme = ''
REGEX = r'^(?P<path>(/+|.:\\+).*)$'
PARAM_MD5 = 'md5'
PARAM_CHECKSUM = 'md5'
PARAM_PATH = 'path'
PARAM_RELPATH = 'relpath'
MD5_DIR_SUFFIX = '.dir'
@@ -53,6 +52,8 @@ class RemoteLOCAL(RemoteBase):
if self.cache_dir is not None and not os.path.exists(self.cache_dir):
os.mkdir(self.cache_dir)
self.path_info = {'scheme': 'local'}
@staticmethod
def compat_config(config):
ret = config.copy()
@@ -68,7 +69,7 @@ class RemoteLOCAL(RemoteBase):
def prefix(self):
return self.cache_dir
def all(self):
def list_cache_paths(self):
clist = []
for entry in os.listdir(self.cache_dir):
subdir = os.path.join(self.cache_dir, entry)
@@ -76,8 +77,7 @@ class RemoteLOCAL(RemoteBase):
continue
for cache in os.listdir(subdir):
path = os.path.join(subdir, cache)
clist.append(self.path_to_md5(path))
clist.append(os.path.join(subdir, cache))
return clist
@@ -87,10 +87,6 @@ class RemoteLOCAL(RemoteBase):
return os.path.join(self.cache_dir, md5[0:2], md5[2:])
def path_to_md5(self, path):
relpath = os.path.relpath(path, self.cache_dir)
return os.path.dirname(relpath) + os.path.basename(relpath)
def changed_cache_file(self, md5):
cache = self.get(md5)
if self.state.changed(cache, md5=md5):
@@ -101,6 +97,11 @@ class RemoteLOCAL(RemoteBase):
return True
return False
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info['scheme'] == 'local'
return os.path.exists(path_info['path'])
def changed_cache(self, md5):
cache = self.get(md5)
clist = [(cache, md5)]
@@ -117,7 +118,7 @@ class RemoteLOCAL(RemoteBase):
continue
for entry in self.load_dir_cache(md5):
md5 = entry[self.PARAM_MD5]
md5 = entry[self.PARAM_CHECKSUM]
cache = self.get(md5)
clist.append((cache, md5))
@@ -160,8 +161,14 @@ class RemoteLOCAL(RemoteBase):
raise DvcException('No possible cache types left to try out.')
@property
def ospath(self):
if os.name == 'nt':
return ntpath
return posixpath
@classmethod
def ospath(cls, path):
def to_ospath(cls, path):
if os.name == 'nt':
return cls.ntpath(path)
return cls.unixpath(path)
@@ -204,7 +211,7 @@ class RemoteLOCAL(RemoteBase):
md5 = self.state.update(path)
dir_info.append({self.PARAM_RELPATH: relpath,
self.PARAM_MD5: md5})
self.PARAM_CHECKSUM: md5})
if bar:
progress.finish_target(title)
@@ -237,7 +244,7 @@ class RemoteLOCAL(RemoteBase):
return []
for info in d:
info['relpath'] = self.ospath(info['relpath'])
info['relpath'] = self.to_ospath(info['relpath'])
return d
@@ -262,36 +269,14 @@ class RemoteLOCAL(RemoteBase):
def is_dir_cache(cls, cache):
return cache.endswith(cls.MD5_DIR_SUFFIX)
def checkout(self, path_info, checksum_info, force=False):
def do_checkout(self, path_info, checksum, force=False):
path = path_info['path']
md5 = checksum_info.get(self.PARAM_MD5)
md5 = checksum
cache = self.get(md5)
if not cache:
msg = 'No cache info for \'{}\'. Skipping checkout.'
logger.warn(msg.format(os.path.relpath(path)))
return
if not self.changed(path_info, checksum_info):
msg = "Data '{}' didn't change."
logger.info(msg.format(os.path.relpath(path)))
return
if self.changed_cache(md5):
msg = u'Cache \'{}\' not found. File \'{}\' won\'t be created.'
logger.warn(msg.format(md5, os.path.relpath(path)))
remove(path)
return
msg = u'Checking out \'{}\' with cache \'{}\'.'
logger.info(msg.format(os.path.relpath(path), md5))
if not self.is_dir_cache(cache):
if os.path.exists(path):
if force or self._already_cached(path):
remove(path)
else:
self._safe_remove(path)
self.safe_remove(path_info, force=force)
self.link(cache, path)
self.state.update_link(path)
@@ -311,20 +296,17 @@ class RemoteLOCAL(RemoteBase):
for processed, entry in enumerate(dir_info):
relpath = entry[self.PARAM_RELPATH]
m = entry[self.PARAM_MD5]
m = entry[self.PARAM_CHECKSUM]
p = os.path.join(path, relpath)
c = self.get(m)
entry_info = {'scheme': path_info['scheme'], self.PARAM_PATH: p}
entry_checksum_info = {self.PARAM_MD5: m}
entry_checksum_info = {self.PARAM_CHECKSUM: m}
if self.changed(entry_info, entry_checksum_info):
if os.path.exists(p):
if force or self._already_cached(p):
remove(p)
else:
self._safe_remove(p)
self.safe_remove(entry_info, force=force)
self.link(c, p)
@@ -338,8 +320,10 @@ class RemoteLOCAL(RemoteBase):
if bar:
progress.finish_target(dir_relpath)
def _already_cached(self, path):
current_md5 = self.state.update(path)
def already_cached(self, path_info):
assert path_info['scheme'] in ['', 'local']
current_md5 = self.state.update(path_info['path'])
if not current_md5:
return False
@@ -361,24 +345,7 @@ class RemoteLOCAL(RemoteBase):
delta = working_dir_files - cached_files
for file in delta:
if force or self._already_cached(file):
remove(file)
else:
self._safe_remove(file)
def _safe_remove(self, file):
msg = (
'File "{}" is going to be removed. '
'Are you sure you want to proceed?'
.format(file)
)
if not prompt.confirm(msg):
raise DvcException('Unable to remove {} without a confirmation'
" from the user. Use '-f' to force."
.format(file))
remove(file)
self.safe_remove({'scheme': 'local', 'path': file}, force=force)
def _move(self, inp, outp):
# moving in two stages to make last the move atomic in
@@ -402,7 +369,7 @@ class RemoteLOCAL(RemoteBase):
self.link(cache, path)
self.state.update_link(path)
return {self.PARAM_MD5: md5}
return {self.PARAM_CHECKSUM: md5}
def _save_dir(self, path_info):
path = path_info['path']
@@ -415,7 +382,7 @@ class RemoteLOCAL(RemoteBase):
for processed, entry in enumerate(dir_info):
relpath = entry[self.PARAM_RELPATH]
m = entry[self.PARAM_MD5]
m = entry[self.PARAM_CHECKSUM]
p = os.path.join(path, relpath)
c = self.get(m)
@@ -434,7 +401,7 @@ class RemoteLOCAL(RemoteBase):
if bar:
progress.finish_target(dir_relpath)
return {self.PARAM_MD5: md5}
return {self.PARAM_CHECKSUM: md5}
def save(self, path_info):
if path_info['scheme'] != 'local':
@@ -455,7 +422,7 @@ class RemoteLOCAL(RemoteBase):
if path_info['scheme'] != 'local':
raise NotImplementedError
return {self.PARAM_MD5: self.state.update(path_info['path'])}
return {self.PARAM_CHECKSUM: self.state.update(path_info['path'])}
def changed(self, path_info, checksum_info):
"""
@@ -465,10 +432,10 @@ class RemoteLOCAL(RemoteBase):
- The checkusm stored in the State is different from the given one
- There's no file in the cache
"""
if not self.exists([path_info])[0]:
if not self.exists(path_info):
return True
md5 = checksum_info.get(self.PARAM_MD5, None)
md5 = checksum_info.get(self.PARAM_CHECKSUM, None)
if md5 is None:
return True
@@ -489,17 +456,9 @@ class RemoteLOCAL(RemoteBase):
move(from_info['path'], to_info['path'])
def md5s_to_path_infos(self, md5s):
return [{'scheme': 'local',
'path': os.path.join(self.prefix,
md5[0:2], md5[2:])} for md5 in md5s]
def exists(self, path_infos):
ret = []
for path_info in path_infos:
assert path_info['scheme'] == 'local'
ret.append(os.path.exists(path_info['path']))
return ret
def cache_exists(self, md5s):
assert isinstance(md5s, list)
return list(filter(lambda md5: not self.changed_cache_file(md5), md5s))
def upload(self, from_infos, to_infos, names=None):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -565,10 +524,10 @@ class RemoteLOCAL(RemoteBase):
by_md5 = {}
for info in checksum_infos:
md5 = info[self.PARAM_MD5]
md5 = info[self.PARAM_CHECKSUM]
if show_checksums:
by_md5[md5] = md5
by_md5[md5] = {'name': md5}
continue
name = info[self.PARAM_PATH]
@@ -577,187 +536,136 @@ class RemoteLOCAL(RemoteBase):
name += '({})'.format(branch)
if md5 not in by_md5.keys():
by_md5[md5] = ''
by_md5[md5] = {'name': name}
else:
by_md5[md5] += ' '
by_md5[md5]['name'] += ' ' + name
by_md5[md5] += name
return list(by_md5.keys()), list(by_md5.values())
def gc(self, checksum_infos):
checksum_infos = checksum_infos['local']
used_md5s = [info[self.PARAM_MD5] for info in checksum_infos]
removed = False
for md5 in self.all():
if md5 in used_md5s:
continue
remove(self.get(md5))
removed = True
return removed
return by_md5
def status(self, checksum_infos, remote, jobs=None, show_checksums=False):
logger.info("Preparing to pull data from {}".format(remote.url))
logger.info("Preparing to collect status from {}".format(remote.url))
title = "Collecting information"
ret = {}
progress.set_n_total(1)
progress.update_target(title, 0, 100)
progress.update_target(title, 10, 100)
md5s, names = self._group(checksum_infos,
show_checksums=show_checksums)
progress.update_target(title, 20, 100)
path_infos = remote.md5s_to_path_infos(md5s)
ret = self._group(checksum_infos, show_checksums=show_checksums)
md5s = list(ret.keys())
progress.update_target(title, 30, 100)
remote_exists = remote.exists(path_infos)
remote_exists = list(remote.cache_exists(md5s))
progress.update_target(title, 90, 100)
local_exists = [not self.changed_cache_file(md5) for md5 in md5s]
local_exists = self.cache_exists(md5s)
progress.finish_target(title)
return [(name, STATUS_MAP[l, r]) for name, l, r in zip(names,
local_exists,
remote_exists)]
for md5, info in ret.items():
info['status'] = STATUS_MAP[(md5 in local_exists,
md5 in remote_exists)]
return ret
def _get_chunks(self, download, remote, status_info, status, jobs):
title = "Analysing status."
progress.set_n_total(1)
total = len(status_info)
current = 0
cache = []
path_infos = []
names = []
for md5, info in status_info.items():
if info['status'] == status:
cache.append(self.checksum_to_path_info(md5))
path_infos.append(remote.checksum_to_path_info(md5))
names.append(info['name'])
current += 1
progress.update_target(title, current, total)
progress.finish_target(title)
progress.set_n_total(len(names))
if download:
to_infos = cache
from_infos = path_infos
else:
to_infos = path_infos
from_infos = cache
return list(zip(to_chunks(from_infos, jobs),
to_chunks(to_infos, jobs),
to_chunks(names, jobs)))
def _process(self,
checksum_infos,
remote,
jobs=None,
show_checksums=False,
download=False):
msg = "Preparing to {} data {} '{}'"
logger.info(msg.format('download' if download else 'upload',
'from' if download else 'to',
remote.url))
if download:
func = remote.download
status = STATUS_DELETED
else:
func = remote.upload
status = STATUS_NEW
if jobs is None:
jobs = remote.JOBS
status_info = self.status(checksum_infos,
remote,
jobs=jobs,
show_checksums=show_checksums)
chunks = self._get_chunks(download, remote, status_info, status, jobs)
if len(chunks) == 0:
return
futures = []
with ThreadPoolExecutor(max_workers=jobs) as executor:
for from_infos, to_infos, names in chunks:
res = executor.submit(func,
from_infos,
to_infos,
names=names)
futures.append(res)
for f in futures:
f.result()
def push(self,
checksum_infos,
remote,
jobs=None,
show_checksums=False):
self._process(checksum_infos,
remote,
jobs=jobs,
show_checksums=show_checksums,
download=False)
def pull(self,
checksum_infos,
remote,
jobs=None,
show_checksums=False):
title = "Collecting information"
progress.set_n_total(1)
progress.update_target(title, 0, 100)
grouped = zip(*self._group(checksum_infos,
show_checksums=show_checksums))
progress.update_target(title, 10, 100)
md5s = []
names = []
# NOTE: filter files that are not corrupted
for md5, name in grouped:
if self.changed_cache_file(md5):
md5s.append(md5)
names.append(name)
progress.update_target(title, 30, 100)
cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s]
progress.update_target(title, 50, 100)
path_infos = remote.md5s_to_path_infos(md5s)
progress.update_target(title, 60, 100)
# NOTE: dummy call to try to establish a connection
# to see if we need to ask user for a password.
remote.exists(remote.md5s_to_path_infos(['000']))
progress.update_target(title, 70, 100)
assert len(path_infos) == len(cache) == len(md5s) == len(names)
if jobs is None:
jobs = remote.JOBS
chunks = list(zip(to_chunks(path_infos, jobs),
to_chunks(cache, jobs),
to_chunks(names, jobs)))
progress.finish_target(title)
progress.set_n_total(len(names))
if len(chunks) == 0:
return
futures = []
with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
for from_infos, to_infos, names in chunks:
res = executor.submit(remote.download,
from_infos,
to_infos,
names=names)
futures.append(res)
for f in futures:
f.result()
def push(self, checksum_infos, remote, jobs=None, show_checksums=False):
logger.info("Preparing to push data to {}".format(remote.url))
title = "Collecting information"
progress.set_n_total(1)
progress.update_target(title, 0, 100)
# NOTE: verifying that our cache is not corrupted
def func(info):
return not self.changed_cache_file(info[self.PARAM_MD5])
checksum_infos = list(filter(func, checksum_infos))
progress.update_target(title, 20, 100)
# NOTE: filter files that are already uploaded
md5s = [i[self.PARAM_MD5] for i in checksum_infos]
exists = remote.exists(remote.md5s_to_path_infos(md5s))
progress.update_target(title, 30, 100)
def func(entry):
return not entry[0]
assert len(exists) == len(checksum_infos)
infos_exist = list(filter(func, zip(exists, checksum_infos)))
checksum_infos = [i for e, i in infos_exist]
progress.update_target(title, 70, 100)
md5s, names = self._group(checksum_infos,
show_checksums=show_checksums)
cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s]
progress.update_target(title, 80, 100)
path_infos = remote.md5s_to_path_infos(md5s)
assert len(path_infos) == len(cache) == len(md5s) == len(names)
progress.update_target(title, 90, 100)
if jobs is None:
jobs = remote.JOBS
chunks = list(zip(to_chunks(path_infos, jobs),
to_chunks(cache, jobs),
to_chunks(names, jobs)))
progress.finish_target(title)
progress.set_n_total(len(names))
if len(chunks) == 0:
return
futures = []
with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
for to_infos, from_infos, names in chunks:
res = executor.submit(remote.upload,
from_infos,
to_infos,
names=names)
futures.append(res)
for f in futures:
f.result()
self._process(checksum_infos,
remote,
jobs=jobs,
show_checksums=show_checksums,
download=True)
dvc/remote/s3.py
+41 -136
View File
@@ -1,6 +1,5 @@
import os
import threading
import posixpath
try:
import boto3
@@ -16,7 +15,6 @@ from dvc.logger import logger
from dvc.progress import progress
from dvc.config import Config
from dvc.remote.base import RemoteBase
from dvc.remote.local import RemoteLOCAL
from dvc.exceptions import DvcException
@@ -37,7 +35,7 @@ class RemoteS3(RemoteBase):
scheme = 's3'
REGEX = r'^s3://(?P<path>.*)$'
REQUIRES = {'boto3': boto3}
PARAM_ETAG = 'etag'
PARAM_CHECKSUM = 'etag'
def __init__(self, project, config):
self.project = project
@@ -66,6 +64,12 @@ class RemoteS3(RemoteBase):
if shared_creds:
os.environ.setdefault('AWS_SHARED_CREDENTIALS_FILE', shared_creds)
parsed = urlparse(self.url)
self.bucket = parsed.netloc
self.prefix = parsed.path.lstrip('/')
self.path_info = {'scheme': self.scheme, 'bucket': self.bucket}
@staticmethod
def compat_config(config):
ret = config.copy()
@@ -73,14 +77,6 @@ class RemoteS3(RemoteBase):
ret[Config.SECTION_REMOTE_URL] = url
return ret
@property
def bucket(self):
return urlparse(self.url).netloc
@property
def prefix(self):
return urlparse(self.url).path.lstrip('/')
@property
def s3(self):
session = boto3.session.Session(profile_name=self.profile,
@@ -90,11 +86,12 @@ class RemoteS3(RemoteBase):
endpoint_url=self.endpoint_url,
use_ssl=self.use_ssl)
def get_etag(self, bucket, key):
def get_etag(self, bucket, path):
try:
obj = self.s3.head_object(Bucket=bucket, Key=key)
obj = self.s3.head_object(Bucket=bucket, Key=path)
except Exception:
raise DvcException('s3://{}/{} does not exist'.format(bucket, key))
raise DvcException('s3://{}/{} does not exist'.format(bucket,
path))
return obj['ETag'].strip('"')
@@ -102,14 +99,14 @@ class RemoteS3(RemoteBase):
if path_info['scheme'] != 's3':
raise NotImplementedError
return {self.PARAM_ETAG: self.get_etag(path_info['bucket'],
path_info['key'])}
return {self.PARAM_CHECKSUM: self.get_etag(path_info['bucket'],
path_info['path'])}
def changed(self, path_info, checksum_info):
if not self.exists([path_info])[0]:
if not self.exists(path_info):
return True
etag = checksum_info.get(self.PARAM_ETAG, None)
etag = checksum_info.get(self.PARAM_CHECKSUM, None)
if etag is None:
return True
@@ -118,94 +115,38 @@ class RemoteS3(RemoteBase):
return checksum_info != self.save_info(path_info)
def _copy(self, from_info, to_info, s3=None):
def copy(self, from_info, to_info, s3=None):
s3 = s3 if s3 else self.s3
source = {'Bucket': from_info['bucket'],
'Key': from_info['key']}
self.s3.copy(source, to_info['bucket'], to_info['key'])
'Key': from_info['path']}
self.s3.copy(source, to_info['bucket'], to_info['path'])
def save(self, path_info):
if path_info['scheme'] != 's3':
raise NotImplementedError
etag = self.get_etag(path_info['bucket'], path_info['key'])
key = posixpath.join(self.prefix, etag[0:2], etag[2:])
to_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}
etag = self.get_etag(path_info['bucket'], path_info['path'])
path = self.checksum_to_path(etag)
to_info = {'scheme': 's3', 'bucket': self.bucket, 'path': path}
self._copy(path_info, to_info)
self.copy(path_info, to_info)
return {self.PARAM_ETAG: etag}
@staticmethod
def to_string(path_info):
return "s3://{}/{}".format(path_info['bucket'], path_info['key'])
def changed_cache(self, etag):
key = posixpath.join(self.prefix, etag[0:2], etag[2:])
cache = {'scheme': 's3', 'bucket': self.bucket, 'key': key}
if {self.PARAM_ETAG: etag} != self.save_info(cache):
if self.exists([cache])[0]:
msg = 'Corrupted cache file {}'
logger.warn(msg.format(self.to_string(cache)))
self.remove(cache)
return True
return False
def checkout(self, path_info, checksum_info):
if path_info['scheme'] != 's3':
raise NotImplementedError
etag = checksum_info.get(self.PARAM_ETAG, None)
if not etag:
return
if not self.changed(path_info, checksum_info):
msg = "Data '{}' didn't change."
logger.info(msg.format(self.to_string(path_info)))
return
if self.changed_cache(etag):
msg = "Cache '{}' not found. File '{}' won't be created."
logger.warn(msg.format(etag, self.to_string(path_info)))
return
if self.exists([path_info])[0]:
msg = "Data '{}' exists. Removing before checkout."
logger.warn(msg.format(self.to_string(path_info)))
self.remove(path_info)
return
msg = "Checking out '{}' with cache '{}'."
logger.info(msg.format(self.to_string(path_info), etag))
key = posixpath.join(self.prefix, etag[0:2], etag[2:])
from_info = {'scheme': 's3', 'bucket': self.bucket, 'key': key}
self._copy(from_info, path_info)
return {self.PARAM_CHECKSUM: etag}
def remove(self, path_info):
if path_info['scheme'] != 's3':
raise NotImplementedError
logger.debug('Removing s3://{}/{}'.format(path_info['bucket'],
path_info['key']))
path_info['path']))
self.s3.delete_object(Bucket=path_info['bucket'],
Key=path_info['key'])
Key=path_info['path'])
def md5s_to_path_infos(self, md5s):
return [{'scheme': self.scheme,
'bucket': self.bucket,
'key': posixpath.join(self.prefix,
md5[0:2], md5[2:])} for md5 in md5s]
def _all_keys(self, bucket, prefix):
def _list_paths(self, bucket, prefix):
s3 = self.s3
keys = []
kwargs = {'Bucket': bucket,
'Prefix': prefix}
while True:
@@ -219,7 +160,7 @@ class RemoteS3(RemoteBase):
break
for obj in contents:
keys.append(obj['Key'])
yield obj['Key']
token = resp.get('NextContinuationToken', None)
if not token:
@@ -227,26 +168,15 @@ class RemoteS3(RemoteBase):
kwargs['ContinuationToken'] = token
return keys
def list_cache_paths(self):
return self._list_paths(self.bucket, self.prefix)
def exists(self, path_infos):
ret = []
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info['scheme'] == 's3'
if len(path_infos) == 0:
return ret
bucket = path_infos[0]['bucket']
for path_info in path_infos:
assert path_info['scheme'] == self.scheme
assert path_info['bucket'] == bucket
exists = False
key = path_info['key']
if key in self._all_keys(bucket, key):
exists = True
ret.append(exists)
return ret
paths = self._list_paths(path_info['bucket'], path_info['path'])
return any(path_info['path'] == path for path in paths)
def upload(self, from_infos, to_infos, names=None):
names = self._verify_path_args(to_infos, from_infos, names)
@@ -262,7 +192,7 @@ class RemoteS3(RemoteBase):
logger.debug("Uploading '{}' to '{}/{}'".format(from_info['path'],
to_info['bucket'],
to_info['key']))
to_info['path']))
if not name:
name = os.path.basename(from_info['path'])
@@ -273,7 +203,7 @@ class RemoteS3(RemoteBase):
try:
s3.upload_file(from_info['path'],
to_info['bucket'],
to_info['key'],
to_info['path'],
Callback=cb)
except Exception as exc:
msg = "Failed to upload '{}'".format(from_info['path'])
@@ -296,14 +226,14 @@ class RemoteS3(RemoteBase):
raise NotImplementedError
if to_info['scheme'] == 's3':
self._copy(from_info, to_info, s3=s3)
self.copy(from_info, to_info, s3=s3)
continue
if to_info['scheme'] != 'local':
raise NotImplementedError
msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'],
from_info['key'],
from_info['path'],
to_info['path'])
logger.debug(msg)
@@ -319,16 +249,16 @@ class RemoteS3(RemoteBase):
else:
total = s3.head_object(
Bucket=from_info['bucket'],
Key=from_info['key'])['ContentLength']
Key=from_info['path'])['ContentLength']
cb = Callback(name, total)
s3.download_file(from_info['bucket'],
from_info['key'],
from_info['path'],
tmp_file,
Callback=cb)
except Exception as exc:
msg = "Failed to download '{}/{}'".format(from_info['bucket'],
from_info['key'])
from_info['path'])
logger.warn(msg, exc)
continue
@@ -336,28 +266,3 @@ class RemoteS3(RemoteBase):
if not no_progress_bar:
progress.finish_target(name)
def _path_to_etag(self, path):
relpath = posixpath.relpath(path, self.prefix)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all(self):
keys = self._all_keys(self.bucket, self.prefix)
return [self._path_to_etag(key) for key in keys]
def gc(self, cinfos):
used_etags = [info[self.PARAM_ETAG] for info in cinfos['s3']]
used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']]
removed = False
for etag in self._all():
if etag in used_etags:
continue
path_info = {'scheme': 's3',
'key': posixpath.join(self.prefix,
etag[0:2], etag[2:]),
'bucket': self.bucket}
self.remove(path_info)
removed = True
return removed
dvc/remote/ssh.py
+22 -119
View File
@@ -11,7 +11,6 @@ import dvc.prompt as prompt
from dvc.logger import logger
from dvc.progress import progress
from dvc.remote.base import RemoteBase, RemoteBaseCmdError
from dvc.remote.local import RemoteLOCAL
from dvc.config import Config
from dvc.exceptions import DvcException
@@ -52,7 +51,7 @@ class RemoteSSH(RemoteBase):
JOBS = 4
PARAM_MD5 = 'md5'
PARAM_CHECKSUM = 'md5'
DEFAULT_PORT = 22
TIMEOUT = 1800
@@ -73,13 +72,10 @@ class RemoteSSH(RemoteBase):
self.ask_password = config.get(Config.SECTION_REMOTE_ASK_PASSWORD,
False)
def md5s_to_path_infos(self, md5s):
return [{'scheme': 'ssh',
'host': self.host,
'user': self.user,
'port': self.port,
'path': posixpath.join(self.prefix,
md5[0:2], md5[2:])} for md5 in md5s]
self.path_info = {'scheme': 'ssh',
'host': self.host,
'user': self.user,
'port': self.port}
def ssh(self, host=None, user=None, port=None):
msg = "Establishing ssh connection with '{}' " \
@@ -105,33 +101,20 @@ class RemoteSSH(RemoteBase):
return ssh
def exists(self, path_infos):
ret = []
def exists(self, path_info):
assert not isinstance(path_info, list)
assert path_info['scheme'] == 'ssh'
if len(path_infos) == 0:
return ret
host = path_infos[0]['host']
user = path_infos[0]['user']
port = path_infos[0]['port']
ssh = self.ssh(host=host,
user=user,
port=port)
for path_info in path_infos:
assert host == path_info['host']
assert user == path_info['user']
assert port == path_info['port']
with self.ssh(path_info['host'],
path_info['user'],
path_info['port']) as ssh:
try:
self._exec(ssh, 'test -e {}'.format(path_info['path']))
exists = True
except RemoteSSHCmdError:
exists = False
ret.append(exists)
ssh.close()
return ret
return exists
def _exec(self, ssh, cmd):
stdin, stdout, stderr = ssh.exec_command(cmd)
@@ -205,7 +188,7 @@ class RemoteSSH(RemoteBase):
return md5
def cp(self, from_info, to_info, ssh=None):
def copy(self, from_info, to_info, ssh=None):
if from_info['scheme'] != 'ssh' or to_info['scheme'] != 'ssh':
raise NotImplementedError
@@ -227,37 +210,13 @@ class RemoteSSH(RemoteBase):
if path_info['scheme'] != 'ssh':
raise NotImplementedError
return {self.PARAM_MD5: self.md5(path_info)}
@staticmethod
def to_string(path_info):
return "{}://{}@{}:{}".format(path_info['scheme'],
path_info['user'],
path_info['host'],
path_info['path'])
def changed_cache(self, md5):
cache = {}
cache['scheme'] = 'ssh'
cache['host'] = self.host
cache['port'] = self.port
cache['user'] = self.user
cache['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:])
if {self.PARAM_MD5: md5} != self.save_info(cache):
if self.exists([cache])[0]:
msg = 'Corrupted cache file {}'
logger.warn(msg.format(self.to_string(cache)))
self.remove(cache)
return True
return False
return {self.PARAM_CHECKSUM: self.md5(path_info)}
def changed(self, path_info, checksum_info):
if not self.exists([path_info])[0]:
if not self.exists(path_info):
return True
md5 = checksum_info.get(self.PARAM_MD5, None)
md5 = checksum_info.get(self.PARAM_CHECKSUM, None)
if md5 is None:
return True
@@ -272,43 +231,11 @@ class RemoteSSH(RemoteBase):
md5 = self.md5(path_info)
dest = path_info.copy()
dest['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:])
dest['path'] = self.checksum_to_path(md5)
self.cp(path_info, dest)
self.copy(path_info, dest)
return {self.PARAM_MD5: md5}
def checkout(self, path_info, checksum_info):
if path_info['scheme'] != 'ssh':
raise NotImplementedError
md5 = checksum_info.get(self.PARAM_MD5, None)
if not md5:
return
if not self.changed(path_info, checksum_info):
msg = "Data '{}' didn't change."
logger.info(msg.format(self.to_string(path_info)))
return
if self.changed_cache(md5):
msg = "Cache '{}' not found. File '{}' won't be created."
logger.warn(msg.format(md5, self.to_string(path_info)))
return
if self.exists([path_info])[0]:
msg = "Data '{}' exists. Removing before checkout."
logger.warn(msg.format(self.to_string(path_info)))
self.remove(path_info)
return
msg = "Checking out '{}' with cache '{}'."
logger.info(msg.format(self.to_string(path_info), md5))
src = path_info.copy()
src['path'] = posixpath.join(self.prefix, md5[0:2], md5[2:])
self.cp(src, path_info)
return {self.PARAM_CHECKSUM: md5}
def remove(self, path_info):
if path_info['scheme'] != 'ssh':
@@ -343,7 +270,7 @@ class RemoteSSH(RemoteBase):
assert from_info['host'] == to_info['host']
assert from_info['port'] == to_info['port']
assert from_info['user'] == to_info['user']
self.cp(from_info, to_info, ssh=ssh)
self.copy(from_info, to_info, ssh=ssh)
continue
if to_info['scheme'] != 'local':
@@ -416,11 +343,7 @@ class RemoteSSH(RemoteBase):
sftp.close()
ssh.close()
def _path_to_md5(self, path):
relpath = posixpath.relpath(path, self.prefix)
return posixpath.dirname(relpath) + posixpath.basename(relpath)
def _all_md5s(self):
def list_cache_paths(self):
ssh = self.ssh(host=self.host,
user=self.user,
port=self.port)
@@ -428,24 +351,4 @@ class RemoteSSH(RemoteBase):
stdout = self._exec(ssh, cmd)
flist = stdout.split()
ssh.close()
return [self._path_to_md5(path) for path in flist]
def gc(self, cinfos):
used = [info[self.PARAM_MD5] for info in cinfos['ssh']]
used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']]
removed = False
for md5 in self._all_md5s():
if md5 in used:
continue
path_info = {'scheme': 'ssh',
'user': self.user,
'host': self.host,
'port': self.port,
'path': posixpath.join(self.prefix,
md5[0:2], md5[2:])}
self.remove(path_info)
removed = True
return removed
return flist
dvc/stage.py
+2 -2
View File
@@ -348,8 +348,8 @@ class Stage(object):
new_d.pop(self.PARAM_MD5, None)
outs = old_d.get(self.PARAM_OUTS, [])
for out in outs:
out.pop(RemoteLOCAL.PARAM_MD5, None)
out.pop(RemoteS3.PARAM_ETAG, None)
out.pop(RemoteLOCAL.PARAM_CHECKSUM, None)
out.pop(RemoteS3.PARAM_CHECKSUM, None)
return old_d == new_d
tests/test_cache.py
+3 -3
View File
@@ -22,10 +22,10 @@ class TestCache(TestDvc):
self.create(self.cache2, '2')
def test_all(self):
md5_list = Cache(self.dvc).local.all()
md5_list = list(Cache(self.dvc).local.all())
self.assertEquals(len(md5_list), 2)
self.assertTrue(self.cache1_md5 in md5_list)
self.assertTrue(self.cache2_md5 in md5_list)
self.assertIn(self.cache1_md5, md5_list)
self.assertIn(self.cache2_md5, md5_list)
def test_get(self):
cache = Cache(self.dvc).local.get(self.cache1_md5)
tests/test_checkout.py
+3 -3
View File
@@ -81,7 +81,7 @@ class TestCheckoutCorruptedCacheDir(TestDvc):
# NOTE: modifying cache file for one of the files inside the directory
# to check if dvc will detect that the cache is corrupted.
entry = self.dvc.cache.local.load_dir_cache(out.md5)[0]
md5 = entry[self.dvc.cache.local.PARAM_MD5]
md5 = entry[self.dvc.cache.local.PARAM_CHECKSUM]
cache = self.dvc.cache.local.get(md5)
with open(cache, 'w+') as fobj:
@@ -281,8 +281,8 @@ class TestCheckoutMissingMd5InStageFile(TestRepro):
with open(self.file1_stage, 'r') as fd:
d = yaml.load(fd)
del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5])
del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5])
del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_CHECKSUM])
del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_CHECKSUM])
with open(self.file1_stage, 'w') as fd:
yaml.dump(d, fd)
tests/test_data_cloud.py
+17 -9
View File
@@ -233,10 +233,12 @@ class TestDataCloudBase(TestDvc):
with self.cloud.project.state:
# Check status
status = self.cloud.status([info], show_checksums=True)
self.assertEqual([(md5, STATUS_NEW)], status)
expected = {md5: {'name': md5, 'status': STATUS_NEW}}
self.assertEqual(status, expected)
status_dir = self.cloud.status([info_dir], show_checksums=True)
self.assertTrue((md5_dir, STATUS_NEW) in status_dir)
expected = {md5_dir: {'name': md5_dir, 'status': STATUS_NEW}}
self.assertEqual(status_dir, expected)
# Push and check status
self.cloud.push([info])
@@ -247,19 +249,23 @@ class TestDataCloudBase(TestDvc):
self.assertTrue(os.path.isfile(cache_dir))
status = self.cloud.status([info], show_checksums=True)
self.assertEqual([(md5, STATUS_OK)], status)
expected = {md5: {'name': md5, 'status': STATUS_OK}}
self.assertEqual(status, expected)
status_dir = self.cloud.status([info_dir], show_checksums=True)
self.assertTrue((md5_dir, STATUS_OK) in status_dir)
expected = {md5_dir: {'name': md5_dir, 'status': STATUS_OK}}
self.assertEqual(status_dir, expected)
# Remove and check status
shutil.rmtree(self.dvc.cache.local.cache_dir)
status = self.cloud.status([info], show_checksums=True)
self.assertEqual([(md5, STATUS_DELETED)], status)
expected = {md5: {'name': md5, 'status': STATUS_DELETED}}
self.assertEqual(status, expected)
status_dir = self.cloud.status([info_dir], show_checksums=True)
self.assertEqual([(md5_dir, STATUS_DELETED)], status_dir)
expected = {md5_dir: {'name': md5_dir, 'status': STATUS_DELETED}}
self.assertEqual(status_dir, expected)
# Pull and check status
self.cloud.pull([info])
@@ -272,10 +278,12 @@ class TestDataCloudBase(TestDvc):
self.assertTrue(os.path.isfile(cache_dir))
status = self.cloud.status([info], show_checksums=True)
self.assertEqual([(md5, STATUS_OK)], status)
expected = {md5: {'name': md5, 'status': STATUS_OK}}
self.assertEqual(status, expected)
status_dir = self.cloud.status([info_dir], show_checksums=True)
self.assertTrue((md5_dir, STATUS_OK) in status_dir)
expected = {md5_dir: {'name': md5_dir, 'status': STATUS_OK}}
self.assertTrue(status_dir, expected)
def test(self):
self._ensure_should_run()
@@ -421,7 +429,7 @@ class TestDataCloudCLIBase(TestDvc):
self.main(['fetch'] + args)
self.main(['pull'] + args)
self.main(['pull', '-f'] + args)
self.assertTrue(os.path.exists(cache))
self.assertTrue(os.path.isfile(cache))
self.assertTrue(os.path.isfile(cache_dir))
tests/test_repro.py
+2 -2
View File
@@ -544,8 +544,8 @@ class TestReproMissingMd5InStageFile(TestRepro):
with open(self.file1_stage, 'r') as fd:
d = yaml.load(fd)
del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_MD5])
del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_MD5])
del(d[Stage.PARAM_OUTS][0][RemoteLOCAL.PARAM_CHECKSUM])
del(d[Stage.PARAM_DEPS][0][RemoteLOCAL.PARAM_CHECKSUM])
with open(self.file1_stage, 'w') as fd:
yaml.dump(d, fd)
tests/test_stage.py
+4 -2
View File
@@ -52,8 +52,10 @@ class TestSchemaDepsOuts(TestSchema):
Stage.validate(d)
def test_list(self):
lst = [{OutputLOCAL.PARAM_PATH: 'foo', RemoteLOCAL.PARAM_MD5: '123'},
{OutputLOCAL.PARAM_PATH: 'bar', RemoteLOCAL.PARAM_MD5: None},
lst = [{OutputLOCAL.PARAM_PATH: 'foo',
RemoteLOCAL.PARAM_CHECKSUM: '123'},
{OutputLOCAL.PARAM_PATH: 'bar',
RemoteLOCAL.PARAM_CHECKSUM: None},
{OutputLOCAL.PARAM_PATH: 'baz'}]
d = {Stage.PARAM_DEPS: lst}
Stage.validate(d)