Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-02 07:18:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #427 from efiop/master

Browse Source 
Remove show-workflow and refactor data_cloud
This commit is contained in:
Ruslan Kuprieiev
2018-01-15 21:16:41 +02:00
committed by GitHub
parent 3e53a0c2c0 a130a25e39
commit 94fa641290
25 changed files with 888 additions and 1514 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/cli.py
-31
View File
@@ -13,7 +13,6 @@ from dvc.command.repro import CmdRepro
from dvc.command.data_sync import CmdDataPush, CmdDataPull, CmdDataStatus
from dvc.command.gc import CmdGC
from dvc.command.add import CmdAdd
from dvc.command.show_workflow import CmdShowWorkflow
from dvc.command.config import CmdConfig
from dvc.command.show_pipeline import CmdShowPipeline
from dvc.command.checkout import CmdCheckout
@@ -223,36 +222,6 @@ def parse_args(argv=None):
help='Target data directory')
pipeline_parser.set_defaults(func=CmdShowPipeline)
workflow_parser = show_subparsers.add_parser(
'workflow',
parents=[parent_parser],
help='Show workflow image. It collapses DVC repro commits if possible.')
workflow_parser.add_argument(
'target',
nargs='?',
help='Target metric file')
workflow_parser.add_argument(
'-d',
'--dvc-commits',
action='store_true',
default=False,
help='Show DVC repro commits.')
workflow_parser.add_argument(
'-a',
'--all-commits',
action='store_true',
default=False,
help='Show all commits')
workflow_parser.add_argument(
'-m',
'--max-commits',
metavar='M',
type=int,
default=4,
help='Max commits per graph vertex. 4 by default.')
workflow_parser.set_defaults(func=CmdShowWorkflow)
# Checkout
checkout_parser = subparsers.add_parser(
'checkout',
dvc/cloud/aws.py
+326
View File
@@ -0,0 +1,326 @@
import os
import math
from boto.s3.connection import S3Connection
try:
import httplib
except ImportError:
# Python3 workaround for ResumableDownloadHandler.
# See https://github.com/boto/boto/pull/3755.
import sys
import http.client as httplib
sys.modules['httplib'] = httplib
from boto.s3.resumable_download_handler import ResumableDownloadHandler
from dvc.logger import Logger
from dvc.progress import progress
from dvc.cloud.credentials_aws import AWSCredentials
from dvc.cloud.base import DataCloudError, DataCloudBase
from dvc.utils import file_md5
def sizeof_fmt(num, suffix='B'):
""" Convert number of bytes to human-readable string """
for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']:
if abs(num) < 1024.0:
return "%3.1f%s%s" % (num, unit, suffix)
num /= 1024.0
return "%.1f%s%s" % (num, 'Y', suffix)
def percent_cb(name, part_complete, part_total, offset=0, multipart_total=None):
""" Callback for updating target progress """
complete = offset + part_complete
total = multipart_total if multipart_total != None else part_total
Logger.debug('{}: {} transferred out of {}'.format(name,
sizeof_fmt(complete),
sizeof_fmt(total)))
progress.update_target(os.path.basename(name), complete, total)
def create_cb(name, offset=0, multipart_total=None):
""" Create callback function for multipart object """
return (lambda cur, tot: percent_cb(name, cur, tot, offset, multipart_total))
class DataCloudAWS(DataCloudBase):
""" DataCloud class for Amazon Web Services """
def __init__(self, cloud_settings):
super(DataCloudAWS, self).__init__(cloud_settings)
self._aws_creds = AWSCredentials(cloud_settings.cloud_config)
@property
def aws_region_host(self):
""" get the region host needed for s3 access
See notes http://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region
"""
region = self._cloud_settings.cloud_config['Region']
if region is None or region == '':
return 's3.amazonaws.com'
if region == 'us-east-1':
return 's3.amazonaws.com'
return 's3.%s.amazonaws.com' % region
def credential_paths(self, default):
"""
Try obtaining path to aws credentials from config file.
"""
paths = []
credpath = self._cloud_settings.cloud_config.get('CredentialPath', None)
if credpath is not None and len(credpath) > 0:
credpath = os.path.expanduser(credpath)
if os.path.isfile(credpath):
paths.append(credpath)
else:
Logger.warn('AWS CredentialPath "%s" not found;'
'falling back to default "%s"' % (credpath, default))
paths.append(default)
else:
paths.append(default)
return paths
def _get_bucket_aws(self, bucket_name):
""" get a bucket object, aws """
if all([self._aws_creds.access_key_id,
self._aws_creds.secret_access_key,
self.aws_region_host]):
conn = S3Connection(self._aws_creds.access_key_id,
self._aws_creds.secret_access_key,
host=self.aws_region_host)
else:
conn = S3Connection()
bucket = conn.lookup(bucket_name)
if bucket is None:
raise DataCloudError('Storage path {} is not setup correctly'.format(bucket_name))
return bucket
@staticmethod
def _cmp_checksum(key, fname):
"""
Verify local and remote checksums. Used 'dvc-md5' metadata if supported
or falls back to etag.
"""
md5_cloud = key.metadata.get('dvc-md5', None)
md5_local = file_md5(fname)[0]
if md5_cloud == None:
md5_cloud = key.etag[1:-1]
if md5_cloud == md5_local:
return True
return False
@staticmethod
def _upload_tracker(fname):
"""
File name for upload tracker.
"""
return fname + '.upload'
@staticmethod
def _download_tracker(fname):
"""
File name for download tracker.
"""
return fname + '.download'
def _import(self, bucket_name, key_name, fname):
bucket = self._get_bucket_aws(bucket_name)
tmp_file = self.tmp_file(fname)
name = os.path.basename(fname)
key = bucket.get_key(key_name)
if not key:
Logger.error('File "{}" does not exist in the cloud'.format(key_name))
return None
if self._cmp_checksum(key, fname):
Logger.debug('File "{}" matches with "{}".'.format(fname, key_name))
return fname
Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name,
key_name,
fname))
res_h = ResumableDownloadHandler(tracker_file_name=self._download_tracker(tmp_file),
num_retries=10)
try:
key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h)
except Exception as exc:
Logger.error('Failed to download "{}": {}'.format(key_name, exc))
return None
os.rename(tmp_file, fname)
progress.finish_target(name)
Logger.debug('Downloading completed')
return fname
def _write_upload_tracker(self, fname, mp_id):
"""
Write multipart id to upload tracker.
"""
try:
open(self._upload_tracker(fname), 'w+').write(mp_id)
except Exception as exc:
Logger.debug("Failed to write upload tracker file for {}: {}".format(fname, exc))
def _unlink_upload_tracker(self, fname):
"""
Remove upload tracker file.
"""
try:
os.unlink(self._upload_tracker(fname))
except Exception as exc:
Logger.debug("Failed to unlink upload tracker file for {}: {}".format(fname, exc))
def _resume_multipart(self, key, fname):
"""
Try resuming multipart upload.
"""
try:
mp_id = open(self._upload_tracker(fname), 'r').read()
except Exception as exc:
Logger.debug("Failed to read upload tracker file for {}: {}".format(fname, exc))
return None
for part in key.bucket.get_all_multipart_uploads():
if part.id != mp_id:
continue
Logger.debug("Found existing multipart {}".format(mp_id))
return part
return None
def _create_multipart(self, key, fname):
"""
Create multipart upload and save info to tracker file.
"""
# AWS doesn't provide easilly accessible md5 for multipart
# objects, so we have to store our own md5 sum to use later.
metadata = {'dvc-md5' : str(file_md5(fname)[0])}
multipart = key.bucket.initiate_multipart_upload(key.name, metadata=metadata)
self._write_upload_tracker(fname, multipart.id)
return multipart
def _get_multipart(self, key, fname):
"""
Try resuming multipart upload if supported.
"""
multipart = self._resume_multipart(key, fname)
if multipart != None:
return multipart
return self._create_multipart(key, fname)
@staticmethod
def _skip_part(multipart, part_num, size):
"""
Skip part of multipart upload if it has been already uploaded to the server.
"""
for part in multipart.get_all_parts():
if part.part_number == part_num and part.size == size:# and p.etag and p.last_modified
Logger.debug("Skipping part #{}".format(str(part_num)))
return True
return False
def _push_multipart(self, key, fname):
"""
Upload local file to cloud as a multipart upload.
"""
multipart = self._get_multipart(key, fname)
source_size = os.stat(fname).st_size
chunk_size = 50*1024*1024
chunk_count = int(math.ceil(source_size / float(chunk_size)))
with open(fname, 'rb') as fobj:
for i in range(chunk_count):
offset = i * chunk_size
left = source_size - offset
size = min([chunk_size, left])
part_num = i + 1
if self._skip_part(multipart, part_num, size):
continue
fobj.seek(offset)
multipart.upload_part_from_file(fp=fobj,
replace=False,
size=size,
num_cb=100,
part_num=part_num,
cb=create_cb(fname, offset, source_size))
if len(multipart.get_all_parts()) != chunk_count:
raise Exception("Couldn't upload all file parts")
multipart.complete_upload()
self._unlink_upload_tracker(fname)
def push(self, path):
""" push, aws version """
aws_key = self.cache_file_key(path)
bucket = self._get_bucket_aws(self.storage_bucket)
key = bucket.get_key(aws_key)
if key:
Logger.debug('File already uploaded to the cloud. Checksum validation...')
if self._cmp_checksum(key, path):
Logger.debug('File checksum matches. No uploading is needed.')
return path
Logger.debug('Checksum miss-match. Re-uploading is required.')
key = bucket.new_key(aws_key)
try:
self._push_multipart(key, path)
except Exception as exc:
Logger.error('Failed to upload "{}": {}'.format(path, exc))
return None
progress.finish_target(os.path.basename(path))
return path
def _status(self, path):
aws_key = self.cache_file_key(path)
bucket = self._get_bucket_aws(self.storage_bucket)
key = bucket.get_key(aws_key)
remote_exists = key is not None
local_exists = os.path.exists(path)
diff = None
if remote_exists and local_exists:
diff = self._cmp_checksum(key, path)
return (local_exists, remote_exists, diff)
def remove(self, path):
aws_file_name = self.cache_file_key(path)
Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name))
if not self._aws_creds.access_key_id or not self._aws_creds.secret_access_key:
Logger.debug('[Cmd-Remove] Unable to check cache file in the cloud')
return
conn = S3Connection(self._aws_creds.access_key_id, self._aws_creds.secret_access_key)
bucket_name = self.storage_bucket
bucket = conn.lookup(bucket_name)
if bucket:
key = bucket.get_key(aws_file_name)
if not key:
Logger.warn('[Cmd-Remove] S3 remove warning: '
'file "{}" does not exist in S3'.format(aws_file_name))
else:
key.delete()
Logger.info('[Cmd-Remove] File "{}" was removed from S3'.format(aws_file_name))
dvc/cloud/base.py
+118
View File
@@ -0,0 +1,118 @@
import os
from dvc.logger import Logger
from dvc.exceptions import DvcException
from dvc.config import ConfigError
STATUS_UNKNOWN = 0
STATUS_OK = 1
STATUS_MODIFIED = 2
STATUS_NEW = 3
STATUS_DELETED = 4
STATUS_MAP = {
# (local_exists, remote_exists, cmp)
(True, True, True) : STATUS_OK,
(True, True, False) : STATUS_MODIFIED,
(True, False, None) : STATUS_NEW,
(False, True, None) : STATUS_DELETED,
}
class DataCloudError(DvcException):
""" Data Cloud exception """
def __init__(self, msg):
super(DataCloudError, self).__init__('Data sync error: {}'.format(msg))
class DataCloudBase(object):
""" Base class for DataCloud """
def __init__(self, cloud_settings):
self._cloud_settings = cloud_settings
@property
def storage_path(self):
""" get storage path
Precedence: Storage, then cloud specific
"""
if self._cloud_settings.global_storage_path:
return self._cloud_settings.global_storage_path
path = self._cloud_settings.cloud_config.get('StoragePath', None)
if path is None:
raise ConfigError('invalid StoragePath: not set for Data or cloud specific')
return path
def _storage_path_parts(self):
"""
Split storage path into parts. I.e. 'dvc-test/myrepo' -> ['dvc', 'myrepo']
"""
return self.storage_path.strip('/').split('/', 1)
@property
def storage_bucket(self):
""" Data -> StoragePath takes precedence; if doesn't exist, use cloud-specific """
return self._storage_path_parts()[0]
@property
def storage_prefix(self):
"""
Prefix within the bucket. I.e. 'myrepo' in 'dvc-test/myrepo'.
"""
parts = self._storage_path_parts()
if len(parts) > 1:
return parts[1]
return ''
def cache_file_key(self, fname):
""" Key of a file within the bucket """
return '{}/{}'.format(self.storage_prefix, os.path.basename(fname)).strip('/')
@staticmethod
def tmp_file(fname):
""" Temporary name for a partial download """
return fname + '.part'
def sanity_check(self):
"""
Cloud-specific method to check config for basic requirements.
"""
pass
def _import(self, bucket, fin, fout):
"""
Cloud-specific method for importing data file.
"""
pass
def push(self, path):
""" Cloud-specific method for pushing data """
pass
def pull(self, path):
""" Generic method for pulling data from the cloud """
key_name = self.cache_file_key(path)
return self._import(self.storage_bucket, key_name, path)
def remove(self, path):
"""
Cloud-specific method for removing data item from the cloud.
"""
pass
def _status(self, path):
"""
Cloud-specific method for checking data item status.
"""
pass
def status(self, path):
"""
Generic method for checking data item status.
"""
return STATUS_MAP.get(self._status(path), STATUS_UNKNOWN)
dvc/cloud/data_cloud.py
+114
View File
@@ -0,0 +1,114 @@
from dvc.cloud.instance_manager import CloudSettings
from dvc.logger import Logger
from dvc.exceptions import DvcException
from dvc.config import ConfigError
from dvc.utils import map_progress
from dvc.cloud.aws import DataCloudAWS
from dvc.cloud.gcp import DataCloudGCP
from dvc.cloud.local import DataCloudLOCAL
from dvc.cloud.http import DataCloudHTTP
class DataCloud(object):
""" Generic class to do initial config parsing and redirect to proper DataCloud methods """
CLOUD_MAP = {
'AWS' : DataCloudAWS,
'GCP' : DataCloudGCP,
'HTTP' : DataCloudHTTP,
'LOCAL' : DataCloudLOCAL,
}
SCHEME_MAP = {
's3' : 'AWS',
'http' : 'HTTP',
'https' : 'HTTP',
'ftp' : 'HTTP',
'gs' : 'GCP',
'' : 'LOCAL',
}
def __init__(self, config):
self._config = config
cloud_type = self._config['Global'].get('Cloud', '').strip().upper()
if cloud_type not in self.CLOUD_MAP.keys():
raise ConfigError('Wrong cloud type %s specified' % cloud_type)
if cloud_type not in self._config.keys():
raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_type)
cloud_settings = self.get_cloud_settings(self._config,
cloud_type)
self.typ = cloud_type
self._cloud = self.CLOUD_MAP[cloud_type](cloud_settings)
self.sanity_check()
@staticmethod
def get_cloud_settings(config, cloud_type):
"""
Obtain cloud settings from config.
"""
if cloud_type not in config.keys():
cloud_config = None
else:
cloud_config = config[cloud_type]
global_storage_path = config['Global'].get('StoragePath', None)
cloud_settings = CloudSettings(global_storage_path, cloud_config)
return cloud_settings
def sanity_check(self):
""" sanity check a config
check that we have a cloud and storagePath
if aws, check can read credentials
if google, check ProjectName
Returns:
(T,) if good
(F, issues) if bad
"""
key = 'Cloud'
if key.lower() not in [k.lower() for k in self._config['Global'].keys()] or len(self._config['Global'][key]) < 1:
raise ConfigError('Please set %s in section Global in config file' % key)
# now that a cloud is chosen, can check StoragePath
storage_path = self._cloud.storage_path
if storage_path is None or len(storage_path) == 0:
raise ConfigError('Please set StoragePath = bucket/{optional path} '
'in config file in a cloud specific section')
self._cloud.sanity_check()
def _map_targets(self, func, targets, jobs):
"""
Process targets as data items in parallel.
"""
return map_progress(func, targets, jobs)
def sync(self, targets, jobs=1):
"""
Sync data items in a cloud-agnostic way.
"""
return self._map_targets(self._cloud.sync, targets, jobs)
def push(self, targets, jobs=1):
"""
Push data items in a cloud-agnostic way.
"""
return self._map_targets(self._cloud.push, targets, jobs)
def pull(self, targets, jobs=1):
"""
Pull data items in a cloud-agnostic way.
"""
return self._map_targets(self._cloud.pull, targets, jobs)
def status(self, targets, jobs=1):
"""
Check status of data items in a cloud-agnostic way.
"""
return self._map_targets(self._cloud.status, targets, jobs)
dvc/cloud/gcp.py
+127
View File
@@ -0,0 +1,127 @@
import base64
import os
from google.cloud import storage as gc
from dvc.logger import Logger
from dvc.config import ConfigError
from dvc.cloud.base import DataCloudError, DataCloudBase
from dvc.utils import file_md5
from dvc.progress import progress
class DataCloudGCP(DataCloudBase):
""" DataCloud class for Google Cloud Platform """
@property
def gc_project_name(self):
"""
Get project name from config.
"""
return self._cloud_settings.cloud_config.get('ProjectName', None)
def sanity_check(self):
project = self.gc_project_name
if project is None or len(project) < 1:
raise ConfigError('can\'t read google cloud project name. '
'Please set ProjectName in section GC.')
def _get_bucket_gc(self, storage_bucket):
""" get a bucket object, gc """
client = gc.Client(project=self.gc_project_name)
bucket = client.bucket(storage_bucket)
if not bucket.exists():
raise DataCloudError('sync up: google cloud bucket {} '
'doesn\'t exist'.format(self.storage_bucket))
return bucket
@staticmethod
def _cmp_checksum(blob, fname):
"""
Verify local and remote checksums.
"""
md5 = file_md5(fname)[1]
b64_encoded_md5 = base64.b64encode(md5).decode() if md5 else None
if blob.md5_hash == b64_encoded_md5:
return True
return False
def _import(self, bucket_name, key, fname):
bucket = self._get_bucket_gc(bucket_name)
name = os.path.basename(fname)
tmp_file = self.tmp_file(fname)
blob = bucket.get_blob(key)
if not blob:
Logger.error('File "{}" does not exist in the cloud'.format(key))
return None
if self._cmp_checksum(blob, fname):
Logger.debug('File "{}" matches with "{}".'.format(fname, key))
return fname
Logger.debug('Downloading cache file from gc "{}/{}"'.format(bucket.name, key))
# percent_cb is not available for download_to_filename, so
# lets at least update progress at keypoints(start, finish)
progress.update_target(name, 0, None)
try:
blob.download_to_filename(tmp_file)
except Exception as exc:
Logger.error('Failed to download "{}": {}'.format(key, exc))
return None
os.rename(tmp_file, fname)
progress.finish_target(name)
Logger.debug('Downloading completed')
return fname
def push(self, path):
""" push, gcp version """
bucket = self._get_bucket_gc(self.storage_bucket)
blob_name = self.cache_file_key(path)
name = os.path.basename(path)
blob = bucket.get_blob(blob_name)
if blob is not None and blob.exists():
if self._cmp_checksum(blob, path):
Logger.debug('checksum %s matches. Skipping upload' % path)
return path
Logger.debug('checksum %s mismatch. re-uploading' % path)
# same as in _import
progress.update_target(name, 0, None)
blob = bucket.blob(blob_name)
blob.upload_from_filename(path)
progress.finish_target(name)
Logger.debug('uploading %s completed' % path)
return path
def _status(self, path):
""" status, gcp version """
bucket = self._get_bucket_gc(self.storage_bucket)
blob_name = self.cache_file_key(path)
blob = bucket.get_blob(blob_name)
remote_exists = blob is not None and blob.exists()
local_exists = os.path.exists(path)
diff = None
if remote_exists and local_exists:
diff = self._cmp_checksum(blob, path)
return (local_exists, remote_exists, diff)
def remove(self, item):
raise Exception('NOT IMPLEMENTED YET')
dvc/cloud/http.py
+126
View File
@@ -0,0 +1,126 @@
import os
import requests
from dvc.logger import Logger
from dvc.progress import progress
from dvc.utils import copyfile
from dvc.cloud.base import DataCloudError, DataCloudBase
from dvc.utils import file_md5
class DataCloudHTTP(DataCloudBase):
"""
Driver for http cloud.
"""
def push(self, path):
raise Exception('Not implemented yet')
def pull(self, path):
raise Exception('Not implemented yet')
def remove(self, path):
raise Exception('Not implemented yet')
def status(self, path):
raise Exception('Not implemented yet')
@staticmethod
def _downloaded_size(fname):
"""
Check how much is already downloaded.
"""
if os.path.exists(fname):
downloaded = os.path.getsize(fname)
header = {'Range': 'bytes=%d-' % downloaded}
Logger.debug('found existing {} file, resuming download'.format(fname))
return (downloaded, header)
return (0, None)
@staticmethod
def _get_header(req, name):
"""
Get header value from request.
"""
val = req.headers.get(name)
if val == None:
Logger.debug('\'{}\' not supported by the server'.format(name))
return val
def _verify_downloaded_size(self, req, downloaded_size):
"""
Check that server supports resuming downloads.
"""
content_range = self._get_header(req, 'content-range')
if downloaded_size and content_range == None:
Logger.debug('Can\'t resume download')
return 0
return downloaded_size
def _download(self, req, fname, downloaded):
"""
Download file with progress bar.
"""
mode = 'ab' if downloaded else 'wb'
name = os.path.basename(req.url)
total_length = self._get_header(req, 'content-length')
chunk_size = 1024 * 100
progress.update_target(name, downloaded, total_length)
with open(fname, mode) as fobj:
for chunk in req.iter_content(chunk_size=chunk_size):
if not chunk: # filter out keep-alive new chunks
continue
fobj.write(chunk)
downloaded += len(chunk)
progress.update_target(name, downloaded, total_length)
progress.finish_target(name)
def _verify_md5(self, req, fname):
"""
Verify md5 of a downloaded file if server supports 'content-md5' header.
"""
md5 = file_md5(fname)[0]
content_md5 = self._get_header(req, 'content-md5')
if content_md5 == None:
return True
if md5 != content_md5:
Logger.error('Checksum mismatch')
return False
Logger.debug('Checksum matches')
return True
def import_data(self, url, path):
"""
Download single file from url.
"""
tmp_file = self.tmp_file(path)
downloaded, header = self._downloaded_size(tmp_file)
req = requests.get(url, stream=True, headers=header)
downloaded = self._verify_downloaded_size(req, downloaded)
try:
self._download(req, tmp_file, downloaded)
except Exception as exc:
Logger.error('Failed to download "{}": {}'.format(url, exc))
return None
if not self._verify_md5(req, tmp_file):
return None
os.rename(tmp_file, path)
return path
dvc/cloud/local.py
+53
View File
@@ -0,0 +1,53 @@
import os
import filecmp
from dvc.logger import Logger
from dvc.utils import copyfile, file_md5
from dvc.cloud.base import DataCloudError, DataCloudBase
class DataCloudLOCAL(DataCloudBase):
"""
Driver for local storage.
"""
def push(self, path):
Logger.debug('sync to cloud ' + path + " " + self.storage_path)
copyfile(path, self.storage_path)
return path
def _import(self, bucket, i, path):
inp = os.path.join(self.storage_path, i)
tmp_file = self.tmp_file(path)
try:
copyfile(inp, tmp_file)
except Exception as exc:
Logger.error('Failed to copy "{}": {}'.format(i, exc))
return None
os.rename(tmp_file, path)
return path
def pull(self, path):
Logger.debug('sync from cloud ' + path)
return self._import(None, path, path)
def remove(self, path):
Logger.debug('rm from cloud ' + path)
os.remove(path)
def import_data(self, path, out):
Logger.debug('import from cloud ' + path + " " + out)
return self._import(None, path, out)
def _status(self, path):
local = path
remote = '{}/{}'.format(self.storage_path, os.path.basename(local))
remote_exists = os.path.exists(remote)
local_exists = os.path.exists(local)
diff = None
if local_exists and remote_exists:
diff = filecmp.cmp(local, remote)
return (local_exists, remote_exists, diff)
dvc/command/data_sync.py
+1 -1
View File
@@ -1,4 +1,4 @@
import dvc.data_cloud as cloud
import dvc.cloud.data_cloud as cloud
from dvc.command.common.base import CmdBase
class CmdDataPull(CmdBase):
dvc/command/fsck.py
+1 -1
View File
@@ -2,7 +2,7 @@ import os
from dvc import stage
from dvc.command.common.base import CmdBase
from dvc.data_cloud import file_md5
from dvc.utils import file_md5
class FsckFileDep(object):
dvc/command/show_workflow.py
-90
View File
@@ -1,90 +0,0 @@
import re
import sys
from collections import defaultdict
from dvc.logger import Logger
from dvc.command.common.base import CmdBase
from dvc.graph.workflow import Workflow
from dvc.graph.commit import Commit
from dvc.executor import Executor, ExecutorError
from dvc.utils import parse_target_metric_file
class CmdShowWorkflow(CmdBase):
COMMIT_LEN = 7
LOG_SEPARATOR = '|'
LOG_FORMAT = ['%h', '%p', '%an', '%ai', '%s']
def get_all_commits(self, target):
# git log --all --abbrev=7 --pretty=format:"%h|%p|%an|%ai|%s"
try:
merges_map = self.get_merges_map()
format_str = self.LOG_SEPARATOR.join(self.LOG_FORMAT)
git_cmd = ['git', 'log', '--all', '--abbrev={}'.format(self.COMMIT_LEN),
'--pretty=format:{}'.format(format_str)]
lines = Executor.exec_cmd_only_success(git_cmd).split('\n')
branches_multimap = self.branches_multimap()
wf = Workflow(target, merges_map, branches_multimap)
for line in lines:
items = line.split(self.LOG_SEPARATOR, len(self.LOG_FORMAT))
assert len(items) == 5, 'Git wrapper: git log format has {} items, 5 expected'.format(len(items))
hash, parent_hash, name, date, comment = items
commit = Commit(hash, parent_hash, name, date, comment,
*self.was_target_changed(hash, target),
branch_tips=branches_multimap.get(hash))
wf.add_commit(commit)
return wf
except ExecutorError:
raise
def was_target_changed(self, hash, target):
git_cmd = ['git', 'show', '--pretty=', '--name-only', hash]
changed_files = set(Executor.exec_cmd_only_success(git_cmd).split('\n'))
if target not in changed_files:
return False, None
metric = parse_target_metric_file(target)
if metric is None:
return False, None
return True, metric
def get_merges_map(self):
# git log --merges --all --abbrev=7
# {'a4b56f1': back_to_600_est}
git_cmd = ['git', 'log', '--all', '--merges',
'--abbrev={}'.format(self.COMMIT_LEN)]
# lines = map(str.strip, Executor.exec_cmd_only_success(git_cmd).split('\n'))
# lines.map
return {}
def branches_multimap(self):
git_cmd = ['git', 'show-ref', '--abbrev={}'.format(self.COMMIT_LEN)]
lines = Executor.exec_cmd_only_success(git_cmd).split('\n')
func = unicode.split if sys.version_info[0] < 3 else str.split
items_full = map(func, lines)
items = map(lambda it: (it[0], re.sub(r'^refs/heads/', '', it[1])), items_full)
result = defaultdict(list)
for (hash, branch) in items:
result[hash].append(branch)
return result
def run(self):
target = self.args.target
if not target:
target = self.project.config._config['Global'].get('Target', '')
self.project.logger.debug(u'Set show workflow target as {}'.format(target))
wf = self.get_all_commits(target)
wf.build_graph(self.args.dvc_commits,
self.args.all_commits,
self.args.max_commits)
return 0
dvc/data_cloud.py
-848
View File
@@ -1,848 +0,0 @@
"""
Data cloud(Local, AWS, GCP) drivers.
"""
import base64
import hashlib
import os
import threading
import requests
import filecmp
import math
from boto.s3.connection import S3Connection
try:
import httplib
except ImportError:
# Python3 workaround for ResumableDownloadHandler.
# See https://github.com/boto/boto/pull/3755.
import sys
import http.client as httplib
sys.modules['httplib'] = httplib
from boto.s3.resumable_download_handler import ResumableDownloadHandler
from google.cloud import storage as gc
from dvc.cloud.instance_manager import CloudSettings
from dvc.logger import Logger
from dvc.exceptions import DvcException
from dvc.config import ConfigError
from dvc.progress import progress
from dvc.utils import copyfile
from dvc.cloud.credentials_aws import AWSCredentials
from dvc.system import System
from dvc.utils import map_progress
STATUS_UNKNOWN = 0
STATUS_OK = 1
STATUS_MODIFIED = 2
STATUS_NEW = 3
STATUS_DELETED = 4
STATUS_MAP = {
# (local_exists, remote_exists, cmp)
(True, True, True) : STATUS_OK,
(True, True, False) : STATUS_MODIFIED,
(True, False, None) : STATUS_NEW,
(False, True, None) : STATUS_DELETED,
}
class DataCloudError(DvcException):
""" Data Cloud exception """
def __init__(self, msg):
super(DataCloudError, self).__init__('Data sync error: {}'.format(msg))
def sizeof_fmt(num, suffix='B'):
""" Convert number of bytes to human-readable string """
for unit in ['', 'K', 'M', 'G', 'T', 'P', 'E', 'Z']:
if abs(num) < 1024.0:
return "%3.1f%s%s" % (num, unit, suffix)
num /= 1024.0
return "%.1f%s%s" % (num, 'Y', suffix)
def percent_cb(name, part_complete, part_total, offset=0, multipart_total=None):
""" Callback for updating target progress """
complete = offset + part_complete
total = multipart_total if multipart_total != None else part_total
Logger.debug('{}: {} transferred out of {}'.format(name,
sizeof_fmt(complete),
sizeof_fmt(total)))
progress.update_target(os.path.basename(name), complete, total)
def create_cb(name, offset=0, multipart_total=None):
""" Create callback function for multipart object """
return (lambda cur, tot: percent_cb(name, cur, tot, offset, multipart_total))
def file_md5(fname):
""" get the (md5 hexdigest, md5 digest) of a file """
if os.path.exists(fname):
hash_md5 = hashlib.md5()
with open(fname, "rb") as fobj:
for chunk in iter(lambda: fobj.read(1024*1000), b""):
hash_md5.update(chunk)
return (hash_md5.hexdigest(), hash_md5.digest())
else:
return (None, None)
class DataCloudBase(object):
""" Base class for DataCloud """
def __init__(self, cloud_settings):
self._cloud_settings = cloud_settings
self._lock = threading.Lock()
@property
def storage_path(self):
""" get storage path
Precedence: Storage, then cloud specific
"""
if self._cloud_settings.global_storage_path:
return self._cloud_settings.global_storage_path
path = self._cloud_settings.cloud_config.get('StoragePath', None)
if path is None:
raise ConfigError('invalid StoragePath: not set for Data or cloud specific')
return path
def _storage_path_parts(self):
"""
Split storage path into parts. I.e. 'dvc-test/myrepo' -> ['dvc', 'myrepo']
"""
return self.storage_path.strip('/').split('/', 1)
@property
def storage_bucket(self):
""" Data -> StoragePath takes precedence; if doesn't exist, use cloud-specific """
return self._storage_path_parts()[0]
@property
def storage_prefix(self):
"""
Prefix within the bucket. I.e. 'myrepo' in 'dvc-test/myrepo'.
"""
parts = self._storage_path_parts()
if len(parts) > 1:
return parts[1]
return ''
def cache_file_key(self, fname):
""" Key of a file within the bucket """
return '{}/{}'.format(self.storage_prefix, os.path.basename(fname)).strip('/')
@staticmethod
def tmp_file(fname):
""" Temporary name for a partial download """
return fname + '.part'
def sanity_check(self):
"""
Cloud-specific method to check config for basic requirements.
"""
pass
def _import(self, bucket, fin, fout):
"""
Cloud-specific method for importing data file.
"""
pass
def push(self, path):
""" Cloud-specific method for pushing data """
pass
def pull(self, path):
""" Generic method for pulling data from the cloud """
key_name = self.cache_file_key(path)
return self._import(self.storage_bucket, key_name, path)
def remove(self, path):
"""
Cloud-specific method for removing data item from the cloud.
"""
pass
def _status(self, path):
"""
Cloud-specific method for checking data item status.
"""
pass
def status(self, path):
"""
Generic method for checking data item status.
"""
return STATUS_MAP.get(self._status(path), STATUS_UNKNOWN)
class DataCloudLOCAL(DataCloudBase):
"""
Driver for local storage.
"""
def push(self, path):
Logger.debug('sync to cloud ' + path + " " + self.storage_path)
copyfile(path, self.storage_path)
return path
def _import(self, bucket, i, path):
inp = os.path.join(self.storage_path, i)
tmp_file = self.tmp_file(path)
try:
copyfile(inp, tmp_file)
except Exception as exc:
Logger.error('Failed to copy "{}": {}'.format(i, exc))
return None
os.rename(tmp_file, path)
return path
def pull(self, path):
Logger.debug('sync from cloud ' + path)
return self._import(None, path, path)
def remove(self, path):
Logger.debug('rm from cloud ' + path)
os.remove(path)
def import_data(self, path, out):
Logger.debug('import from cloud ' + path + " " + out)
return self._import(None, path, out)
def _status(self, path):
local = path
remote = '{}/{}'.format(self.storage_path, os.path.basename(local))
remote_exists = os.path.exists(remote)
local_exists = os.path.exists(local)
diff = None
if local_exists and remote_exists:
diff = filecmp.cmp(local, remote)
return (local_exists, remote_exists, diff)
class DataCloudHTTP(DataCloudBase):
"""
Driver for http cloud.
"""
def push(self, path):
raise Exception('Not implemented yet')
def pull(self, path):
raise Exception('Not implemented yet')
def remove(self, path):
raise Exception('Not implemented yet')
def status(self, path):
raise Exception('Not implemented yet')
@staticmethod
def _downloaded_size(fname):
"""
Check how much is already downloaded.
"""
if os.path.exists(fname):
downloaded = os.path.getsize(fname)
header = {'Range': 'bytes=%d-' % downloaded}
Logger.debug('found existing {} file, resuming download'.format(fname))
return (downloaded, header)
return (0, None)
@staticmethod
def _get_header(req, name):
"""
Get header value from request.
"""
val = req.headers.get(name)
if val == None:
Logger.debug('\'{}\' not supported by the server'.format(name))
return val
def _verify_downloaded_size(self, req, downloaded_size):
"""
Check that server supports resuming downloads.
"""
content_range = self._get_header(req, 'content-range')
if downloaded_size and content_range == None:
Logger.debug('Can\'t resume download')
return 0
return downloaded_size
def _download(self, req, fname, downloaded):
"""
Download file with progress bar.
"""
mode = 'ab' if downloaded else 'wb'
name = os.path.basename(req.url)
total_length = self._get_header(req, 'content-length')
chunk_size = 1024 * 100
progress.update_target(name, downloaded, total_length)
with open(fname, mode) as fobj:
for chunk in req.iter_content(chunk_size=chunk_size):
if not chunk: # filter out keep-alive new chunks
continue
fobj.write(chunk)
downloaded += len(chunk)
progress.update_target(name, downloaded, total_length)
progress.finish_target(name)
def _verify_md5(self, req, fname):
"""
Verify md5 of a downloaded file if server supports 'content-md5' header.
"""
md5 = file_md5(fname)[0]
content_md5 = self._get_header(req, 'content-md5')
if content_md5 == None:
return True
if md5 != content_md5:
Logger.error('Checksum mismatch')
return False
Logger.debug('Checksum matches')
return True
def import_data(self, url, path):
"""
Download single file from url.
"""
tmp_file = self.tmp_file(path)
downloaded, header = self._downloaded_size(tmp_file)
req = requests.get(url, stream=True, headers=header)
downloaded = self._verify_downloaded_size(req, downloaded)
try:
self._download(req, tmp_file, downloaded)
except Exception as exc:
Logger.error('Failed to download "{}": {}'.format(url, exc))
return None
if not self._verify_md5(req, tmp_file):
return None
os.rename(tmp_file, path)
return path
class DataCloudAWS(DataCloudBase):
""" DataCloud class for Amazon Web Services """
def __init__(self, cloud_settings):
super(DataCloudAWS, self).__init__(cloud_settings)
self._aws_creds = AWSCredentials(cloud_settings.cloud_config)
@property
def aws_region_host(self):
""" get the region host needed for s3 access
See notes http://docs.aws.amazon.com/general/latest/gr/rande.html#s3_region
"""
region = self._cloud_settings.cloud_config['Region']
if region is None or region == '':
return 's3.amazonaws.com'
if region == 'us-east-1':
return 's3.amazonaws.com'
return 's3.%s.amazonaws.com' % region
def credential_paths(self, default):
"""
Try obtaining path to aws credentials from config file.
"""
paths = []
credpath = self._cloud_settings.cloud_config.get('CredentialPath', None)
if credpath is not None and len(credpath) > 0:
credpath = os.path.expanduser(credpath)
if os.path.isfile(credpath):
paths.append(credpath)
else:
Logger.warn('AWS CredentialPath "%s" not found;'
'falling back to default "%s"' % (credpath, default))
paths.append(default)
else:
paths.append(default)
return paths
def _get_bucket_aws(self, bucket_name):
""" get a bucket object, aws """
if all([self._aws_creds.access_key_id,
self._aws_creds.secret_access_key,
self.aws_region_host]):
conn = S3Connection(self._aws_creds.access_key_id,
self._aws_creds.secret_access_key,
host=self.aws_region_host)
else:
conn = S3Connection()
bucket = conn.lookup(bucket_name)
if bucket is None:
raise DataCloudError('Storage path {} is not setup correctly'.format(bucket_name))
return bucket
@staticmethod
def _cmp_checksum(key, fname):
"""
Verify local and remote checksums. Used 'dvc-md5' metadata if supported
or falls back to etag.
"""
md5_cloud = key.metadata.get('dvc-md5', None)
md5_local = file_md5(fname)[0]
if md5_cloud == None:
md5_cloud = key.etag[1:-1]
if md5_cloud == md5_local:
return True
return False
@staticmethod
def _upload_tracker(fname):
"""
File name for upload tracker.
"""
return fname + '.upload'
@staticmethod
def _download_tracker(fname):
"""
File name for download tracker.
"""
return fname + '.download'
def _import(self, bucket_name, key_name, fname):
bucket = self._get_bucket_aws(bucket_name)
tmp_file = self.tmp_file(fname)
name = os.path.basename(fname)
key = bucket.get_key(key_name)
if not key:
Logger.error('File "{}" does not exist in the cloud'.format(key_name))
return None
if self._cmp_checksum(key, fname):
Logger.debug('File "{}" matches with "{}".'.format(fname, key_name))
return fname
Logger.debug('Downloading cache file from S3 "{}/{}" to "{}"'.format(bucket.name,
key_name,
fname))
res_h = ResumableDownloadHandler(tracker_file_name=self._download_tracker(tmp_file),
num_retries=10)
try:
key.get_contents_to_filename(tmp_file, cb=create_cb(name), res_download_handler=res_h)
except Exception as exc:
Logger.error('Failed to download "{}": {}'.format(key_name, exc))
return None
os.rename(tmp_file, fname)
progress.finish_target(name)
Logger.debug('Downloading completed')
return fname
def _write_upload_tracker(self, fname, mp_id):
"""
Write multipart id to upload tracker.
"""
try:
open(self._upload_tracker(fname), 'w+').write(mp_id)
except Exception as exc:
Logger.debug("Failed to write upload tracker file for {}: {}".format(fname, exc))
def _unlink_upload_tracker(self, fname):
"""
Remove upload tracker file.
"""
try:
os.unlink(self._upload_tracker(fname))
except Exception as exc:
Logger.debug("Failed to unlink upload tracker file for {}: {}".format(fname, exc))
def _resume_multipart(self, key, fname):
"""
Try resuming multipart upload.
"""
try:
mp_id = open(self._upload_tracker(fname), 'r').read()
except Exception as exc:
Logger.debug("Failed to read upload tracker file for {}: {}".format(fname, exc))
return None
for part in key.bucket.get_all_multipart_uploads():
if part.id != mp_id:
continue
Logger.debug("Found existing multipart {}".format(mp_id))
return part
return None
def _create_multipart(self, key, fname):
"""
Create multipart upload and save info to tracker file.
"""
# AWS doesn't provide easilly accessible md5 for multipart
# objects, so we have to store our own md5 sum to use later.
metadata = {'dvc-md5' : str(file_md5(fname)[0])}
multipart = key.bucket.initiate_multipart_upload(key.name, metadata=metadata)
self._write_upload_tracker(fname, multipart.id)
return multipart
def _get_multipart(self, key, fname):
"""
Try resuming multipart upload if supported.
"""
multipart = self._resume_multipart(key, fname)
if multipart != None:
return multipart
return self._create_multipart(key, fname)
@staticmethod
def _skip_part(multipart, part_num, size):
"""
Skip part of multipart upload if it has been already uploaded to the server.
"""
for part in multipart.get_all_parts():
if part.part_number == part_num and part.size == size:# and p.etag and p.last_modified
Logger.debug("Skipping part #{}".format(str(part_num)))
return True
return False
def _push_multipart(self, key, fname):
"""
Upload local file to cloud as a multipart upload.
"""
multipart = self._get_multipart(key, fname)
source_size = os.stat(fname).st_size
chunk_size = 50*1024*1024
chunk_count = int(math.ceil(source_size / float(chunk_size)))
with open(fname, 'rb') as fobj:
for i in range(chunk_count):
offset = i * chunk_size
left = source_size - offset
size = min([chunk_size, left])
part_num = i + 1
if self._skip_part(multipart, part_num, size):
continue
fobj.seek(offset)
multipart.upload_part_from_file(fp=fobj,
replace=False,
size=size,
num_cb=100,
part_num=part_num,
cb=create_cb(fname, offset, source_size))
if len(multipart.get_all_parts()) != chunk_count:
raise Exception("Couldn't upload all file parts")
multipart.complete_upload()
self._unlink_upload_tracker(fname)
def push(self, path):
""" push, aws version """
aws_key = self.cache_file_key(path)
bucket = self._get_bucket_aws(self.storage_bucket)
key = bucket.get_key(aws_key)
if key:
Logger.debug('File already uploaded to the cloud. Checksum validation...')
if self._cmp_checksum(key, path):
Logger.debug('File checksum matches. No uploading is needed.')
return path
Logger.debug('Checksum miss-match. Re-uploading is required.')
key = bucket.new_key(aws_key)
try:
self._push_multipart(key, path)
except Exception as exc:
Logger.error('Failed to upload "{}": {}'.format(path, exc))
return None
progress.finish_target(os.path.basename(path))
return path
def _status(self, path):
aws_key = self.cache_file_key(path)
bucket = self._get_bucket_aws(self.storage_bucket)
key = bucket.get_key(aws_key)
remote_exists = key is not None
local_exists = os.path.exists(path)
diff = None
if remote_exists and local_exists:
diff = self._cmp_checksum(key, path)
return (local_exists, remote_exists, diff)
def remove(self, path):
aws_file_name = self.cache_file_key(path)
Logger.debug(u'[Cmd-Remove] Remove from cloud {}.'.format(aws_file_name))
if not self._aws_creds.access_key_id or not self._aws_creds.secret_access_key:
Logger.debug('[Cmd-Remove] Unable to check cache file in the cloud')
return
conn = S3Connection(self._aws_creds.access_key_id, self._aws_creds.secret_access_key)
bucket_name = self.storage_bucket
bucket = conn.lookup(bucket_name)
if bucket:
key = bucket.get_key(aws_file_name)
if not key:
Logger.warn('[Cmd-Remove] S3 remove warning: '
'file "{}" does not exist in S3'.format(aws_file_name))
else:
key.delete()
Logger.info('[Cmd-Remove] File "{}" was removed from S3'.format(aws_file_name))
class DataCloudGCP(DataCloudBase):
""" DataCloud class for Google Cloud Platform """
@property
def gc_project_name(self):
"""
Get project name from config.
"""
return self._cloud_settings.cloud_config.get('ProjectName', None)
def sanity_check(self):
project = self.gc_project_name
if project is None or len(project) < 1:
raise ConfigError('can\'t read google cloud project name. '
'Please set ProjectName in section GC.')
def _get_bucket_gc(self, storage_bucket):
""" get a bucket object, gc """
client = gc.Client(project=self.gc_project_name)
bucket = client.bucket(storage_bucket)
if not bucket.exists():
raise DataCloudError('sync up: google cloud bucket {} '
'doesn\'t exist'.format(self.storage_bucket))
return bucket
@staticmethod
def _cmp_checksum(blob, fname):
"""
Verify local and remote checksums.
"""
md5 = file_md5(fname)[1]
b64_encoded_md5 = base64.b64encode(md5).decode() if md5 else None
if blob.md5_hash == b64_encoded_md5:
return True
return False
def _import(self, bucket_name, key, fname):
bucket = self._get_bucket_gc(bucket_name)
name = os.path.basename(fname)
tmp_file = self.tmp_file(fname)
blob = bucket.get_blob(key)
if not blob:
Logger.error('File "{}" does not exist in the cloud'.format(key))
return None
if self._cmp_checksum(blob, fname):
Logger.debug('File "{}" matches with "{}".'.format(fname, key))
return fname
Logger.debug('Downloading cache file from gc "{}/{}"'.format(bucket.name, key))
# percent_cb is not available for download_to_filename, so
# lets at least update progress at keypoints(start, finish)
progress.update_target(name, 0, None)
try:
blob.download_to_filename(tmp_file)
except Exception as exc:
Logger.error('Failed to download "{}": {}'.format(key, exc))
return None
os.rename(tmp_file, fname)
progress.finish_target(name)
Logger.debug('Downloading completed')
return fname
def push(self, path):
""" push, gcp version """
bucket = self._get_bucket_gc(self.storage_bucket)
blob_name = self.cache_file_key(path)
name = os.path.basename(path)
blob = bucket.get_blob(blob_name)
if blob is not None and blob.exists():
if self._cmp_checksum(blob, path):
Logger.debug('checksum %s matches. Skipping upload' % path)
return path
Logger.debug('checksum %s mismatch. re-uploading' % path)
# same as in _import
progress.update_target(name, 0, None)
blob = bucket.blob(blob_name)
blob.upload_from_filename(path)
progress.finish_target(name)
Logger.debug('uploading %s completed' % path)
return path
def _status(self, path):
""" status, gcp version """
bucket = self._get_bucket_gc(self.storage_bucket)
blob_name = self.cache_file_key(path)
blob = bucket.get_blob(blob_name)
remote_exists = blob is not None and blob.exists()
local_exists = os.path.exists(path)
diff = None
if remote_exists and local_exists:
diff = self._cmp_checksum(blob, path)
return (local_exists, remote_exists, diff)
def remove(self, item):
raise Exception('NOT IMPLEMENTED YET')
class DataCloud(object):
""" Generic class to do initial config parsing and redirect to proper DataCloud methods """
CLOUD_MAP = {
'AWS' : DataCloudAWS,
'GCP' : DataCloudGCP,
'HTTP' : DataCloudHTTP,
'LOCAL' : DataCloudLOCAL,
}
SCHEME_MAP = {
's3' : 'AWS',
'http' : 'HTTP',
'https' : 'HTTP',
'ftp' : 'HTTP',
'gs' : 'GCP',
'' : 'LOCAL',
}
def __init__(self, config):
self._config = config
cloud_type = self._config['Global'].get('Cloud', '').strip().upper()
if cloud_type not in self.CLOUD_MAP.keys():
raise ConfigError('Wrong cloud type %s specified' % cloud_type)
if cloud_type not in self._config.keys():
raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_type)
cloud_settings = self.get_cloud_settings(self._config,
cloud_type)
self.typ = cloud_type
self._cloud = self.CLOUD_MAP[cloud_type](cloud_settings)
self.sanity_check()
@staticmethod
def get_cloud_settings(config, cloud_type):
"""
Obtain cloud settings from config.
"""
if cloud_type not in config.keys():
cloud_config = None
else:
cloud_config = config[cloud_type]
global_storage_path = config['Global'].get('StoragePath', None)
cloud_settings = CloudSettings(global_storage_path, cloud_config)
return cloud_settings
def sanity_check(self):
""" sanity check a config
check that we have a cloud and storagePath
if aws, check can read credentials
if google, check ProjectName
Returns:
(T,) if good
(F, issues) if bad
"""
key = 'Cloud'
if key.lower() not in [k.lower() for k in self._config['Global'].keys()] or len(self._config['Global'][key]) < 1:
raise ConfigError('Please set %s in section Global in config file' % key)
# now that a cloud is chosen, can check StoragePath
storage_path = self._cloud.storage_path
if storage_path is None or len(storage_path) == 0:
raise ConfigError('Please set StoragePath = bucket/{optional path} '
'in config file in a cloud specific section')
self._cloud.sanity_check()
def _map_targets(self, func, targets, jobs):
"""
Process targets as data items in parallel.
"""
return map_progress(func, targets, jobs)
def sync(self, targets, jobs=1):
"""
Sync data items in a cloud-agnostic way.
"""
return self._map_targets(self._cloud.sync, targets, jobs)
def push(self, targets, jobs=1):
"""
Push data items in a cloud-agnostic way.
"""
return self._map_targets(self._cloud.push, targets, jobs)
def pull(self, targets, jobs=1):
"""
Pull data items in a cloud-agnostic way.
"""
return self._map_targets(self._cloud.pull, targets, jobs)
def status(self, targets, jobs=1):
"""
Check status of data items in a cloud-agnostic way.
"""
return self._map_targets(self._cloud.status, targets, jobs)
dvc/graph/__init__.py
View File
dvc/graph/commit.py
-123
View File
@@ -1,123 +0,0 @@
class Commit(object):
TEXT_LIMIT = 30
DVC_REPRO_PREFIX = 'DVC repro'
COLLAPSED_TEXT = DVC_REPRO_PREFIX + ' <<collapsed>>'
def __init__(self, hash, parents, name, date, comment,
is_target=False,
target_metric=None,
branch_tips=None):
self._hash = hash
self._parent_hashes = set(parents.split())
self._name = name
self._date = date
self._comment = comment
self._is_target = is_target
self._target_metric = target_metric
self._target_metric_delta = None
self._branch_tips = [] if branch_tips is None else branch_tips
self._collapsed_commits = []
self._is_collapsed = False
@property
def hash(self):
return self._hash
@property
def parent_hashes(self):
return self._parent_hashes
def add_parents(self, parent_hashes):
self._parent_hashes |= parent_hashes
def remove_parent(self, hash):
self._parent_hashes.remove(hash)
def add_collapsed_commit(self, commit):
self._collapsed_commits.append(commit)
self._collapsed_commits += commit._collapsed_commits
def text(self, max_commits=100):
branch_text = ''
if self._branch_tips:
branch_text = 'BRANCH TIPS: {}\n'.format(', '.join(self._branch_tips))
return self._text_metrics_line() + branch_text + self._comments_text(max_commits)
@staticmethod
def _text_comment(commit):
if commit._is_collapsed:
text = Commit.COLLAPSED_TEXT
else:
text = Commit._limit_text(commit._comment)
return '[{}] {}'.format(commit.hash, text)
@staticmethod
def _limit_text(comment):
if len(comment) < Commit.TEXT_LIMIT:
return comment
else:
return comment[:Commit.TEXT_LIMIT-3] + '...'
def _comments_text(self, max_commits=100):
res = [self._text_comment(self)]
for commit in self._collapsed_commits:
if not commit.is_repro:
text = self._text_comment(commit)
res.append(text)
was_collapsed = len(res) > max_commits
res = res[:max_commits]
max_len = max(map(lambda x: len(x), res + [""]))
res_extended_len = map(lambda x: x.ljust(max_len), res)
if was_collapsed:
res_extended_len.append('<<Collapsed commits>>')
return '\n'.join(res_extended_len)
def _text_metrics_line(self):
result = ''
if self._is_target and self._target_metric:
result = 'TARGET: {}'.format(self._target_metric)
if self._target_metric_delta is not None:
result += ' ({:+f})'.format(self._target_metric_delta)
result += '\n'
return result
@property
def is_repro(self):
return self._comment.startswith(self.DVC_REPRO_PREFIX)
def make_collapsed(self):
self._is_collapsed = True
@property
def has_target_metric(self):
return self._is_target and self._target_metric is not None
@property
def target_metric(self):
return self._target_metric
def set_target_metric(self, value):
self._target_metric = value
self._is_target = True
@property
def branch_tips(self):
return self._branch_tips
def add_branch_tips(self, tips):
for tip in tips:
self._branch_tips.append(tip)
def set_target_metric_delta(self, value):
self._target_metric_delta = value
@property
def target_metric_delta(self):
return self._target_metric_delta
dvc/graph/workflow.py
-233
View File
@@ -1,233 +0,0 @@
import networkx as nx
from dvc.exceptions import DvcException
from dvc.logger import Logger
class WorkflowError(DvcException):
def __init__(self, msg):
super(WorkflowError, self).__init__('Workflow error: ' + msg)
class CommitCollapseStrategy(object):
def __init__(self, workflow):
self._workflow = workflow
pass
def is_change_needed(self, commit, child_commits):
raise NotImplementedError()
def is_remove_needed(self, commit):
raise NotImplementedError()
def make_collapsed(self, commit):
commit.make_collapsed()
def upstream(self, commit):
for child in self._workflow.child_commits(commit.hash):
self.upstream_to_child(commit, child)
pass
def upstream_to_child(self, commit, child):
if commit.has_target_metric:
if not child.has_target_metric:
child.set_target_metric(commit.target_metric)
if commit.branch_tips:
child.add_branch_tips(commit.branch_tips)
class CollapseDvcReproCommitsStrategy(CommitCollapseStrategy):
def __init__(self, workflow):
super(CollapseDvcReproCommitsStrategy, self).__init__(workflow)
def is_change_needed(self, commit, hash):
return commit.is_repro
def is_remove_needed(self, commit):
child_commits = self._workflow.child_commits(commit.hash)
return child_commits and all(ch.is_repro for ch in child_commits)
class CollapseNotMeticsCommitsStrategy(CommitCollapseStrategy):
def __init__(self, workflow):
super(CollapseNotMeticsCommitsStrategy, self).__init__(workflow)
def is_change_needed(self, commit, hash):
return not commit.has_target_metric
def is_remove_needed(self, commit):
child_commits = self._workflow.child_commits(commit.hash)
return len(child_commits) == 1 and not commit.has_target_metric
def upstream_to_child(self, commit, child):
super(CollapseNotMeticsCommitsStrategy, self).upstream_to_child(commit, child)
if not commit.is_repro:
child.add_collapsed_commit(commit)
def make_collapsed(self, commit):
pass
class Workflow(object):
def __init__(self, target, merges_map, branches_map=None):
self._target = target
self._merges_map = merges_map
self._branches_map = branches_map
self._commits = {}
self._root_hash = None
self._edges = {}
self._back_edges = {}
pass
def add_commit(self, commit):
self._commits[commit.hash] = commit
for p in commit.parent_hashes:
if p not in self._edges:
self._edges[p] = set()
self._edges[p].add(commit.hash)
if commit.hash not in self._back_edges:
self._back_edges[commit.hash] = set()
self._back_edges[commit.hash].add(p)
if not commit.parent_hashes:
self._root_hash = commit.hash
pass
def get_commit(self, hash):
return self._commits.get(hash)
def child_commits(self, hash):
if hash not in self._edges:
return []
return [self._commits[h] for h in self._edges[hash]]
def build_graph(self, show_dvc_commits, show_all_commits, max_commits):
self.modify_workflow(show_all_commits, show_dvc_commits)
g = nx.DiGraph(name='DVC Workflow', directed=False)
for hash in set(list(self._edges.keys()) + list(self._back_edges.keys())):
commit = self._commits[hash]
g.add_node(hash,
attr_dict={
'label': commit.text(max_commits),
'color': self.node_color(commit)
}
)
for commit in self._commits.values():
for p in commit.parent_hashes:
g.add_edge(commit.hash, p)
A = nx.nx_agraph.to_agraph(g)
fname = 'workflow'
A = A.to_undirected()
A.write(fname + '.dot')
A.draw(fname + '.jpeg', format='jpeg', prog='dot')
pass
def modify_workflow(self, show_all_commits, show_dvc_commits):
self.derive_target_metric_deltas()
if not show_all_commits:
self.collapse_commits(CollapseDvcReproCommitsStrategy(self))
if not show_dvc_commits:
self.collapse_commits(CollapseNotMeticsCommitsStrategy(self))
@staticmethod
def node_color(commit):
if commit.target_metric_delta is not None:
if commit.target_metric_delta >= 0:
return 'green'
if commit.target_metric_delta < 0:
return 'red'
return 'black'
def _build_graph(self, curr, g):
if '' not in self._commits or curr.hash not in self._commits:
return
next = self._commits[curr.hash]
g.add_node(next.hash)
g.add_edge(curr.hash, next.hash)
self._build_graph(next, g)
def collapse_commits(self, strategy):
hashes_to_remove = []
for commit in self._commits.values():
if strategy.is_change_needed(commit, commit.hash):
if self._remove_or_collapse(commit, strategy):
hashes_to_remove.append(commit.hash)
self.remove_hashes(hashes_to_remove)
def remove_hashes(self, hashes_to_remove):
for hash in hashes_to_remove:
del self._commits[hash]
if hash in self._edges:
del self._edges[hash]
if hash in self._back_edges:
del self._back_edges[hash]
pass
def _remove_or_collapse(self, commit, strategy):
parent_commit_hashes = self._back_edges.get(commit.hash)
child_commit_hashes = self._edges.get(commit.hash)
if child_commit_hashes is None:
return False
if strategy.is_remove_needed(commit):
strategy.upstream(commit)
for hash in child_commit_hashes:
self._commits[hash].add_parents(commit.parent_hashes)
self._commits[hash].remove_parent(commit.hash)
self._redirect_edges(self._edges, child_commit_hashes, commit.hash, parent_commit_hashes)
self._redirect_edges(self._back_edges, parent_commit_hashes, commit.hash, child_commit_hashes)
return True
else:
strategy.make_collapsed(commit)
return False
pass
@staticmethod
def _redirect_edges(edges, child_commit_hashes, hash, commit_hashes):
if commit_hashes:
for h in commit_hashes:
edges[h].remove(hash)
if child_commit_hashes:
edges[h] |= child_commit_hashes
def derive_target_metric_deltas(self):
if not self._target:
return
if self._root_hash not in self._commits:
Logger.warn('Cannot derive target metrics deltas: root commit was not found')
return
current_target_metric = None
commit = self._commits[self._root_hash]
self._traverse_target_metric(commit, current_target_metric)
pass
def _traverse_target_metric(self, commit, current_target_metric):
new_target_metric = current_target_metric
if commit.target_metric is not None:
new_target_metric = commit.target_metric
if current_target_metric is None:
commit.set_target_metric_delta(0.0)
else:
commit.set_target_metric_delta(new_target_metric - current_target_metric)
if commit.hash in self._edges:
for hash in self._edges[commit.hash]:
self._traverse_target_metric(self._commits[hash], new_target_metric)
pass
dvc/project.py
+1 -1
View File
@@ -9,7 +9,7 @@ from dvc.state import State
from dvc.lock import Lock
from dvc.scm import SCM
from dvc.cache import Cache
from dvc.data_cloud import DataCloud
from dvc.cloud.data_cloud import DataCloud
class PipelineError(DvcException):
dvc/stage.py
+1 -1
View File
@@ -4,7 +4,7 @@ import yaml
import itertools
from dvc.system import System
from dvc.data_cloud import file_md5
from dvc.utils import file_md5
from dvc.exceptions import DvcException
from dvc.executor import Executor
dvc/utils.py
+13
View File
@@ -5,6 +5,7 @@ import os
import re
import stat
import shutil
import hashlib
from multiprocessing.pool import ThreadPool
@@ -15,6 +16,18 @@ from dvc.logger import Logger
LOCAL_CHUNK_SIZE = 1024*1024
def file_md5(fname):
""" get the (md5 hexdigest, md5 digest) of a file """
if os.path.exists(fname):
hash_md5 = hashlib.md5()
with open(fname, "rb") as fobj:
for chunk in iter(lambda: fobj.read(1024*1000), b""):
hash_md5.update(chunk)
return (hash_md5.hexdigest(), hash_md5.digest())
else:
return (None, None)
def cached_property(func):
'''A decorator for caching properties in classes.'''
def get(self):
tests/test_add.py
+1 -1
View File
@@ -3,7 +3,7 @@ import shutil
import filecmp
from dvc.main import main
from dvc.data_cloud import file_md5
from dvc.utils import file_md5
from dvc.stage import Stage, CmdOutputNoCacheError, CmdOutputOutsideOfRepoError
from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileError
from dvc.stage import CmdOutputAlreadyTrackedError
tests/test_cli.py
-20
View File
@@ -10,7 +10,6 @@ from dvc.command.remove import CmdRemove
from dvc.command.gc import CmdGC
from dvc.command.config import CmdConfig
from dvc.command.show_pipeline import CmdShowPipeline
from dvc.command.show_workflow import CmdShowWorkflow
from dvc.command.checkout import CmdCheckout
from dvc.command.fsck import CmdFsck
@@ -158,25 +157,6 @@ class TestShowPipeline(TestDvc):
self.assertEqual(args.target, [target1, target2])
class TestShowWorkflow(TestDvc):
def test(self):
target = '1'
args = parse_args(['show',
'workflow',
'-d', '--dvc-commits',
'-a', '--all-commits',
'-m', '1',
'--max-commits', '1',
target])
self.assertIsInstance(args.func(args), CmdShowWorkflow)
self.assertEqual(args.target, target)
self.assertTrue(args.dvc_commits)
self.assertTrue(args.all_commits)
self.assertEqual(args.max_commits, 1)
class TestCheckout(TestDvc):
def test(self):
args = parse_args(['checkout'])
tests/test_config.py
+1 -1
View File
@@ -16,7 +16,7 @@ except ImportError:
# Python 3
builtin_module_name = 'builtins'
from dvc.data_cloud import DataCloud
from dvc.cloud.data_cloud import DataCloud
class TestConfigTest(TestCase):
tests/test_data_cloud.py
+2 -2
View File
@@ -2,8 +2,8 @@ import os
import uuid
from dvc.config import ConfigError
from dvc.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudHTTP, DataCloudGCP
from dvc.data_cloud import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED
from dvc.cloud.data_cloud import DataCloud, DataCloudAWS, DataCloudLOCAL, DataCloudHTTP, DataCloudGCP
from dvc.cloud.base import STATUS_UNKNOWN, STATUS_OK, STATUS_MODIFIED, STATUS_NEW, STATUS_DELETED
from dvc.cloud.instance_manager import CloudSettings
from tests.basic_env import TestDvc
tests/test_gc.py
+1 -1
View File
@@ -3,7 +3,7 @@ import shutil
import filecmp
from dvc.main import main
from dvc.data_cloud import file_md5
from dvc.utils import file_md5
from dvc.stage import Stage, CmdOutputNoCacheError, CmdOutputOutsideOfRepoError
from dvc.stage import CmdOutputDoesNotExistError, CmdOutputIsNotFileError
from dvc.project import StageNotFoundError
tests/test_repro.py
+1 -1
View File
@@ -6,7 +6,7 @@ import filecmp
from dvc.main import main
from dvc.command.repro import CmdRepro
from dvc.project import ReproductionError
from dvc.data_cloud import file_md5
from dvc.utils import file_md5
from tests.basic_env import TestDvc
tests/test_run.py
+1 -1
View File
@@ -2,7 +2,7 @@ import os
import filecmp
from dvc.main import main
from dvc.data_cloud import file_md5
from dvc.utils import file_md5
from dvc.stage import Stage
from dvc.command.run import CmdRun
tests/test_workflow.py
-158
View File
@@ -1,158 +0,0 @@
from dvc.main import main
from dvc.graph.workflow import Workflow,\
CollapseDvcReproCommitsStrategy, CollapseNotMeticsCommitsStrategy
from dvc.graph.commit import Commit
from tests.basic_env import TestDvc
class TestWorkflow(TestDvc):
def setUp(self):
super(TestWorkflow, self).setUp()
self._commit4 = Commit('4', '3', 'name1', 'today', 'comment4')
self._commit3 = Commit('3', '2', 'name1', 'today', 'DVC repro-run ...')
self._commit2 = Commit('2', '1', 'name1', 'today', 'DVC repro-run ...')
self._commit1 = Commit('1', '', 'name1', 'today', 'comment1')
self._showed_text1 = '[' + self._commit1.hash + '] ' + self._commit1._comment
self._showed_text2 = '[' + self._commit2.hash + '] ' + self._commit2._comment
self._showed_text3 = '[' + self._commit3.hash + '] ' + self._commit3._comment
self._showed_text4 = '[' + self._commit4.hash + '] ' + self._commit4._comment
self._showed_text3_collapsed = '[' + self._commit3.hash + '] ' + Commit.COLLAPSED_TEXT
def commits_basic_test(self):
self.assertFalse(self._commit1.is_repro)
self.assertTrue(self._commit2.is_repro)
self.assertTrue(self._commit3.is_repro)
self.assertFalse(self._commit4.is_repro)
pass
def workflow_basic_test(self):
wf = Workflow('', '')
wf.add_commit(self._commit1)
wf.add_commit(self._commit2)
wf.add_commit(self._commit3)
wf.add_commit(self._commit4)
self.assertEqual(len(wf._commits), 4)
self.assertEqual(wf._commits['1'].text(), self._showed_text1)
self.assertEqual(wf._commits['2'].text(), self._showed_text2)
self.assertEqual(wf._commits['3'].text(), self._showed_text3)
self.assertEqual(wf._commits['4'].text(), self._showed_text4)
pass
def collapse_test(self):
wf = Workflow('', '')
wf.add_commit(self._commit1)
wf.add_commit(self._commit2)
wf.add_commit(self._commit3)
wf.add_commit(self._commit4)
wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
self.assertEqual(len(wf._commits), 3)
self.assertEqual(wf._commits[self._commit1.hash].text(), self._showed_text1)
self.assertEqual(wf._commits[self._commit3.hash].text(), self._showed_text3_collapsed)
self.assertTrue('2' not in wf._commits)
self.assertFalse('2' in wf._edges)
self.assertFalse('2' in wf._back_edges)
pass
def collapse_at_dead_end_test(self):
wf = Workflow('', '')
wf.add_commit(self._commit1)
wf.add_commit(self._commit2)
wf.add_commit(self._commit3) # Dead end which cannot be collapsed
self.assertEqual(len(wf._commits), 3)
wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
self.assertEqual(len(wf._commits), 2)
self.assertTrue('1' in wf._commits)
self.assertTrue('3' in wf._commits)
self.assertFalse('2' in wf._commits)
self.assertEqual(wf._commits[self._commit1.hash].text(), self._showed_text1)
self.assertEqual(wf._commits[self._commit3.hash].text(), self._showed_text3)
self.assertTrue('2' not in wf._commits)
pass
def collapse_metric_commit_test(self):
value = 0.812345
branches = ['master', 'try_smth']
metric_commit2 = Commit('2', '1', 'name1', 'today', 'DVC repro-run ...',
True, value, branch_tips=branches)
wf = Workflow('', '')
wf.add_commit(self._commit1)
wf.add_commit(metric_commit2)
wf.add_commit(self._commit3)
self.assertEqual(len(wf._commits), 3)
wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
self.assertEqual(len(wf._commits), 2)
self.assertEqual(wf._commits['3']._target_metric, value)
self.assertEqual(wf._commits['3'].branch_tips, branches)
pass
def collapse_all_commits_test(self):
value = 0.812345
branches = ['master', 'try_smth']
not_metric_commit2 = Commit('2', '1', 'name1', 'today', 'change smth 2...',
True, branch_tips=branches)
metric_commit3 = Commit('3', '2', 'name3', 'today3', 'hello3',
True, value)
wf = Workflow('', '')
wf.add_commit(self._commit1)
wf.add_commit(not_metric_commit2)
wf.add_commit(metric_commit3)
wf.add_commit(self._commit4)
self.assertEqual(len(wf._commits), 4)
wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
self.assertEqual(len(wf._commits), 4)
wf.collapse_commits(CollapseNotMeticsCommitsStrategy(wf))
self.assertEqual(len(wf._commits), 2)
self.assertEqual(wf._commits['3']._target_metric, value)
self.assertEqual(wf._commits['3'].branch_tips, branches)
self.assertTrue('4' in wf._commits)
pass
def collapsed_all_commits_comments_test(self):
value = 0.812345
branches = ['master', 'try_smth']
not_metric_commit2 = Commit('2', '1', 'name1', 'today', 'change smth 2...',
True, branch_tips=branches)
metric_commit3 = Commit('3', '2', 'name3', 'today3', 'hello3',
True, value)
wf = Workflow('', '')
wf.add_commit(self._commit1)
wf.add_commit(not_metric_commit2)
wf.add_commit(metric_commit3)
wf.add_commit(self._commit4)
wf.collapse_commits(CollapseDvcReproCommitsStrategy(wf))
wf.collapse_commits(CollapseNotMeticsCommitsStrategy(wf))
self.assertEqual(len(wf._commits), 2)
def is_collapsed_commit_presented(self, commit, collapsed_commit):
self.assertTrue('[' + collapsed_commit.hash + ']' in commit.text())
self.assertTrue(collapsed_commit._comment in commit.text())
is_collapsed_commit_presented(self, wf._commits['3'], metric_commit3)
is_collapsed_commit_presented(self, wf._commits['3'], not_metric_commit2)
is_collapsed_commit_presented(self, wf._commits['3'], self._commit1)
pass
def test_cli(self):
ret = main(['show',
'workflow'])
self.assertEqual(ret, 0)