From 60f1deec47ec1de3b0aea7fc1e70fdb669a30b09 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 25 Jul 2018 14:16:53 +0300 Subject: [PATCH 1/3] dvc: don't forget to pass all_branches to push/pull/fetch Signed-off-by: Ruslan Kuprieiev --- dvc/command/data_sync.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index f3606630f..4fe85cc7a 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -19,7 +19,10 @@ class CmdDataBase(CmdBase): class CmdDataPull(CmdDataBase): def do_run(self, target=None): try: - self.project.pull(target=target, jobs=self.args.jobs, remote=self.args.remote) + self.project.pull(target=target, + jobs=self.args.jobs, + remote=self.args.remote, + all_branches=self.args.all_branches) except Exception as exc: self.project.logger.error('Failed to pull data from the cloud', exc) return 1 @@ -29,7 +32,10 @@ class CmdDataPull(CmdDataBase): class CmdDataPush(CmdDataBase): def do_run(self, target=None): try: - self.project.push(target=target, jobs=self.args.jobs, remote=self.args.remote) + self.project.push(target=target, + jobs=self.args.jobs, + remote=self.args.remote, + all_branches=self.args.all_branches) except Exception as exc: self.project.logger.error('Failed to push data to the cloud', exc) return 1 @@ -39,7 +45,10 @@ class CmdDataPush(CmdDataBase): class CmdDataFetch(CmdDataBase): def do_run(self, target=None): try: - self.project.fetch(target=target, jobs=self.args.jobs, remote=self.args.remote) + self.project.fetch(target=target, + jobs=self.args.jobs, + remote=self.args.remote, + all_branches=self.args.all_branches) except Exception as exc: self.project.logger.error('Failed to fetch data from the cloud', exc) return 1 From 0e9774a3f8d49e5429ccba492c54d3a7ca0cb172 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 25 Jul 2018 17:15:12 +0300 Subject: [PATCH 2/3] RFC: follow flake8 rules Signed-off-by: Ruslan Kuprieiev --- .travis.yml | 6 +- dvc/__init__.py | 9 +- dvc/cache.py | 25 +++-- dvc/cli.py | 200 +++++++++++++++++++++++-------------- dvc/command/add.py | 3 +- dvc/command/base.py | 2 +- dvc/command/checkout.py | 1 - dvc/command/config.py | 7 +- dvc/command/data_sync.py | 6 +- dvc/command/destroy.py | 8 +- dvc/command/imp.py | 3 +- dvc/command/lock.py | 3 +- dvc/command/move.py | 5 +- dvc/command/remote.py | 11 +- dvc/command/remove.py | 3 +- dvc/command/repro.py | 5 +- dvc/command/run.py | 3 - dvc/command/status.py | 12 ++- dvc/config.py | 84 +++++++++------- dvc/data_cloud.py | 63 ++++++------ dvc/dependency/__init__.py | 13 ++- dvc/dependency/base.py | 16 +-- dvc/dependency/gs.py | 1 - dvc/dependency/local.py | 17 ++-- dvc/dependency/s3.py | 1 - dvc/dependency/ssh.py | 3 +- dvc/exceptions.py | 1 - dvc/istextfile.py | 4 +- dvc/lock.py | 3 +- dvc/output/__init__.py | 10 +- dvc/output/base.py | 9 +- dvc/output/gs.py | 14 ++- dvc/output/hdfs.py | 13 ++- dvc/output/local.py | 26 +++-- dvc/output/s3.py | 13 ++- dvc/output/ssh.py | 13 ++- dvc/progress.py | 11 +- dvc/project.py | 100 ++++++++++++------- dvc/remote/azure.py | 24 ++--- dvc/remote/base.py | 18 ++-- dvc/remote/gs.py | 72 ++++++++----- dvc/remote/hdfs.py | 55 ++++++---- dvc/remote/local.py | 65 ++++++++---- dvc/remote/s3.py | 50 ++++++---- dvc/remote/ssh.py | 76 ++++++++------ dvc/scm.py | 17 +++- dvc/stage.py | 66 +++++++----- dvc/state.py | 3 +- dvc/system.py | 18 +++- dvc/updater.py | 14 ++- dvc/utils.py | 4 - 51 files changed, 741 insertions(+), 468 deletions(-) diff --git a/.travis.yml b/.travis.yml index 906cc9d4c..c0ecf47e4 100644 --- a/.travis.yml +++ b/.travis.yml @@ -73,10 +73,8 @@ install: - bash ./scripts/ci/install.sh - pip install flake8 before_script: - # stop the build if there are Python syntax errors or undefined names - - flake8 . --count --select=E901,E999,F821,F822,F823 --show-source --statistics - # exit-zero treats all errors as warnings. The GitHub editor is 127 chars wide - - flake8 . --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics + # stop the build if there are any flake8 errors + - flake8 dvc script: - bash ./scripts/ci/script.sh deploy: diff --git a/dvc/__init__.py b/dvc/__init__.py index 3e8da84a9..6899d25d9 100644 --- a/dvc/__init__.py +++ b/dvc/__init__.py @@ -15,7 +15,8 @@ VERSIONPATH = os.path.join(PACKAGEPATH, 'version.py') if os.path.exists(os.path.join(HOMEPATH, 'setup.py')): # dvc is run directly from source without installation or # __version__ is called from setup.py - if os.getenv('APPVEYOR_REPO_TAG', '').lower() != 'true' and os.getenv('TRAVIS_TAG', '') == '': + if os.getenv('APPVEYOR_REPO_TAG', '').lower() != 'true' \ + and os.getenv('TRAVIS_TAG', '') == '': # Dynamically update version try: import git @@ -29,8 +30,8 @@ if os.path.exists(os.path.join(HOMEPATH, 'setup.py')): # and will provide a true version of the installed dvc with open(VERSIONPATH, 'w+') as fd: fd.write('# AUTOGENERATED by dvc/__init__.py\n') - fd.write('version = "{}"'.format(__version__)) - except: + fd.write('version = "{}"\n'.format(__version__)) + except Exception: pass else: # Remove version.py so that it doesn't get into the release @@ -42,7 +43,7 @@ else: try: from dvc.version import version __version__ = version - except: + except Exception: pass VERSION = __version__ diff --git a/dvc/cache.py b/dvc/cache.py index 14596d488..fdaad2db8 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -9,16 +9,20 @@ class Cache(object): def __init__(self, project): from dvc.remote import Remote + self.project = project + config = project.config._config[Config.SECTION_CACHE] local = config.get(Config.SECTION_CACHE_LOCAL, None) if local: - sect = project.config._config[Config.SECTION_REMOTE_FMT.format(local)] + name = Config.SECTION_REMOTE_FMT.format(local) + sect = project.config._config[name] else: sect = {} cache_dir = config.get(Config.SECTION_CACHE_DIR, self.CACHE_DIR) if not os.path.isabs(cache_dir): - cache_dir = os.path.abspath(os.path.realpath(os.path.join(project.dvc_dir, cache_dir))) + cache_dir = os.path.join(project.dvc_dir, cache_dir) + cache_dir = os.path.abspath(os.path.realpath(cache_dir)) sect[Config.SECTION_REMOTE_URL] = cache_dir t = config.get(Config.SECTION_CACHE_TYPE, None) if t: @@ -26,18 +30,19 @@ class Cache(object): self.local = Remote(project, sect) - self.s3 = self._get_remote(project, config, Config.SECTION_CACHE_S3) - self.gs = self._get_remote(project, config, Config.SECTION_CACHE_GS) - self.ssh = self._get_remote(project, config, Config.SECTION_CACHE_SSH) - self.hdfs = self._get_remote(project, config, Config.SECTION_CACHE_HDFS) - self.azure = self._get_remote(project, config, Config.SECTION_CACHE_AZURE) + self.s3 = self._get_remote(config, Config.SECTION_CACHE_S3) + self.gs = self._get_remote(config, Config.SECTION_CACHE_GS) + self.ssh = self._get_remote(config, Config.SECTION_CACHE_SSH) + self.hdfs = self._get_remote(config, Config.SECTION_CACHE_HDFS) + self.azure = self._get_remote(config, Config.SECTION_CACHE_AZURE) - def _get_remote(self, project, config, name): + def _get_remote(self, config, name): from dvc.remote import Remote remote = config.get(name, None) if not remote: return None - sect = project.config._config[Config.SECTION_REMOTE_FMT.format(remote)] - return Remote(project, sect) + name = Config.SECTION_REMOTE_FMT.format(remote) + sect = self.project.config._config[name] + return Remote(self.project, sect) diff --git a/dvc/cli.py b/dvc/cli.py index 3d7d53166..b025fb490 100644 --- a/dvc/cli.py +++ b/dvc/cli.py @@ -5,7 +5,6 @@ import sys import argparse from multiprocessing import cpu_count -from dvc.logger import Logger from dvc.command.init import CmdInit from dvc.command.destroy import CmdDestroy from dvc.command.remove import CmdRemove @@ -81,33 +80,35 @@ def parse_args(argv=None): init_parser = subparsers.add_parser( 'init', parents=[parent_parser], - help='Initialize dvc over a directory (should already be a git dir)') + help='Initialize dvc over a directory' + '(should already be a git dir)') init_parser.add_argument( '--no-scm', action='store_true', default=False, - help="Initiate dvc in directory that is not tracked by any scm tool(e.g. git)") - init_parser.add_argument('-f', + help="Initiate dvc in directory that is " + "not tracked by any scm tool(e.g. git)") + init_parser.add_argument( + '-f', '--force', action='store_true', default=False, help='Force initialization even if \'.dvc\' exists') init_parser.set_defaults(func=CmdInit) - # Destroy destroy_parser = subparsers.add_parser( 'destroy', parents=[parent_parser], help='Destroy dvc') - destroy_parser.add_argument('-f', + destroy_parser.add_argument( + '-f', '--force', action='store_true', default=False, help='Force destruction') destroy_parser.set_defaults(func=CmdDestroy) - # Add add_parser = subparsers.add_parser( 'add', @@ -147,39 +148,50 @@ def parse_args(argv=None): run_parser = subparsers.add_parser( 'run', parents=[parent_parser], - help='Generate a stage file from a given command and execute the command') - run_parser.add_argument('-d', + help='Generate a stage file from a given ' + 'command and execute the command') + run_parser.add_argument( + '-d', '--deps', action='append', default=[], help='Declare dependencies for reproducible cmd.') - run_parser.add_argument('-o', + run_parser.add_argument( + '-o', '--outs', action='append', default=[], help='Declare output data file or data directory.') - run_parser.add_argument('-O', + run_parser.add_argument( + '-O', '--outs-no-cache', action='append', default=[], - help='Declare output regular file or directory (sync to Git, not DVC cache).') - run_parser.add_argument('-M', + help='Declare output regular file or ' + 'directory (sync to Git, not DVC cache).') + run_parser.add_argument( + '-M', '--metrics-no-cache', action='append', default=[], - help='Declare output metric file or directory (not cached by DVC).') - run_parser.add_argument('-f', + help='Declare output metric file or ' + 'directory (not cached by DVC).') + run_parser.add_argument( + '-f', '--file', help='Specify name of the state file') - run_parser.add_argument('-c', + run_parser.add_argument( + '-c', '--cwd', default=os.path.curdir, - help='Directory to run your command and place state file in') + help='Directory to run your command and place state ' + 'file in') run_parser.add_argument( '--no-exec', action='store_true', default=False, - help="Only create stage file without actually running it") + help="Only create stage file without actually " + "running it") run_parser.add_argument( 'command', nargs=argparse.REMAINDER, @@ -190,7 +202,8 @@ def parse_args(argv=None): parent_cache_parser = argparse.ArgumentParser( add_help=False, parents=[parent_parser]) - parent_cache_parser.add_argument('-j', + parent_cache_parser.add_argument( + '-j', '--jobs', type=int, default=8*cpu_count(), @@ -206,10 +219,12 @@ def parse_args(argv=None): 'pull', parents=[parent_cache_parser], help='Pull data files from the cloud') - pull_parser.add_argument('-r', + pull_parser.add_argument( + '-r', '--remote', help='Remote repository to pull from') - pull_parser.add_argument('-a', + pull_parser.add_argument( + '-a', '--all-branches', action='store_true', default=False, @@ -221,10 +236,12 @@ def parse_args(argv=None): 'push', parents=[parent_cache_parser], help='Push data files to the cloud') - push_parser.add_argument('-r', + push_parser.add_argument( + '-r', '--remote', help='Remote repository to push to') - push_parser.add_argument('-a', + push_parser.add_argument( + '-a', '--all-branches', action='store_true', default=False, @@ -236,10 +253,12 @@ def parse_args(argv=None): 'fetch', parents=[parent_cache_parser], help='Fetch data files from the cloud') - fetch_parser.add_argument('-r', + fetch_parser.add_argument( + '-r', '--remote', help='Remote repository to fetch from') - fetch_parser.add_argument('-a', + fetch_parser.add_argument( + '-a', '--all-branches', action='store_true', default=False, @@ -251,12 +270,15 @@ def parse_args(argv=None): 'status', parents=[parent_cache_parser], help='Show the project status') - status_parser.add_argument('-c', + status_parser.add_argument( + '-c', '--cloud', action='store_true', default=False, - help='Show status of a local cache compared to a remote repository') - status_parser.add_argument('-r', + help='Show status of a local cache compared to a ' + 'remote repository') + status_parser.add_argument( + '-r', '--remote', help='Remote repository to compare local cache to') status_parser.set_defaults(func=CmdDataStatus) @@ -265,27 +287,34 @@ def parse_args(argv=None): repro_parser = subparsers.add_parser( 'repro', parents=[parent_parser], - help='Reproduce DVC file. Default file name - \'Dvcfile\'') + help='Reproduce DVC file. Default file name ' + '- \'Dvcfile\'') repro_parser.add_argument( 'targets', nargs='*', default=['Dvcfile'], help='DVC file to reproduce.') - repro_parser.add_argument('-f', + repro_parser.add_argument( + '-f', '--force', action='store_true', default=False, - help='Reproduce even if dependencies were not changed.') - repro_parser.add_argument('-s', + help='Reproduce even if dependencies were not ' + 'changed.') + repro_parser.add_argument( + '-s', '--single-item', action='store_true', default=False, - help='Reproduce only single data item without recursive dependencies check.') - repro_parser.add_argument('-c', + help='Reproduce only single data item without ' + 'recursive dependencies check.') + repro_parser.add_argument( + '-c', '--cwd', default=os.path.curdir, help='Directory to reproduce from.') - repro_parser.add_argument('-m', + repro_parser.add_argument( + '-m', '--metrics', action='store_true', default=False, @@ -298,17 +327,20 @@ def parse_args(argv=None): parents=[parent_parser], help='Remove outputs of DVC file.') remove_parser_group = remove_parser.add_mutually_exclusive_group() - remove_parser_group.add_argument('-o', + remove_parser_group.add_argument( + '-o', '--outs', action='store_true', default=True, help='Only remove DVC file outputs.') - remove_parser_group.add_argument('-p', + remove_parser_group.add_argument( + '-p', '--purge', action='store_true', default=False, help='Remove DVC file and all its outputs') - remove_parser.add_argument('targets', + remove_parser.add_argument( + 'targets', nargs='+', help='DVC files.') remove_parser.set_defaults(func=CmdRemove) @@ -331,17 +363,20 @@ def parse_args(argv=None): 'gc', parents=[parent_parser], help='Collect garbage') - gc_parser.add_argument('-a', + gc_parser.add_argument( + '-a', '--all-branches', action='store_true', default=False, help='Collect garbage for all branches.') - gc_parser.add_argument('-c', + gc_parser.add_argument( + '-c', '--cloud', action='store_true', default=False, help='Collect garbage in remote repository') - gc_parser.add_argument('-r', + gc_parser.add_argument( + '-r', '--remote', help='Remote repository to collect garbage in') gc_parser.set_defaults(func=CmdGC) @@ -351,14 +386,17 @@ def parse_args(argv=None): 'config', parents=[parent_parser], help='Get or set config options') - config_parser.add_argument('-u', + config_parser.add_argument( + '-u', '--unset', default=False, action='store_true', help='Unset option') - config_parser.add_argument('name', + config_parser.add_argument( + 'name', help='Option name') - config_parser.add_argument('value', + config_parser.add_argument( + 'value', nargs='?', default=None, help='Option value') @@ -369,7 +407,6 @@ def parse_args(argv=None): help='Use local config') config_parser.set_defaults(func=CmdConfig) - # Remote remote_parser = subparsers.add_parser( 'remote', @@ -378,7 +415,8 @@ def parse_args(argv=None): remote_subparsers = remote_parser.add_subparsers( dest='cmd', - help='Use dvc remote CMD --help for command-specific help') + help='Use dvc remote CMD --help for ' + 'command-specific help') _fix_subparsers(remote_subparsers) @@ -397,14 +435,14 @@ def parse_args(argv=None): action='store_true', default=False, help='Use local config') - remote_add_parser.add_argument('-d', + remote_add_parser.add_argument( + '-d', '--default', action='store_true', default=False, help='Set as default remote') remote_add_parser.set_defaults(func=CmdRemoteAdd) - remote_remove_parser = remote_subparsers.add_parser( 'remove', parents=[parent_parser], @@ -419,7 +457,6 @@ def parse_args(argv=None): help='Use local config') remote_remove_parser.set_defaults(func=CmdRemoteRemove) - remote_modify_parser = remote_subparsers.add_parser( 'modify', parents=[parent_parser], @@ -434,7 +471,8 @@ def parse_args(argv=None): 'value', nargs='?', help='Value') - remote_modify_parser.add_argument('-u', + remote_modify_parser.add_argument( + '-u', '--unset', default=False, action='store_true', @@ -446,7 +484,6 @@ def parse_args(argv=None): help='Use local config') remote_modify_parser.set_defaults(func=CmdRemoteModify) - remote_list_parser = remote_subparsers.add_parser( 'list', parents=[parent_parser], @@ -466,11 +503,11 @@ def parse_args(argv=None): metrics_subparsers = metrics_parser.add_subparsers( dest='cmd', - help='Use dvc metrics CMD --help for command-specific help') + help='Use dvc metrics CMD --help for ' + 'command-specific help') _fix_subparsers(metrics_subparsers) - metrics_show_parser = metrics_subparsers.add_parser( 'show', parents=[parent_parser], @@ -479,44 +516,48 @@ def parse_args(argv=None): 'path', nargs='?', help='Path to metrics file') - metrics_show_parser.add_argument('-t', + metrics_show_parser.add_argument( + '-t', '--type', help='Type of metrics(RAW/JSON/TSV/HTSV/CSV/HCSV)') - metrics_show_parser.add_argument('-x', + metrics_show_parser.add_argument( + '-x', '--xpath', help='JSON/TSV/HTSV/CSV/HCSV path') - metrics_show_parser_group = metrics_show_parser.add_mutually_exclusive_group() - metrics_show_parser_group.add_argument( + metrics_show_group = metrics_show_parser.add_mutually_exclusive_group() + metrics_show_group.add_argument( '--json-path', help='JSON path') - metrics_show_parser_group.add_argument( + metrics_show_group.add_argument( '--tsv-path', help='TSV path \'row,column\'(e.g. \'1,2\')') - metrics_show_parser_group.add_argument( + metrics_show_group.add_argument( '--htsv-path', help='Headed TSV path \'row,column\'(e.g. \'Name,3\'') - metrics_show_parser_group.add_argument( + metrics_show_group.add_argument( '--csv-path', help='CSV path \'row,column\'(e.g. \'1,2\')') - metrics_show_parser_group.add_argument( + metrics_show_group.add_argument( '--hcsv-path', help='Headed CSV path \'row,column\'(e.g. \'Name,3\'') - metrics_show_parser.add_argument('-a', + metrics_show_parser.add_argument( + '-a', '--all-branches', action='store_true', default=False, help='Show metrics for all branches') metrics_show_parser.set_defaults(func=CmdMetricsShow) - metrics_add_parser = metrics_subparsers.add_parser( 'add', parents=[parent_parser], help='Add metrics') - metrics_add_parser.add_argument('-t', + metrics_add_parser.add_argument( + '-t', '--type', help='Type of metrics(RAW/JSON/TSV/HTSV/CSV/HCSV)') - metrics_add_parser.add_argument('-x', + metrics_add_parser.add_argument( + '-x', '--xpath', help='JSON/TSV/HTSV/CSV/HCSV path') metrics_add_parser.add_argument( @@ -524,15 +565,16 @@ def parse_args(argv=None): help='Path to metrics file') metrics_add_parser.set_defaults(func=CmdMetricsAdd) - metrics_modify_parser = metrics_subparsers.add_parser( 'modify', parents=[parent_parser], help='Modify metrics') - metrics_modify_parser.add_argument('-t', + metrics_modify_parser.add_argument( + '-t', '--type', help='Type of metrics(RAW/JSON/TSV/HTSV/CSV/HCSV)') - metrics_modify_parser.add_argument('-x', + metrics_modify_parser.add_argument( + '-x', '--xpath', help='JSON/TSV/HTSV/CSV/HCSV path') metrics_modify_parser.add_argument( @@ -540,7 +582,6 @@ def parse_args(argv=None): help='Metrics file') metrics_modify_parser.set_defaults(func=CmdMetricsModify) - metrics_remove_parser = metrics_subparsers.add_parser( 'remove', parents=[parent_parser], @@ -569,7 +610,8 @@ def parse_args(argv=None): 'lock', parents=[parent_parser], help='Lock DVC file') - lock_parser.add_argument('targets', + lock_parser.add_argument( + 'targets', nargs='+', help='DVC files.') lock_parser.set_defaults(func=CmdLock) @@ -579,7 +621,8 @@ def parse_args(argv=None): 'unlock', parents=[parent_parser], help='Unlock DVC file') - unlock_parser.add_argument('targets', + unlock_parser.add_argument( + 'targets', nargs='+', help='DVC files.') unlock_parser.set_defaults(func=CmdUnlock) @@ -592,7 +635,8 @@ def parse_args(argv=None): pipeline_subparsers = pipeline_parser.add_subparsers( dest='cmd', - help='Use dvc pipeline CMD --help for command-specific help') + help='Use dvc pipeline CMD --help ' + 'for command-specific help') _fix_subparsers(pipeline_subparsers) @@ -600,23 +644,25 @@ def parse_args(argv=None): 'show', parents=[parent_parser], help='Show pipeline') - pipeline_show_parser_group = pipeline_show_parser.add_mutually_exclusive_group() - pipeline_show_parser_group.add_argument('-c', + pipeline_show_group = pipeline_show_parser.add_mutually_exclusive_group() + pipeline_show_group.add_argument( + '-c', '--commands', action='store_true', default=False, help='Print commands instead of paths to DVC files.') - pipeline_show_parser_group.add_argument('-o', + pipeline_show_group.add_argument( + '-o', '--outs', action='store_true', default=False, - help='Print output files instead of paths to DVC files.') + help='Print output files instead of' + 'paths to DVC files.') pipeline_show_parser.add_argument( 'targets', nargs='+', help='DVC files.') pipeline_show_parser.set_defaults(func=CmdPipelineShow) - args = parser.parse_args(argv) diff --git a/dvc/command/add.py b/dvc/command/add.py index 813a497eb..d21bc1493 100644 --- a/dvc/command/add.py +++ b/dvc/command/add.py @@ -1,5 +1,6 @@ from dvc.exceptions import DvcException from dvc.command.base import CmdBase +from dvc.logger import Logger class CmdAdd(CmdBase): @@ -8,6 +9,6 @@ class CmdAdd(CmdBase): try: self.project.add(target) except DvcException as ex: - self.project.logger.error('Failed to add \'{}\''.format(target), ex) + Logger.error('Failed to add \'{}\''.format(target), ex) return 1 return 0 diff --git a/dvc/command/base.py b/dvc/command/base.py index 59b201a34..571f99a55 100644 --- a/dvc/command/base.py +++ b/dvc/command/base.py @@ -39,7 +39,7 @@ class CmdBase(object): with self.project.lock: return self.run() except LockError as ex: - self.project.logger.error('Failed to lock before running a command', ex) + Logger.error('Failed to lock before running a command', ex) return 1 # Abstract methods that have to be implemented by any inheritance class diff --git a/dvc/command/checkout.py b/dvc/command/checkout.py index 50e519f9a..54ce65932 100644 --- a/dvc/command/checkout.py +++ b/dvc/command/checkout.py @@ -1,4 +1,3 @@ -import os from dvc.command.base import CmdBase diff --git a/dvc/command/config.py b/dvc/command/config.py index 731ba5c8a..bd7c5b204 100644 --- a/dvc/command/config.py +++ b/dvc/command/config.py @@ -39,7 +39,8 @@ class CmdConfig(CmdBase): try: self.configobj.write() except Exception as exc: - Logger.error("Failed to write config '{}'".format(self.configobj.filename), exc) + msg = "Failed to write config '{}'".format(self.configobj.filename) + Logger.error(msg, exc) return 1 return 0 @@ -51,7 +52,7 @@ class CmdConfig(CmdBase): if opt in self.configobj[section].keys(): del self.configobj[section][opt] - if len(self.configobj[section]) == 0 or opt == None: + if len(self.configobj[section]) == 0 or opt is None: del self.configobj[section] return self.save() @@ -70,7 +71,7 @@ class CmdConfig(CmdBase): def check_opt(self): _section, _opt = self.args.name.strip().split('.', 1) - add = (self.args.value != None and self.args.unset == False) + add = (self.args.value is not None and self.args.unset is False) section = self._get_key(self.configobj, _section, add) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index 4fe85cc7a..5c0b2fac6 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -24,7 +24,8 @@ class CmdDataPull(CmdDataBase): remote=self.args.remote, all_branches=self.args.all_branches) except Exception as exc: - self.project.logger.error('Failed to pull data from the cloud', exc) + self.project.logger.error('Failed to pull data from the cloud', + exc) return 1 return 0 @@ -50,6 +51,7 @@ class CmdDataFetch(CmdDataBase): remote=self.args.remote, all_branches=self.args.all_branches) except Exception as exc: - self.project.logger.error('Failed to fetch data from the cloud', exc) + self.project.logger.error('Failed to fetch data from the cloud', + exc) return 1 return 0 diff --git a/dvc/command/destroy.py b/dvc/command/destroy.py index 6b96aacd1..97e66de7e 100644 --- a/dvc/command/destroy.py +++ b/dvc/command/destroy.py @@ -6,13 +6,13 @@ from dvc.exceptions import DvcException class CmdDestroy(CmdBase): def run_cmd(self): try: - msg = u'This will destroy all information about your pipelines as ' \ - u'well as cache in .dvc/cache.\n' \ + msg = u'This will destroy all information about your pipelines ' \ + u'as well as cache in .dvc/cache.\n' \ u'Are you sure you want to continue?' if not self.args.force and not prompt(msg, False): - msg = u'Cannot destroy without a confirmation from the user. ' \ - u'Use \'-f\' to force.' + msg = u'Cannot destroy without a confirmation from the ' \ + u'user. Use \'-f\' to force.' raise DvcException(msg) self.project.destroy() diff --git a/dvc/command/imp.py b/dvc/command/imp.py index 8ed3c0950..9f13a730b 100644 --- a/dvc/command/imp.py +++ b/dvc/command/imp.py @@ -1,5 +1,6 @@ from dvc.exceptions import DvcException from dvc.command.base import CmdBase +from dvc.logger import Logger class CmdImport(CmdBase): @@ -7,6 +8,6 @@ class CmdImport(CmdBase): try: self.project.imp(self.args.url, self.args.out) except DvcException as ex: - self.project.logger.error('Failed to import {}'.format(self.args.url), ex) + Logger.error('Failed to import {}'.format(self.args.url), ex) return 1 return 0 diff --git a/dvc/command/lock.py b/dvc/command/lock.py index 24a1c42cf..12c0f82eb 100644 --- a/dvc/command/lock.py +++ b/dvc/command/lock.py @@ -8,7 +8,8 @@ class CmdLockBase(CmdBase): try: self.project.lock_stage(target, unlock=unlock) except DvcException as ex: - msg = 'Failed to {}lock \'{}\''.format('un' if unlock else '', target) + msg = 'Failed to {}lock \'{}\'' + msg.format('un' if unlock else '', target) self.project.logger.error(msg, ex) return 1 return 0 diff --git a/dvc/command/move.py b/dvc/command/move.py index 7f2ac6fb7..87c6a8f2c 100644 --- a/dvc/command/move.py +++ b/dvc/command/move.py @@ -7,7 +7,8 @@ class CmdMove(CmdBase): try: self.project.move(self.args.src, self.args.dst) except DvcException as ex: - self.project.logger.error('Failed to move \'{}\' -> \'{}\''.format(self.args.src, - self.args.dst), ex) + msg = 'Failed to move \'{}\' -> \'{}\''.format(self.args.src, + self.args.dst) + self.project.logger.error(msg, ex) return 1 return 0 diff --git a/dvc/command/remote.py b/dvc/command/remote.py index cca6ddb90..ecda2e4f4 100644 --- a/dvc/command/remote.py +++ b/dvc/command/remote.py @@ -13,11 +13,15 @@ class CmdRemoteAdd(CmdConfig): return ret if self.args.default: - Logger.info('Setting \'{}\' as a default remote.'.format(self.args.name)) - ret = self.set(Config.SECTION_CORE, Config.SECTION_CORE_REMOTE, self.args.name) + msg = 'Setting \'{}\' as a default remote.'.format(self.args.name) + Logger.info(msg) + ret = self.set(Config.SECTION_CORE, + Config.SECTION_CORE_REMOTE, + self.args.name) return ret + class CmdRemoteRemove(CmdConfig): def run(self): section = Config.SECTION_REMOTE_FMT.format(self.args.name) @@ -37,6 +41,7 @@ class CmdRemoteList(CmdConfig): r = re.match(Config.SECTION_REMOTE_REGEX, section) if r: name = r.group('name') - url = self.configobj[section].get(Config.SECTION_REMOTE_URL, '') + url = self.configobj[section].get(Config.SECTION_REMOTE_URL, + '') Logger.info('{}\t{}'.format(name, url)) return 0 diff --git a/dvc/command/remove.py b/dvc/command/remove.py index 3fa0260c2..f37d1bb8c 100644 --- a/dvc/command/remove.py +++ b/dvc/command/remove.py @@ -1,5 +1,6 @@ from dvc.exceptions import DvcException from dvc.command.base import CmdBase +from dvc.logger import Logger class CmdRemove(CmdBase): @@ -9,6 +10,6 @@ class CmdRemove(CmdBase): try: self.project.remove(target, outs_only=outs_only) except DvcException as ex: - self.project.logger.error('Failed to remove {}'.format(target), ex) + Logger.error('Failed to remove {}'.format(target), ex) return 1 return 0 diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 433702ff8..aed5390f3 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -6,7 +6,6 @@ from dvc.command.base import CmdBase class CmdRepro(CmdBase): def run(self): from dvc.project import ReproductionError - from dvc.stage import Stage recursive = not self.args.single_item saved_dir = os.path.realpath(os.curdir) @@ -27,8 +26,8 @@ class CmdRepro(CmdBase): ret = 1 break except Exception as ex: - msg = 'Failed to reproduce \'{}\' - unexpected error'.format(target) - self.project.logger.error(msg, ex) + msg = 'Failed to reproduce \'{}\' - unexpected error' + self.project.logger.error(msg.format(target), ex) ret = 1 break diff --git a/dvc/command/run.py b/dvc/command/run.py index b279e47ce..e9e9f29a5 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -1,7 +1,4 @@ -import os - from dvc.command.base import CmdBase -from dvc.logger import Logger from dvc.exceptions import DvcException diff --git a/dvc/command/status.py b/dvc/command/status.py index cce4c3f57..63b92e6a4 100644 --- a/dvc/command/status.py +++ b/dvc/command/status.py @@ -1,5 +1,3 @@ -import os - from dvc.command.data_sync import CmdDataBase @@ -8,7 +6,7 @@ class CmdDataStatus(CmdDataBase): STATUS_INDENT = '\t' def _normalize(self, s): - s+=':' + s += ':' assert len(s) < self.STATUS_LEN return s + (self.STATUS_LEN - len(s))*' ' @@ -20,12 +18,16 @@ class CmdDataStatus(CmdDataBase): self.project.logger.info('{}{}'.format(ind, key)) self._show(value, indent+1) else: - self.project.logger.info('{}{}{}'.format(ind, self._normalize(value), key)) + msg = '{}{}{}'.format(ind, self._normalize(value), key) + self.project.logger.info(msg) def do_run(self, target=None): indent = 1 if self.args.cloud else 0 try: - status = self.project.status(target=target, jobs=self.args.jobs, cloud=self.args.cloud, remote=self.args.remote) + status = self.project.status(target=target, + jobs=self.args.jobs, + cloud=self.args.cloud, + remote=self.args.remote) self._show(status, indent) except Exception as exc: self.project.logger.error('Failed to obtain data status', exc) diff --git a/dvc/config.py b/dvc/config.py index 1bc915605..cac6c3b8f 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -2,8 +2,8 @@ DVC config objects. """ import os -import schema import configobj +from schema import Schema, Optional, And, Use, Regex from dvc.exceptions import DvcException @@ -28,13 +28,21 @@ def supported_cache_type(types): return True +def supported_loglevel(level): + return level in ['info', 'debug', 'warning', 'error'] + + +def supported_cloud(cloud): + return cloud in ['aws', 'gcp', 'local', ''] + + class Config(object): CONFIG = 'config' CONFIG_LOCAL = 'config.local' SECTION_CORE = 'core' SECTION_CORE_LOGLEVEL = 'loglevel' - SECTION_CORE_LOGLEVEL_SCHEMA = schema.And(schema.Use(str.lower), lambda l: l in ('info', 'debug', 'warning', 'error')) + SECTION_CORE_LOGLEVEL_SCHEMA = And(Use(str.lower), supported_loglevel) SECTION_CORE_REMOTE = 'remote' SECTION_CACHE = 'cache' @@ -48,30 +56,32 @@ class Config(object): SECTION_CACHE_HDFS = 'hdfs' SECTION_CACHE_AZURE = 'azure' SECTION_CACHE_SCHEMA = { - schema.Optional(SECTION_CACHE_LOCAL): str, - schema.Optional(SECTION_CACHE_S3): str, - schema.Optional(SECTION_CACHE_GS): str, - schema.Optional(SECTION_CACHE_HDFS): str, - schema.Optional(SECTION_CACHE_SSH): str, - schema.Optional(SECTION_CACHE_AZURE): str, + Optional(SECTION_CACHE_LOCAL): str, + Optional(SECTION_CACHE_S3): str, + Optional(SECTION_CACHE_GS): str, + Optional(SECTION_CACHE_HDFS): str, + Optional(SECTION_CACHE_SSH): str, + Optional(SECTION_CACHE_AZURE): str, # backward compatibility - schema.Optional(SECTION_CACHE_DIR, default='cache'): str, - schema.Optional(SECTION_CACHE_TYPE, default=None): SECTION_CACHE_TYPE_SCHEMA, + Optional(SECTION_CACHE_DIR, default='cache'): str, + Optional(SECTION_CACHE_TYPE, default=None): SECTION_CACHE_TYPE_SCHEMA, } # backward compatibility SECTION_CORE_CLOUD = 'cloud' - SECTION_CORE_CLOUD_SCHEMA = schema.And(schema.Use(str.lower), lambda c: c in ('aws', 'gcp', 'local', '')) + SECTION_CORE_CLOUD_SCHEMA = And(Use(str.lower), supported_cloud) SECTION_CORE_STORAGEPATH = 'storagepath' SECTION_CORE_SCHEMA = { - schema.Optional(SECTION_CORE_LOGLEVEL, default='info'): schema.And(str, schema.Use(str.lower), SECTION_CORE_LOGLEVEL_SCHEMA), - schema.Optional(SECTION_CORE_REMOTE, default=''): schema.And(str, schema.Use(str.lower)), + Optional(SECTION_CORE_LOGLEVEL, + default='info'): And(str, Use(str.lower), + SECTION_CORE_LOGLEVEL_SCHEMA), + Optional(SECTION_CORE_REMOTE, default=''): And(str, Use(str.lower)), # backward compatibility - schema.Optional(SECTION_CORE_CLOUD, default=''): SECTION_CORE_CLOUD_SCHEMA, - schema.Optional(SECTION_CORE_STORAGEPATH, default=''): str, + Optional(SECTION_CORE_CLOUD, default=''): SECTION_CORE_CLOUD_SCHEMA, + Optional(SECTION_CORE_STORAGEPATH, default=''): str, } # backward compatibility @@ -83,10 +93,10 @@ class Config(object): SECTION_AWS_PROFILE = 'profile' SECTION_AWS_SCHEMA = { SECTION_AWS_STORAGEPATH: str, - schema.Optional(SECTION_AWS_REGION): str, - schema.Optional(SECTION_AWS_PROFILE, default='default'): str, - schema.Optional(SECTION_AWS_CREDENTIALPATH, default = ''): str, - schema.Optional(SECTION_AWS_ENDPOINT_URL, default=None): str, + Optional(SECTION_AWS_REGION): str, + Optional(SECTION_AWS_PROFILE, default='default'): str, + Optional(SECTION_AWS_CREDENTIALPATH, default=''): str, + Optional(SECTION_AWS_ENDPOINT_URL, default=None): str, } # backward compatibility @@ -95,7 +105,7 @@ class Config(object): SECTION_GCP_PROJECTNAME = 'projectname' SECTION_GCP_SCHEMA = { SECTION_GCP_STORAGEPATH: str, - schema.Optional(SECTION_GCP_PROJECTNAME): str, + Optional(SECTION_GCP_PROJECTNAME): str, } # backward compatibility @@ -105,31 +115,30 @@ class Config(object): SECTION_LOCAL_STORAGEPATH: str, } - SECTION_REMOTE_REGEX = r'^\s*remote\s*"(?P.*)"\s*$' SECTION_REMOTE_FMT = 'remote "{}"' SECTION_REMOTE_URL = 'url' SECTION_REMOTE_USER = 'user' SECTION_REMOTE_SCHEMA = { SECTION_REMOTE_URL: supported_url, - schema.Optional(SECTION_AWS_REGION): str, - schema.Optional(SECTION_AWS_PROFILE, default='default'): str, - schema.Optional(SECTION_AWS_CREDENTIALPATH, default = ''): str, - schema.Optional(SECTION_AWS_ENDPOINT_URL, default=None): str, - schema.Optional(SECTION_GCP_PROJECTNAME): str, - schema.Optional(SECTION_CACHE_TYPE): SECTION_CACHE_TYPE_SCHEMA, - schema.Optional(SECTION_REMOTE_USER): str, + Optional(SECTION_AWS_REGION): str, + Optional(SECTION_AWS_PROFILE, default='default'): str, + Optional(SECTION_AWS_CREDENTIALPATH, default=''): str, + Optional(SECTION_AWS_ENDPOINT_URL, default=None): str, + Optional(SECTION_GCP_PROJECTNAME): str, + Optional(SECTION_CACHE_TYPE): SECTION_CACHE_TYPE_SCHEMA, + Optional(SECTION_REMOTE_USER): str, } SCHEMA = { - schema.Optional(SECTION_CORE, default={}): SECTION_CORE_SCHEMA, - schema.Optional(schema.Regex(SECTION_REMOTE_REGEX)): SECTION_REMOTE_SCHEMA, - schema.Optional(SECTION_CACHE, default={}): SECTION_CACHE_SCHEMA, + Optional(SECTION_CORE, default={}): SECTION_CORE_SCHEMA, + Optional(Regex(SECTION_REMOTE_REGEX)): SECTION_REMOTE_SCHEMA, + Optional(SECTION_CACHE, default={}): SECTION_CACHE_SCHEMA, # backward compatibility - schema.Optional(SECTION_AWS, default={}): SECTION_AWS_SCHEMA, - schema.Optional(SECTION_GCP, default={}): SECTION_GCP_SCHEMA, - schema.Optional(SECTION_LOCAL, default={}): SECTION_LOCAL_SCHEMA, + Optional(SECTION_AWS, default={}): SECTION_AWS_SCHEMA, + Optional(SECTION_GCP, default={}): SECTION_GCP_SCHEMA, + Optional(SECTION_LOCAL, default={}): SECTION_LOCAL_SCHEMA, } def __init__(self, dvc_dir): @@ -142,15 +151,16 @@ class Config(object): local = configobj.ConfigObj(self.config_local_file) # NOTE: schema doesn't support ConfigObj.Section validation, so we - # need to convert our config to dict before passing it to schema. + # need to convert our config to dict before passing it to self._config = self._lower(self._config) local = self._lower(local) self._config.update(local) - self._config = schema.Schema(self.SCHEMA).validate(self._config) + self._config = Schema(self.SCHEMA).validate(self._config) # NOTE: now converting back to ConfigObj - self._config = configobj.ConfigObj(self._config, write_empty_values=True) + self._config = configobj.ConfigObj(self._config, + write_empty_values=True) self._config.filename = self.config_file except Exception as ex: raise ConfigError(ex) diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index 4a163fb68..1f296e5cc 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -1,8 +1,4 @@ -import re -from multiprocessing.pool import ThreadPool - from dvc.logger import Logger -from dvc.exceptions import DvcException from dvc.config import Config, ConfigError from dvc.remote.s3 import RemoteS3 @@ -11,30 +7,29 @@ from dvc.remote.azure import RemoteAzure from dvc.remote.ssh import RemoteSSH from dvc.remote.hdfs import RemoteHDFS from dvc.remote.local import RemoteLOCAL -from dvc.remote.base import RemoteBase class DataCloud(object): - """ Generic class to do initial config parsing and redirect to proper DataCloud methods """ - CLOUD_MAP = { - 'aws' : RemoteS3, - 'gcp' : RemoteGS, - 'azure' : RemoteAzure, - 'ssh' : RemoteSSH, - 'hdfs' : RemoteHDFS, - 'local' : RemoteLOCAL, + 'aws': RemoteS3, + 'gcp': RemoteGS, + 'azure': RemoteAzure, + 'ssh': RemoteSSH, + 'hdfs': RemoteHDFS, + 'local': RemoteLOCAL, } def __init__(self, project, config=None): self.project = project self._config = config + self._core = self._config[Config.SECTION_CORE] - remote = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_REMOTE, '') + remote = self._core.get(Config.SECTION_CORE_REMOTE, '') if remote == '': - if config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, None): + if self._core.get(Config.SECTION_CORE_CLOUD, None): # backward compatibility - Logger.warn('Using obsoleted config format. Consider updating.') + msg = 'Using obsoleted config format. Consider updating.' + Logger.warn(msg) self._cloud = self.__init__compat() else: self._cloud = None @@ -53,7 +48,8 @@ class DataCloud(object): section = Config.SECTION_REMOTE_FMT.format(remote) cloud_config = self._config.get(section, None) if not cloud_config: - raise ConfigError("Can't find remote section '{}' in config".format(section)) + msg = "Can't find remote section '{}' in config" + raise ConfigError(msg.format(section)) cloud_type = self.supported(cloud_config) if not cloud_type: @@ -62,23 +58,25 @@ class DataCloud(object): return self._init_cloud(cloud_config, cloud_type) def __init__compat(self): - cloud_name = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_CLOUD, '').strip().lower() - if cloud_name == '': + name = self._core.get(Config.SECTION_CORE_CLOUD, '').strip().lower() + if name == '': self._cloud = None return - cloud_type = self.CLOUD_MAP.get(cloud_name, None) + cloud_type = self.CLOUD_MAP.get(name, None) if not cloud_type: - raise ConfigError('Wrong cloud type %s specified' % cloud_name) + msg = "Wrong cloud type {} specified" + raise ConfigError(msg.format(name)) - cloud_config = self._config.get(cloud_name, None) + cloud_config = self._config.get(name, None) if not cloud_config: - raise ConfigError('Can\'t find cloud section \'[%s]\' in config' % cloud_name) + msg = "Can't find cloud section '{}' in config" + raise ConfigError(msg.format(name)) return self._init_cloud(cloud_config, cloud_type) def _init_cloud(self, cloud_config, cloud_type): - global_storage_path = self._config[Config.SECTION_CORE].get(Config.SECTION_CORE_STORAGEPATH, None) + global_storage_path = self._core.get(Config.SECTION_CORE_STORAGEPATH) if global_storage_path: Logger.warn('Using obsoleted config format. Consider updating.') @@ -92,23 +90,30 @@ class DataCloud(object): if self._cloud: return self._cloud - raise ConfigError("No remote repository specified. Setup default repository " \ - "with 'dvc config core.remote ' or use '-r '.") + msg = "No remote repository specified. Setup default repository " \ + "with 'dvc config core.remote ' or use '-r '." + raise ConfigError(msg) def push(self, targets, jobs=1, remote=None): """ Push data items in a cloud-agnostic way. """ - return self.project.cache.local.push(targets, jobs=jobs, remote=self._get_cloud(remote)) + return self.project.cache.local.push(targets, + jobs=jobs, + remote=self._get_cloud(remote)) def pull(self, targets, jobs=1, remote=None): """ Pull data items in a cloud-agnostic way. """ - return self.project.cache.local.pull(targets, jobs=jobs, remote=self._get_cloud(remote)) + return self.project.cache.local.pull(targets, + jobs=jobs, + remote=self._get_cloud(remote)) def status(self, targets, jobs=1, remote=None): """ Check status of data items in a cloud-agnostic way. """ - return self.project.cache.local.status(targets, jobs=jobs, remote=self._get_cloud(remote)) + return self.project.cache.local.status(targets, + jobs=jobs, + remote=self._get_cloud(remote)) diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 73490fecd..4c0ae0a53 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -18,17 +18,19 @@ from dvc.dependency.ssh import DependencySSH from dvc.remote import Remote from dvc.remote.local import RemoteLOCAL from dvc.remote.s3 import RemoteS3 -from dvc.remote.gs import RemoteGS -from dvc.remote.ssh import RemoteSSH from dvc.remote.hdfs import RemoteHDFS -DEPS = [DependencyHDFS, DependencyS3, DependencyGS, DependencySSH, DependencyLOCAL] +DEPS = [DependencyHDFS, + DependencyS3, + DependencyGS, + DependencySSH, + DependencyLOCAL] DEP_MAP = {'': DependencyLOCAL, 'ssh': DependencySSH, 's3': DependencyS3, 'gs': DependencyGS, - 'hdfs': DependencyHDFS,} + 'hdfs': DependencyHDFS} SCHEMA = { DependencyBase.PARAM_PATH: str, @@ -41,7 +43,8 @@ SCHEMA = { def _get(stage, p, info): parsed = urlparse(p) if parsed.scheme == 'remote': - sect = stage.project.config._config[Config.SECTION_REMOTE_FMT.format(parsed.netloc)] + name = Config.SECTION_REMOTE_FMT.format(parsed.netloc) + sect = stage.project.config._config[name] remote = Remote(stage.project, sect) return DEP_MAP[remote.scheme](stage, p, info, remote=remote) diff --git a/dvc/dependency/base.py b/dvc/dependency/base.py index 9fb7c43b6..96f37a60f 100644 --- a/dvc/dependency/base.py +++ b/dvc/dependency/base.py @@ -1,24 +1,24 @@ import re -import schema -import posixpath -import ntpath from dvc.exceptions import DvcException class DependencyError(DvcException): def __init__(self, path, msg): - super(DependencyError, self).__init__('Dependency \'{}\' error: {}'.format(path, msg)) + msg = 'Dependency \'{}\' error: {}' + super(DependencyError, self).__init__(msg.format(path, msg)) class DependencyDoesNotExistError(DependencyError): def __init__(self, path): - super(DependencyDoesNotExistError, self).__init__(path, 'does not exist') + msg = 'does not exist' + super(DependencyDoesNotExistError, self).__init__(path, msg) class DependencyIsNotFileOrDirError(DependencyError): def __init__(self, path): - super(DependencyIsNotFileOrDirError, self).__init__(path, 'not a file or directory') + msg = 'not a file or directory' + super(DependencyIsNotFileOrDirError, self).__init__(path, msg) class DependencyBase(object): @@ -43,7 +43,7 @@ class DependencyBase(object): @classmethod def supported(cls, url): - return cls.match(url) != None + return cls.match(url) is not None @property def sep(self): @@ -58,7 +58,7 @@ class DependencyBase(object): def status(self): if self.changed(): - #FIXME better msgs + # FIXME better msgs return {self.rel_path: 'changed'} return {} diff --git a/dvc/dependency/gs.py b/dvc/dependency/gs.py index d65068a2e..2c0be604e 100644 --- a/dvc/dependency/gs.py +++ b/dvc/dependency/gs.py @@ -7,7 +7,6 @@ except ImportError: from dvc.dependency.s3 import DependencyS3 from dvc.remote.gs import RemoteGS -from dvc.config import Config class DependencyGS(DependencyS3): diff --git a/dvc/dependency/local.py b/dvc/dependency/local.py index df1294fad..1d9a9a0c5 100644 --- a/dvc/dependency/local.py +++ b/dvc/dependency/local.py @@ -1,15 +1,14 @@ import os -import schema try: from urlparse import urlparse except ImportError: from urllib.parse import urlparse -from dvc.dependency.base import * +from dvc.dependency.base import DependencyBase +from dvc.dependency.base import DependencyDoesNotExistError +from dvc.dependency.base import DependencyIsNotFileOrDirError from dvc.logger import Logger -from dvc.utils import remove -from dvc.config import Config from dvc.remote.local import RemoteLOCAL @@ -23,7 +22,10 @@ class DependencyLOCAL(DependencyBase): self.stage = stage self.project = stage.project self.info = info - self.remote = remote if remote != None else RemoteLOCAL(stage.project, {}) + if remote is not None: + self.remote = remote + else: + self.remote = RemoteLOCAL(stage.project, {}) if remote: path = os.path.join(remote.prefix, urlparse(path).path.lstrip('/')) @@ -61,13 +63,14 @@ class DependencyLOCAL(DependencyBase): if (os.path.isfile(self.path) and os.path.getsize(self.path) == 0) or \ (os.path.isdir(self.path) and len(os.listdir(self.path)) == 0): - self.project.logger.warn("File/directory '{}' is empty.".format(self.rel_path)) + Logger.warn("File/directory '{}' is empty.".format(self.rel_path)) self.info = self.remote.save_info(self.path_info) def dumpd(self): if self.path.startswith(self.stage.project.root_dir): - path = self.remote.unixpath(os.path.relpath(self.path, self.stage.cwd)) + path = self.remote.unixpath(os.path.relpath(self.path, + self.stage.cwd)) else: path = self.path diff --git a/dvc/dependency/s3.py b/dvc/dependency/s3.py index e75c20515..f8dd499f1 100644 --- a/dvc/dependency/s3.py +++ b/dvc/dependency/s3.py @@ -7,7 +7,6 @@ except ImportError: from dvc.dependency.base import DependencyBase from dvc.remote.s3 import RemoteS3 -from dvc.config import Config class DependencyS3(DependencyBase): diff --git a/dvc/dependency/ssh.py b/dvc/dependency/ssh.py index 4be409082..a6d555062 100644 --- a/dvc/dependency/ssh.py +++ b/dvc/dependency/ssh.py @@ -20,7 +20,8 @@ class DependencySSH(DependencyBase): host = remote.host if remote else self.group('host') user = remote.user if remote else self.group('user') if remote: - path = posixpath.join(remote.prefix, urlparse(path).path.lstrip('/')) + path = posixpath.join(remote.prefix, + urlparse(path).path.lstrip('/')) else: path = self.match(self.path).group('path') diff --git a/dvc/exceptions.py b/dvc/exceptions.py index 0232c612c..f1d6c6ec1 100644 --- a/dvc/exceptions.py +++ b/dvc/exceptions.py @@ -27,4 +27,3 @@ class OutputDuplicationError(DvcException): msg = "File '{}' is specified as an output in more than one stage: {}" super(OutputDuplicationError, self).__init__(msg.format(output, str(stages))) - diff --git a/dvc/istextfile.py b/dvc/istextfile.py index 1c624acba..310d4738b 100644 --- a/dvc/istextfile.py +++ b/dvc/istextfile.py @@ -1,4 +1,6 @@ -# Based on https://eli.thegreenplace.net/2011/10/19/perls-guess-if-file-is-text-or-binary-implemented-in-python +# Based on https://eli.thegreenplace.net/2011/10/19/ +# perls-guess-if-file-is-text-or-binary-implemented-in-python + import sys PY3 = sys.version_info[0] == 3 diff --git a/dvc/lock.py b/dvc/lock.py index 68a659562..81781975f 100644 --- a/dvc/lock.py +++ b/dvc/lock.py @@ -25,7 +25,8 @@ class Lock(object): try: self._lock = zc.lockfile.LockFile(self.lock_file) except zc.lockfile.LockError: - raise LockError('Cannot perform the cmd since DVC is busy and locked. Please retry the cmd later.') + raise LockError('Cannot perform the cmd since DVC is busy and ' + 'locked. Please retry the cmd later.') def lock(self): try: diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index baffdbc2c..5a6488350 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -29,9 +29,15 @@ SCHEMA[schema.Optional(OutputLOCAL.PARAM_METRIC)] = OutputLOCAL.METRIC_SCHEMA def _get(stage, p, info, cache, metric): parsed = urlparse(p) if parsed.scheme == 'remote': - sect = stage.project.config._config[Config.SECTION_REMOTE_FMT.format(parsed.netloc)] + name = Config.SECTION_REMOTE_FMT.format(parsed.netloc) + sect = stage.project.config._config[name] remote = Remote(stage.project, sect) - return OUTS_MAP[remote.scheme](stage, p, info, cache=cache, remote=remote, metric=metric) + return OUTS_MAP[remote.scheme](stage, + p, + info, + cache=cache, + remote=remote, + metric=metric) for o in OUTS: if o.supported(p): diff --git a/dvc/output/base.py b/dvc/output/base.py index b4fb074eb..9ca2a17e1 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -3,7 +3,8 @@ from dvc.exceptions import DvcException class OutputError(DvcException): def __init__(self, path, msg): - super(OutputError, self).__init__('Output \'{}\' error: {}'.format(path, msg)) + msg = 'Output \'{}\' error: {}'.format(path, msg) + super(OutputError, self).__init__(msg) class OutputDoesNotExistError(OutputError): @@ -13,9 +14,11 @@ class OutputDoesNotExistError(OutputError): class OutputIsNotFileOrDirError(OutputError): def __init__(self, path): - super(OutputIsNotFileOrDirError, self).__init__(path, 'not a file or directory') + msg = 'not a file or directory' + super(OutputIsNotFileOrDirError, self).__init__(path, msg) class OutputAlreadyTrackedError(OutputError): def __init__(self, path): - super(OutputAlreadyTrackedError, self).__init__(path, 'already tracked by scm(e.g. git)') + msg = 'already tracked by scm(e.g. git)' + super(OutputAlreadyTrackedError, self).__init__(path, msg) diff --git a/dvc/output/gs.py b/dvc/output/gs.py index e53a20fea..82e53b154 100644 --- a/dvc/output/gs.py +++ b/dvc/output/gs.py @@ -6,11 +6,17 @@ class OutputGS(DependencyGS): PARAM_CACHE = 'cache' PARAM_METRIC = 'metric' - def __init__(self, stage, path, info=None, remote=None, cache=True, metric=False): + def __init__(self, + stage, + path, + info=None, + remote=None, + cache=True, + metric=False): super(OutputGS, self).__init__(stage, path, info, remote=remote) self.use_cache = cache self.metric = metric - if cache and self.project.cache.gs == None: + if cache and self.project.cache.gs is None: raise DvcException("No cache location setup for \'gs\' outputs.") def dumpd(self): @@ -23,7 +29,8 @@ class OutputGS(DependencyGS): if super(OutputGS, self).changed(): return True - if self.use_cache and self.info != self.project.cache.gs.save_info(self.path_info): + if self.use_cache \ + and self.info != self.project.cache.gs.save_info(self.path_info): return True return False @@ -40,7 +47,6 @@ class OutputGS(DependencyGS): if not self.use_cache: return - self.info = self.project.cache.gs.save(self.path_info) def remove(self, ignore_remove=False): diff --git a/dvc/output/hdfs.py b/dvc/output/hdfs.py index 49bb6b3fb..ec81f23a0 100644 --- a/dvc/output/hdfs.py +++ b/dvc/output/hdfs.py @@ -6,11 +6,17 @@ class OutputHDFS(DependencyHDFS): PARAM_CACHE = 'cache' PARAM_METRIC = 'metric' - def __init__(self, stage, path, info=None, remote=None, cache=True, metric=False): + def __init__(self, + stage, + path, + info=None, + remote=None, + cache=True, + metric=False): super(OutputHDFS, self).__init__(stage, path, info, remote=remote) self.use_cache = cache self.metric = metric - if cache and self.project.cache.hdfs == None: + if cache and self.project.cache.hdfs is None: raise DvcException("No cache location setup for \'hdfs\' outputs.") def dumpd(self): @@ -23,7 +29,8 @@ class OutputHDFS(DependencyHDFS): if super(OutputHDFS, self).changed(): return True - if self.use_cache and self.info != self.project.cache.hdfs.save_info(self.path_info): + if self.use_cache \ + and self.info != self.project.cache.hdfs.save_info(self.path_info): return True return False diff --git a/dvc/output/local.py b/dvc/output/local.py index 8be5220d1..4f978418e 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -1,13 +1,10 @@ import os -import schema -import posixpath -import ntpath +from schema import Optional, Or -from dvc.output.base import * +from dvc.output.base import OutputDoesNotExistError, OutputIsNotFileOrDirError +from dvc.output.base import OutputAlreadyTrackedError from dvc.dependency.local import DependencyLOCAL from dvc.exceptions import DvcException -from dvc.logger import Logger -from dvc.utils import remove from dvc.istextfile import istextfile @@ -17,14 +14,20 @@ class OutputLOCAL(DependencyLOCAL): PARAM_METRIC_TYPE = 'type' PARAM_METRIC_XPATH = 'xpath' - METRIC_SCHEMA = schema.Or(None, bool, - {schema.Optional(PARAM_METRIC_TYPE): schema.Or(str, None), - schema.Optional(PARAM_METRIC_XPATH): schema.Or(str, None)}) + METRIC_SCHEMA = Or(None, bool, + {Optional(PARAM_METRIC_TYPE): Or(str, None), + Optional(PARAM_METRIC_XPATH): Or(str, None)}) DoesNotExistError = OutputDoesNotExistError IsNotFileOrDirError = OutputIsNotFileOrDirError - def __init__(self, stage, path, info=None, remote=None, cache=True, metric=False): + def __init__(self, + stage, + path, + info=None, + remote=None, + cache=True, + metric=False): super(OutputLOCAL, self).__init__(stage, path, info, remote=remote) self.use_cache = cache self.metric = metric @@ -100,7 +103,8 @@ class OutputLOCAL(DependencyLOCAL): if (os.path.isfile(self.path) and os.path.getsize(self.path) == 0) or \ (os.path.isdir(self.path) and len(os.listdir(self.path)) == 0): - self.project.logger.warn("File/directory '{}' is empty.".format(self.rel_path)) + msg = "File/directory '{}' is empty.".format(self.rel_path) + self.project.logger.warn(msg) if not self.changed(): msg = 'Output \'{}\' didn\'t change. Skipping saving.' diff --git a/dvc/output/s3.py b/dvc/output/s3.py index fd80bc182..620ea7d7c 100644 --- a/dvc/output/s3.py +++ b/dvc/output/s3.py @@ -6,11 +6,17 @@ class OutputS3(DependencyS3): PARAM_CACHE = 'cache' PARAM_METRIC = 'metric' - def __init__(self, stage, path, info=None, remote=None, cache=True, metric=False): + def __init__(self, + stage, + path, + info=None, + remote=None, + cache=True, + metric=False): super(OutputS3, self).__init__(stage, path, info, remote=remote) self.use_cache = cache self.metric = metric - if cache and self.project.cache.s3 == None: + if cache and self.project.cache.s3 is None: raise DvcException("No cache location setup for \'s3\' outputs.") def dumpd(self): @@ -23,7 +29,8 @@ class OutputS3(DependencyS3): if super(OutputS3, self).changed(): return True - if self.use_cache and self.info != self.project.cache.s3.save_info(self.path_info): + if self.use_cache \ + and self.info != self.project.cache.s3.save_info(self.path_info): return True return False diff --git a/dvc/output/ssh.py b/dvc/output/ssh.py index c7318881d..8fa04c2ab 100644 --- a/dvc/output/ssh.py +++ b/dvc/output/ssh.py @@ -6,11 +6,17 @@ class OutputSSH(DependencySSH): PARAM_CACHE = 'cache' PARAM_METRIC = 'metric' - def __init__(self, stage, path, info=None, remote=None, cache=True, metric=False): + def __init__(self, + stage, + path, + info=None, + remote=None, + cache=True, + metric=False): super(OutputSSH, self).__init__(stage, path, info, remote=remote) self.use_cache = cache self.metric = metric - if cache and self.project.cache.ssh == None: + if cache and self.project.cache.ssh is None: raise DvcException("No cache location setup for \'ssh\' outputs.") def dumpd(self): @@ -23,7 +29,8 @@ class OutputSSH(DependencySSH): if super(OutputSSH, self).changed(): return True - if self.use_cache and self.info != self.project.cache.ssh.save_info(self.path_info): + if self.use_cache \ + and self.info != self.project.cache.ssh.save_info(self.path_info): return True return False diff --git a/dvc/progress.py b/dvc/progress.py index a40ba83f8..b49c25045 100644 --- a/dvc/progress.py +++ b/dvc/progress.py @@ -1,6 +1,7 @@ from __future__ import print_function import sys -import threading +import threading + class Progress(object): """ @@ -55,7 +56,7 @@ class Progress(object): total_len = 100 bar_len = 30 - if total == None: + if total is None: progress = 0 percent = "?% " else: @@ -70,8 +71,12 @@ class Progress(object): bar = "[" + '#'*n_sh + ' '*n_sp + "] " name_len = total_len - len(num + bar + percent) - name = target_name[:name_len] if len(target_name) > name_len else target_name + if len(target_name) > name_len: + name = target_name[:name_len] + else: + name = target_name return num + bar + percent + name + progress = Progress() diff --git a/dvc/project.py b/dvc/project.py index 641ae6309..26375d46b 100644 --- a/dvc/project.py +++ b/dvc/project.py @@ -1,5 +1,4 @@ import os -import stat from dvc.exceptions import DvcException from dvc.stage import Stage @@ -34,7 +33,7 @@ class Project(object): from dvc.config import Config from dvc.state import LinkState, State from dvc.lock import Lock - from dvc.scm import SCM, Base + from dvc.scm import SCM from dvc.cache import Cache from dvc.data_cloud import DataCloud from dvc.updater import Updater @@ -49,7 +48,10 @@ class Project(object): # any possible state corruption in 'shared cache dir' scenario. self.state = State(self) self.link_state = LinkState(self) - self.logger = Logger(self.config._config[Config.SECTION_CORE].get(Config.SECTION_CORE_LOGLEVEL, None)) + + core = self.config._config[Config.SECTION_CORE] + self.logger = Logger(core.get(Config.SECTION_CORE_LOGLEVEL, None)) + self.cache = Cache(self) self.cloud = DataCloud(self, config=self.config._config) self.updater = Updater(self.dvc_dir) @@ -81,8 +83,8 @@ class Project(object): scm = SCM(root_dir) if type(scm) == Base and not no_scm: - msg = "{} is not tracked by any supported scm tool(e.g. git).".format(root_dir) - raise InitError(msg) + msg = "{} is not tracked by any supported scm tool(e.g. git)." + raise InitError(msg.format(root_dir)) if os.path.isdir(dvc_dir): if not force: @@ -110,18 +112,18 @@ class Project(object): shutil.rmtree(self.dvc_dir) def _ignore(self): - l = [self.state.state_file, - self.state._lock_file.lock_file, - self.link_state.state_file, - self.link_state._lock_file.lock_file, - self.lock.lock_file, - self.config.config_local_file, - self.updater.updater_file] + flist = [self.state.state_file, + self.state._lock_file.lock_file, + self.link_state.state_file, + self.link_state._lock_file.lock_file, + self.lock.lock_file, + self.config.config_local_file, + self.updater.updater_file] if self.cache.local.cache_dir.startswith(self.root_dir): - l += [self.cache.local.cache_dir] + flist += [self.cache.local.cache_dir] - self.scm.ignore_list(l) + self.scm.ignore_list(flist) def install(self): self.scm.install() @@ -163,7 +165,8 @@ class Project(object): def move(self, from_path, to_path): import dvc.output as Output - from_out = Output.loads_from(Stage(self, cwd=os.curdir), [from_path])[0] + from_out = Output.loads_from(Stage(self, cwd=os.curdir), + [from_path])[0] found = False for stage in self.stages(): @@ -172,24 +175,32 @@ class Project(object): continue if not stage.is_data_source: - raise DvcException('Dvcfile \'{}\' is not a data source.'.format(stage.rel_path)) + msg = 'Dvcfile \'{}\' is not a data source.' + raise DvcException(msg.format(stage.rel_path)) found = True - to_out = Output.loads_from(out.stage, [to_path], out.cache, out.metric)[0] + to_out = Output.loads_from(out.stage, + [to_path], + out.cache, + out.metric)[0] out.move(to_out) - stage_base = os.path.basename(stage.path).rstrip(Stage.STAGE_FILE_SUFFIX) + stage_base = os.path.basename(stage.path) + stage_base = stage_base.rstrip(Stage.STAGE_FILE_SUFFIX) + stage_dir = os.path.dirname(stage.path) from_base = os.path.basename(from_path) to_base = os.path.basename(to_path) if stage_base == from_base: os.unlink(stage.path) - stage.path = os.path.join(stage_dir, to_base + Stage.STAGE_FILE_SUFFIX) + path = to_base + Stage.STAGE_FILE_SUFFIX + stage.path = os.path.join(stage_dir, path) stage.dump() if not found: - raise DvcException('Unable to find dvcfile with output \'{}\''.format(from_path)) + msg = 'Unable to find dvcfile with output \'{}\'' + raise DvcException(msg.format(from_path)) def run(self, cmd=None, @@ -223,7 +234,6 @@ class Project(object): stage.dump() return stage - def _reproduce_stage(self, stages, node, force): stage = stages[node] @@ -308,8 +318,8 @@ class Project(object): for stage in stages: if active and not target and stage.locked: - msg = 'DVC file \'{}\' is locked. Its dependecies are not ' \ - 'going to be pushed/pulled/fetched.' + msg = 'DVC file \'{}\' is locked. Its dependecies are ' \ + 'not going to be pushed/pulled/fetched.' self.logger.warn(msg.format(stage.relpath)) for out in stage.outs: @@ -345,10 +355,14 @@ class Project(object): self.cloud._get_cloud(remote).gc(clist) def push(self, target=None, jobs=1, remote=None, all_branches=False): - self.cloud.push(self._used_cache(target, all_branches)['local'], jobs, remote=remote) + self.cloud.push(self._used_cache(target, all_branches)['local'], + jobs, + remote=remote) def fetch(self, target=None, jobs=1, remote=None, all_branches=False): - self.cloud.pull(self._used_cache(target, all_branches)['local'], jobs, remote=remote) + self.cloud.pull(self._used_cache(target, all_branches)['local'], + jobs, + remote=remote) def pull(self, target=None, jobs=1, remote=None, all_branches=False): self.fetch(target, jobs, remote=remote, all_branches=all_branches) @@ -376,7 +390,9 @@ class Project(object): import dvc.remote.base as cloud status = {} - for md5, ret in self.cloud.status(self._used_cache(target)['local'], jobs, remote=remote): + for md5, ret in self.cloud.status(self._used_cache(target)['local'], + jobs, + remote=remote): if ret == cloud.STATUS_OK: continue @@ -402,11 +418,11 @@ class Project(object): return [x.value for x in parser.find(json.load(fd))] def _do_read_metric_xsv(self, reader, row, col): - if col != None and row != None: + if col is not None and row is not None: return [reader[row][col]] - elif col != None: + elif col is not None: return [r[col] for r in reader] - elif row != None: + elif row is not None: return reader[row] return None @@ -433,7 +449,7 @@ class Project(object): if not os.path.exists(path): return ret - try: + try: with open(path, 'r') as fd: if typ == 'json': ret = self._read_metric_json(fd, xpath) @@ -448,16 +464,17 @@ class Project(object): else: ret = fd.read() except Exception as exc: - self.logger.error('Unable to read metric in \'{}\''.format(path), exc) + self.logger.error('Unable to read metric in \'{}\''.format(path), + exc) return ret - def _find_output_by_path(self, path, outs=None): from dvc.exceptions import OutputDuplicationError if not outs: - outs = [out for stage in self.active_stages() for out in stage.outs] + astages = self.active_stages() + outs = [out for stage in astages for out in stage.outs] abs_path = os.path.abspath(path) matched = [out for out in outs if out.path == abs_path] @@ -467,15 +484,22 @@ class Project(object): return matched[0] if matched else None - def metrics_show(self, path=None, typ=None, xpath=None, all_branches=False): + def metrics_show(self, + path=None, + typ=None, + xpath=None, + all_branches=False): res = {} for branch in self.scm.brancher(all_branches=all_branches): - outs = [out for stage in self.active_stages() for out in stage.outs] + astages = self.active_stages() + outs = [out for stage in astages for out in stage.outs] if path: out = self._find_output_by_path(path, outs=outs) stage = out.stage.path if out else None - if out and all([out.metric, not typ, isinstance(out.metric, dict)]): + if out and all([out.metric, + not typ, + isinstance(out.metric, dict)]): entries = [(path, out.metric.get(out.PARAM_METRIC_TYPE, None), out.metric.get(out.PARAM_METRIC_XPATH, None))] @@ -527,7 +551,6 @@ class Project(object): raise DvcException(msg) def _metrics_modify(self, path, typ=None, xpath=None, delete=False): - found = False out = self._find_output_by_path(path) if not out: msg = 'Unable to find file \'{}\' in the pipeline'.format(path) @@ -589,7 +612,8 @@ class Project(object): for dep in stage.deps: for out in outs: - if out.path != dep.path and not dep.path.startswith(out.path + out.sep): + if out.path != dep.path \ + and not dep.path.startswith(out.path + out.sep): continue dep_stage = out.stage diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index b36c5aadf..351aa859d 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -1,23 +1,16 @@ from __future__ import absolute_import import os import re -import time try: from azure.storage.blob import BlockBlobService except ImportError: BlockBlobService = None -try: - from urlparse import urlparse -except ImportError: - from urllib.parse import urlparse - 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 class Callback(object): @@ -180,11 +173,16 @@ class RemoteAzure(RemoteBase): self.blob_service.create_blob_from_path( bucket, key, from_info['path'], progress_callback=cb) except Exception as ex: - Logger.error("Failed to upload '{}'".format(from_info['path']), ex) + Logger.error("Failed to upload '{}'".format(from_info['path']), + ex) else: progress.finish_target(name) - def download(self, from_infos, to_infos, no_progress_bar=False, names=None): + def download(self, + from_infos, + to_infos, + no_progress_bar=False, + names=None): names = self._verify_path_args(from_infos, to_infos, names) for to_info, from_info, name in zip(to_infos, from_infos, names): @@ -223,15 +221,15 @@ class RemoteAzure(RemoteBase): # FIXME: temporarily disabled because of the lack of test for external azure # dependencies/outputs/cache. # -# def gc(self, checksum_infos): -# used_etags = [info[self.PARAM_ETAG] for info in checksum_infos['azure']] -# used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in checksum_infos['local']] +# def gc(self, cinfos): +# used = [info[self.PARAM_ETAG] for info in cinfos['azure']] +# used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] # # all_blobs = self.blob_service.list_blobs(self.bucket) # # for blob in all_blobs: # etag = blob.properties.etag -# if etag in used_etags: +# if etag in used: # continue # path_info = {'scheme': self.scheme, # 'key': blob.name, diff --git a/dvc/remote/base.py b/dvc/remote/base.py index 0a6b6b4af..a9e7911b0 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -1,7 +1,5 @@ import os import re -import tempfile -import posixpath from dvc.config import Config from dvc.logger import Logger @@ -15,10 +13,10 @@ STATUS_DELETED = 4 STATUS_MAP = { # (local_exists, remote_exists) - (True, True) : STATUS_OK, - (False, False) : STATUS_OK, - (True, False) : STATUS_NEW, - (False, True) : STATUS_DELETED, + (True, True): STATUS_OK, + (False, False): STATUS_OK, + (True, False): STATUS_NEW, + (False, True): STATUS_DELETED, } @@ -28,7 +26,6 @@ class DataCloudError(DvcException): super(DataCloudError, self).__init__('Data sync error: {}'.format(msg)) - class RemoteBase(object): REGEX = None REQUIRES = {} @@ -42,8 +39,9 @@ class RemoteBase(object): url_ok = cls.match(url) deps_ok = all(cls.REQUIRES.values()) if url_ok and not deps_ok: - missing = [k for k,v in cls.REQUIRES.items() if v == None] - msg = "URL \'{}\' is supported but requires these missing dependencies: {}" + missing = [k for k, v in cls.REQUIRES.items() if v is None] + msg = "URL \'{}\' is supported but requires " \ + "these missing dependencies: {}" Logger.warn(msg.format(url, str(missing))) return url_ok and deps_ok @@ -60,7 +58,7 @@ class RemoteBase(object): @staticmethod def tmp_file(fname): """ Temporary name for a partial download """ - #FIXME probably better use uuid() + # FIXME probably better use uuid() return fname + '.part' def save_info(self, path_info): diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index 9413e6be1..aab3b4bf3 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -27,7 +27,9 @@ class RemoteGS(RemoteBase): def __init__(self, project, config): self.project = project - storagepath = 'gs://' + config.get(Config.SECTION_AWS_STORAGEPATH, '/').lstrip('/') + storagepath = 'gs://' + storagepath += config.get(Config.SECTION_AWS_STORAGEPATH, '/') + storagepath.lstrip('/') self.url = config.get(Config.SECTION_REMOTE_URL, storagepath) self.projectname = config.get(Config.SECTION_GCP_PROJECTNAME, None) @@ -54,16 +56,21 @@ class RemoteGS(RemoteBase): if path_info['scheme'] != 'gs': raise NotImplementedError - return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], path_info['key'])} + return {self.PARAM_ETAG: self.get_etag(path_info['bucket'], + path_info['key'])} 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']) if not blob: - raise DvcException('{} doesn\'t exist in the cloud'.format(from_info['key'])) + msg = '{} doesn\'t exist in the cloud' + raise DvcException(msg.format(from_info['key'])) - self.gs.bucket(to_info['bucket']).copy_blob(blob, self.gs.bucket(to_info['bucket']), new_name=to_info['key']) + bucket = self.gs.bucket(to_info['bucket']) + bucket.copy_blob(blob, + self.gs.bucket(to_info['bucket']), + new_name=to_info['key']) def save(self, path_info): if path_info['scheme'] != 'gs': @@ -106,13 +113,15 @@ class RemoteGS(RemoteBase): 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] + 'key': posixpath.join(self.prefix, + md5[0:2], md5[2:])} for md5 in md5s] def exists(self, path_infos): ret = [] gs = self.gs - keys = [blob.name for blob in gs.bucket(self.bucket).list_blobs(prefix=self.prefix)] + bucket = gs.bucket(self.bucket) + keys = [blob.name for blob in bucket.list_blobs(prefix=self.prefix)] for path_info in path_infos: exists = False @@ -143,16 +152,23 @@ class RemoteGS(RemoteBase): progress.update_target(name, 0, None) try: - gs.bucket(to_info['bucket']).blob(to_info['key']).upload_from_filename(from_info['path']) + bucket = gs.bucket(to_info['bucket']) + blob = bucket.blob(to_info['key']) + blob.upload_from_filename(from_info['path']) except Exception as exc: - Logger.error("Failed to upload '{}' to '{}/{}'".format(from_info['path'], - to_info['bucket'], - to_info['key']), exc) + msg = "Failed to upload '{}' to '{}/{}'" + Logger.error(msg.format(from_info['path'], + to_info['bucket'], + to_info['key']), exc) continue progress.finish_target(name) - def download(self, from_infos, to_infos, no_progress_bar=False, names=None): + def download(self, + from_infos, + to_infos, + no_progress_bar=False, + names=None): names = self._verify_path_args(from_infos, to_infos, names) gs = self.gs @@ -168,9 +184,10 @@ class RemoteGS(RemoteBase): if to_info['scheme'] != 'local': raise NotImplementedError - Logger.debug("Downloading '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], - to_info['path'])) + msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'], + from_info['key'], + to_info['path']) + Logger.debug(msg) tmp_file = self.tmp_file(to_info['path']) if not name: @@ -184,11 +201,14 @@ class RemoteGS(RemoteBase): self._makedirs(to_info['path']) try: - gs.bucket(from_info['bucket']).get_blob(from_info['key']).download_to_filename(tmp_file) + bucket = gs.bucket(from_info['bucket']) + blob = bucket.get_blob(from_info['key']) + blob.download_to_filename(tmp_file) except Exception as exc: - Logger.error("Failed to download '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], - to_info['path']), exc) + msg = "Failed to download '{}/{}' to '{}'" + Logger.error(msg.format(from_info['bucket'], + from_info['key'], + to_info['path']), exc) continue os.rename(tmp_file, to_info['path']) @@ -201,17 +221,19 @@ class RemoteGS(RemoteBase): return posixpath.dirname(relpath) + posixpath.basename(relpath) def _all_etags(self): - blobs = list(self.gs.bucket(self.bucket).list_blobs(prefix=self.prefix)) + blobs = self.gs.bucket(self.bucket).list_blobs(prefix=self.prefix) + blobs = list(blobs) return [self._path_to_etag(blob.name) for blob in blobs] - def gc(self, checksum_infos): - used_etags = [info[self.PARAM_ETAG] for info in checksum_infos['gs']] - used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in checksum_infos['local']] + def gc(self, cinfos): + used = [info[self.PARAM_ETAG] for info in cinfos['gs']] + used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] for etag in self._all_etags(): - if etag in used_etags: + if etag in used: continue path_info = {'scheme': 'gs', 'bucket': self.bucket, - 'key': posixpath.join(self.prefix, etag[0:2], etag[2:])} + 'key': posixpath.join(self.prefix, + etag[0:2], etag[2:])} self.remove(path_info) diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 9c73c0b0e..612ab57fe 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -4,11 +4,6 @@ import getpass import posixpath from subprocess import Popen, PIPE -try: - from urlparse import urlparse -except ImportError: - from urllib.parse import urlparse - from dvc.config import Config from dvc.remote.base import RemoteBase from dvc.remote.local import RemoteLOCAL @@ -17,7 +12,7 @@ from dvc.logger import Logger class RemoteHDFS(RemoteBase): - scheme='hdfs' + scheme = 'hdfs' REGEX = r'^hdfs://((?P.*)@)?.*$' PARAM_CHECKSUM = 'checksum' @@ -26,7 +21,8 @@ class RemoteHDFS(RemoteBase): self.url = config.get(Config.SECTION_REMOTE_URL, '/') self.user = self.group('user') if not self.user: - self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser()) + self.user = config.get(Config.SECTION_REMOTE_USER, + getpass.getuser()) def hadoop_fs(self, cmd, user=None): cmd = 'hadoop fs -' + cmd @@ -47,20 +43,24 @@ class RemoteHDFS(RemoteBase): @staticmethod def _group(regex, s, gname): match = re.match(regex, s) - assert match != None + assert match is not None return match.group(gname) def checksum(self, path_info): regex = r'.*\t.*\t(?P.*)' - stdout = self.hadoop_fs('checksum {}'.format(path_info['url']), user=path_info['user']) + stdout = self.hadoop_fs('checksum {}'.format(path_info['url']), + 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']), user=to_info['user']) + 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']), + user=to_info['user']) def rm(self, path_info): - self.hadoop_fs('rm {}'.format(path_info['url']), user=path_info['user']) + self.hadoop_fs('rm {}'.format(path_info['url']), + user=path_info['user']) def save_info(self, path_info): if path_info['scheme'] != 'hdfs': @@ -112,7 +112,8 @@ class RemoteHDFS(RemoteBase): 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] + 'url': posixpath.join(self.url, + md5[0:2], md5[2:])} for md5 in md5s] def exists(self, path_infos): try: @@ -146,10 +147,18 @@ class RemoteHDFS(RemoteBase): if from_info['scheme'] != 'local': raise NotImplementedError - self.hadoop_fs('mkdir -p {}'.format(posixpath.dirname(to_info['url'])), user=to_info['user']) - self.hadoop_fs('copyFromLocal {} {}'.format(from_info['path'], to_info['url']), user=to_info['user']) + cmd = 'mkdir -p {}'.format(posixpath.dirname(to_info['url'])) + self.hadoop_fs(cmd, user=to_info['user']) - def download(self, from_infos, to_infos, no_progress_bar=False, names=None): + cmd = 'copyFromLocal {} {}'.format(from_info['path'], + to_info['url']) + self.hadoop_fs(cmd, user=to_info['user']) + + def download(self, + from_infos, + to_infos, + no_progress_bar=False, + names=None): names = self._verify_path_args(from_infos, to_infos, names) for to_info, from_info, name in zip(to_infos, from_infos, names): @@ -167,7 +176,8 @@ class RemoteHDFS(RemoteBase): if not os.path.exists(dname): os.makedirs(dname) - self.hadoop_fs('copyToLocal {} {}'.format(from_info['url'], to_info['path']), user=from_info['user']) + cmd = 'copyToLocal {} {}'.format(from_info['url'], to_info['path']) + self.hadoop_fs(cmd, user=from_info['user']) def _path_to_checksum(self, path): relpath = posixpath.relpath(path, self.url) @@ -183,14 +193,15 @@ class RemoteHDFS(RemoteBase): flist.append(line.split()[-1]) return [self._path_to_checksum(path) for path in flist] - def gc(self, checksum_infos): - used_checksums = [info[self.PARAM_CHECKSUM] for info in checksum_infos['hdfs']] - used_checksums += [info[RemoteLOCAL.PARAM_MD5] for info in checksum_infos['local']] + def gc(self, cinfos): + used = [info[self.PARAM_CHECKSUM] for info in cinfos['hdfs']] + used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] for checksum in self._all_checksums(): - if checksum in used_checksums: + if checksum in used: continue path_info = {'scheme': 'hdfs', 'user': self.user, - 'url': posixpath.join(self.url, checksum[0:2], checksum[2:])} + 'url': posixpath.join(self.url, + checksum[0:2], checksum[2:])} self.remove(path_info) diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 75d18d8dc..7464ab9eb 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -3,7 +3,6 @@ import uuid import json import ntpath import shutil -import filecmp import posixpath from operator import itemgetter @@ -14,7 +13,7 @@ from dvc.utils import remove, move, copyfile, file_md5, dict_md5, to_chunks from dvc.config import Config from dvc.exceptions import DvcException from dvc.progress import progress -from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor +from concurrent.futures import ThreadPoolExecutor class RemoteLOCAL(RemoteBase): @@ -47,7 +46,7 @@ class RemoteLOCAL(RemoteBase): else: self.cache_types = self.CACHE_TYPES - if self.cache_dir != None and not os.path.exists(self.cache_dir): + if self.cache_dir is not None and not os.path.exists(self.cache_dir): os.mkdir(self.cache_dir) @property @@ -81,7 +80,8 @@ class RemoteLOCAL(RemoteBase): cache = self.get(md5) if self.state.changed(cache, md5=md5): if os.path.exists(cache): - Logger.warn('Corrupted cache file {}'.format(os.path.relpath(cache))) + msg = 'Corrupted cache file {}' + Logger.warn(msg.format(os.path.relpath(cache))) remove(cache) return True @@ -100,8 +100,8 @@ class RemoteLOCAL(RemoteBase): self.CACHE_TYPE_MAP[self.cache_types[0]](cache, path) return except Exception as exc: - msg = 'Cache type \'{}\' is not supported'.format(self.cache_types[0]) - Logger.debug(msg) + msg = 'Cache type \'{}\' is not supported: {}' + Logger.debug(msg.format(self.cache_types[0], str(exc))) del self.cache_types[0] i -= 1 @@ -133,7 +133,7 @@ class RemoteLOCAL(RemoteBase): path = os.path.join(root, fname) relpath = self.unixpath(os.path.relpath(path, dname)) - # FIXME: we could've used `md5 = state.update(path, dump=False)` + # FIXME: we could've used md5 = state.update(path, dump=False) # here, but it is around twice as slow(on ssd, don't know about # hdd) for a directory with small files. What we could do here # is introduce some kind of a limit for file size, after which @@ -201,7 +201,8 @@ class RemoteLOCAL(RemoteBase): cache = self.get(md5) if not cache: - Logger.warn('No cache info for \'{}\'. Skipping checkout.'.format(os.path.relpath(path))) + msg = 'No cache info for \'{}\'. Skipping checkout.' + Logger.warn(msg.format(os.path.relpath(path))) return if self.changed(md5): @@ -246,7 +247,7 @@ class RemoteLOCAL(RemoteBase): def _save_file(self, path_info): path = path_info['path'] md5 = self.state.update(path) - assert md5 != None + assert md5 is not None cache = self.get(md5) @@ -312,7 +313,8 @@ class RemoteLOCAL(RemoteBase): 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] + 'path': os.path.join(self.prefix, + md5[0:2], md5[2:])} for md5 in md5s] def exists(self, path_infos): ret = [] @@ -330,7 +332,8 @@ class RemoteLOCAL(RemoteBase): if from_info['scheme'] != 'local': raise NotImplementedError - Logger.debug("Uploading '{}' to '{}'".format(from_info['path'], to_info['path'])) + Logger.debug("Uploading '{}' to '{}'".format(from_info['path'], + to_info['path'])) if not name: name = os.path.basename(from_info['path']) @@ -340,9 +343,15 @@ class RemoteLOCAL(RemoteBase): try: copyfile(from_info['path'], to_info['path'], name=name) except Exception as exc: - Logger.error("Failed to upload '{}' tp '{}'".format(from_info['path'], to_info['path']), exc) + msg = "Failed to upload '{}' tp '{}'" + Logger.error(msg.format(from_info['path'], + to_info['path']), exc) - def download(self, from_infos, to_infos, no_progress_bar=False, names=None): + def download(self, + from_infos, + to_infos, + no_progress_bar=False, + names=None): names = self._verify_path_args(from_infos, to_infos, names) for to_info, from_info, name in zip(to_infos, from_infos, names): @@ -352,7 +361,8 @@ class RemoteLOCAL(RemoteBase): if to_info['scheme'] != 'local': raise NotImplementedError - Logger.debug("Downloading '{}' to '{}'".format(from_info['path'], to_info['path'])) + Logger.debug("Downloading '{}' to '{}'".format(from_info['path'], + to_info['path'])) if not name: name = os.path.basename(to_info['path']) @@ -360,9 +370,14 @@ class RemoteLOCAL(RemoteBase): self._makedirs(to_info['path']) tmp_file = self.tmp_file(to_info['path']) try: - copyfile(from_info['path'], tmp_file, no_progress_bar=no_progress_bar, name=name) + copyfile(from_info['path'], + tmp_file, + no_progress_bar=no_progress_bar, + name=name) except Exception as exc: - Logger.error("Failed to download '{}' to '{}'".format(from_info['path'], to_info['path']), exc) + msg = "Failed to download '{}' to '{}'" + Logger.error(msg.format(from_info['path'], + to_info['path']), exc) continue os.rename(tmp_file, to_info['path']) @@ -383,7 +398,8 @@ class RemoteLOCAL(RemoteBase): return collected, missing def gc(self, checksum_infos): - used_md5s = [info[self.PARAM_MD5] for info in self._collect(checksum_infos['local'])[0]] + checksum_infos = self._collect(checksum_infos['local'])[0] + used_md5s = [info[self.PARAM_MD5] for info in checksum_infos] for md5 in self.all(): if md5 in used_md5s: @@ -397,7 +413,9 @@ class RemoteLOCAL(RemoteBase): remote_exists = remote.exists(path_infos) local_exists = [not self.changed(md5) for md5 in md5s] - return [(md5, STATUS_MAP[l,r]) for md5, l, r in zip(md5s, local_exists, remote_exists)] + return [(md5, STATUS_MAP[l, r]) for md5, l, r in zip(md5s, + local_exists, + remote_exists)] def _do_pull(self, checksum_infos, remote, jobs=1, no_progress_bar=False): md5s = [info[self.PARAM_MD5] for info in checksum_infos] @@ -442,14 +460,16 @@ class RemoteLOCAL(RemoteBase): self._do_pull(checksum_infos, remote, jobs) def push(self, checksum_infos, remote, jobs=1): - md5s = [info[self.PARAM_MD5] for info in self._collect(checksum_infos)[0]] + checksum_infos = self._collect(checksum_infos)[0] + md5s = [info[self.PARAM_MD5] for info in checksum_infos] # NOTE: verifying that our cache is not corrupted md5s = list(filter(lambda md5: not self.changed(md5), md5s)) # NOTE: filter files that are already uploaded path_infos = remote.md5s_to_path_infos(md5s) - md5s_exist = filter(lambda x: not x[1], list(zip(md5s, remote.exists(path_infos)))) + lexists = remote.exists(path_infos) + md5s_exist = filter(lambda x: not x[1], list(zip(md5s, lexists))) md5s = [md5 for md5, exists in md5s_exist] cache = [{'scheme': 'local', 'path': self.get(md5)} for md5 in md5s] @@ -469,7 +489,10 @@ class RemoteLOCAL(RemoteBase): futures = [] with ThreadPoolExecutor(max_workers=len(chunks)) as executor: for to_infos, from_infos, md5s in chunks: - res = executor.submit(remote.upload, from_infos, to_infos, names=md5s) + res = executor.submit(remote.upload, + from_infos, + to_infos, + names=md5s) futures.append(res) for f in futures: diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 0e5662f57..c5d5958fd 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -1,5 +1,4 @@ import os -import math import threading import posixpath @@ -44,7 +43,8 @@ class RemoteS3(RemoteBase): import configobj self.project = project - storagepath = 's3://' + config.get(Config.SECTION_AWS_STORAGEPATH, '').lstrip('/') + storagepath = 's3://' + config.get(Config.SECTION_AWS_STORAGEPATH, + '').lstrip('/') self.url = config.get(Config.SECTION_REMOTE_URL, storagepath) self.region = config.get(Config.SECTION_AWS_REGION, None) self.profile = config.get(Config.SECTION_AWS_PROFILE, 'default') @@ -90,7 +90,8 @@ 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_ETAG: self.get_etag(path_info['bucket'], + path_info['key'])} def _copy(self, from_info, to_info, s3=None): s3 = s3 if s3 else self.s3 @@ -137,7 +138,8 @@ class RemoteS3(RemoteBase): 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] + 'key': posixpath.join(self.prefix, + md5[0:2], md5[2:])} for md5 in md5s] def _all_keys(self): s3 = self.s3 @@ -202,14 +204,22 @@ class RemoteS3(RemoteBase): cb = Callback(name, total) try: - s3.upload_file(from_info['path'], to_info['bucket'], to_info['key'], Callback=cb) + s3.upload_file(from_info['path'], + to_info['bucket'], + to_info['key'], + Callback=cb) except Exception as exc: - Logger.error("Failed to upload '{}'".format(from_info['path']), exc) + msg = "Failed to upload '{}'".format(from_info['path']) + Logger.error(msg, exc) continue progress.finish_target(name) - def download(self, from_infos, to_infos, no_progress_bar=False, names=None): + def download(self, + from_infos, + to_infos, + no_progress_bar=False, + names=None): names = self._verify_path_args(from_infos, to_infos, names) s3 = self.s3 @@ -225,9 +235,10 @@ class RemoteS3(RemoteBase): if to_info['scheme'] != 'local': raise NotImplementedError - Logger.debug("Downloading '{}/{}' to '{}'".format(from_info['bucket'], - from_info['key'], - to_info['path'])) + msg = "Downloading '{}/{}' to '{}'".format(from_info['bucket'], + from_info['key'], + to_info['path']) + Logger.debug(msg) tmp_file = self.tmp_file(to_info['path']) if not name: @@ -243,10 +254,14 @@ class RemoteS3(RemoteBase): self._makedirs(to_info['path']) try: - s3.download_file(from_info['bucket'], from_info['key'], tmp_file, Callback=cb) + s3.download_file(from_info['bucket'], + from_info['key'], + tmp_file, + Callback=cb) except Exception as exc: - Logger.error("Failed to download '{}/{}'".format(from_info['bucket'], - from_info['key']), exc) + msg = "Failed to download '{}/{}'".format(from_info['bucket'], + from_info['key']) + Logger.error(msg, exc) return os.rename(tmp_file, to_info['path']) @@ -262,14 +277,15 @@ class RemoteS3(RemoteBase): keys = self._all_keys() return [self._path_to_etag(key) for key in keys] - def gc(self, checksum_infos): - used_etags = [info[self.PARAM_ETAG] for info in checksum_infos['s3']] - used_etags += [info[RemoteLOCAL.PARAM_MD5] for info in checksum_infos['local']] + 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']] 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:]), + 'key': posixpath.join(self.prefix, + etag[0:2], etag[2:]), 'bucket': self.bucket} self.remove(path_info) diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index c64fd57e3..54f6f54b5 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -1,7 +1,5 @@ import os -import errno import getpass -import filecmp import posixpath try: @@ -10,7 +8,6 @@ except ImportError: paramiko = None from dvc.logger import Logger -from dvc.utils import copyfile, file_md5 from dvc.progress import progress from dvc.remote.base import RemoteBase from dvc.remote.local import RemoteLOCAL @@ -43,7 +40,7 @@ def create_cb(name): class RemoteSSH(RemoteBase): scheme = 'ssh' - #NOTE: temporarily only absolute paths are allowed + # NOTE: temporarily only absolute paths are allowed REGEX = r'^ssh://((?P.*)@)?(?P[^/]*):(?P/.*)$' REQUIRES = {'paramiko': paramiko} @@ -55,17 +52,20 @@ class RemoteSSH(RemoteBase): self.host = self.group('host') self.user = self.group('user') if not self.user: - self.user = config.get(Config.SECTION_REMOTE_USER, getpass.getuser()) + self.user = config.get(Config.SECTION_REMOTE_USER, + getpass.getuser()) self.prefix = self.group('path') def md5s_to_path_infos(self, md5s): return [{'scheme': 'ssh', 'host': self.host, 'user': self.user, - 'path': posixpath.join(self.prefix, md5[0:2], md5[2:])} for md5 in md5s] + 'path': posixpath.join(self.prefix, + md5[0:2], md5[2:])} for md5 in md5s] def ssh(self, host=None, user=None): - Logger.debug("Establishing ssh connection with '{}' as user '{}'".format(host, user)) + msg = "Establishing ssh connection with '{}' as user '{}'" + Logger.debug(msg.format(host, user)) ssh = paramiko.SSHClient() @@ -79,7 +79,8 @@ class RemoteSSH(RemoteBase): def exists(self, path_infos): ret = [] ssh = self.ssh(host=self.host, user=self.user) - stdout = self._exec(ssh, 'find {} -type f -follow -print'.format(self.prefix)) + cmd = 'find {} -type f -follow -print'.format(self.prefix) + stdout = self._exec(ssh, cmd) plist = stdout.split() ssh.close() @@ -94,7 +95,8 @@ class RemoteSSH(RemoteBase): def _exec(self, ssh, cmd): stdin, stdout, stderr = ssh.exec_command(cmd) if stdout.channel.recv_exit_status() != 0: - DvcException('SSH command \'{}\' failed: {}'.format(cmd, stderr.read())) + DvcException('SSH command \'{}\' failed: {}'.format(cmd, + stderr.read())) return stdout.read().decode('utf-8') def md5(self, path_info): @@ -112,7 +114,8 @@ class RemoteSSH(RemoteBase): md5cmd = 'md5sum' index = 0 else: - raise DvcException('\'{}\' is not supported as a remote'.format(stdout)) + msg = '\'{}\' is not supported as a remote'.format(stdout) + raise DvcException(msg) stdout = self._exec(ssh, '{} {}'.format(md5cmd, path_info['path'])) md5 = stdout.split()[index] @@ -129,7 +132,8 @@ class RemoteSSH(RemoteBase): assert from_info['host'] == to_info['host'] assert from_info['user'] == to_info['user'] - s = ssh if ssh else self.ssh(host=from_info['host'], user=from_info['user']) + s = ssh if ssh else self.ssh(host=from_info['host'], + user=from_info['user']) dname = posixpath.dirname(to_info['path']) self._exec(s, 'mkdir -p {}'.format(dname)) @@ -181,7 +185,11 @@ class RemoteSSH(RemoteBase): ssh.open_sftp().remove(path_info['path']) ssh.close() - def download(self, from_infos, to_infos, no_progress_bar=False, names=None): + def download(self, + from_infos, + to_infos, + no_progress_bar=False, + names=None): names = self._verify_path_args(from_infos, to_infos, names) ssh = self.ssh(host=from_infos[0]['host'], user=from_infos[0]['user']) @@ -199,20 +207,25 @@ class RemoteSSH(RemoteBase): if to_info['scheme'] != 'local': raise NotImplementedError - Logger.debug("Downloading '{}/{}' to '{}'".format(from_info['host'], - from_info['path'], - to_info['path'])) + msg = "Downloading '{}/{}' to '{}'".format(from_info['host'], + from_info['path'], + to_info['path']) + Logger.debug(msg) + if not name: name = os.path.basename(to_info['path']) self._makedirs(to_info['path']) tmp_file = self.tmp_file(to_info['path']) try: - ssh.open_sftp().get(from_info['path'], tmp_file, callback=create_cb(name)) + ssh.open_sftp().get(from_info['path'], + tmp_file, + callback=create_cb(name)) except Exception as exc: - Logger.error("Failed to download '{}/{}' to '{}'".format(from_info['host'], - from_info['path'], - to_info['path']), exc) + msg = "Failed to download '{}/{}' to '{}'" + Logger.error(msg.format(from_info['host'], + from_info['path'], + to_info['path']), exc) continue os.rename(tmp_file, to_info['path']) @@ -244,11 +257,14 @@ class RemoteSSH(RemoteBase): self._exec(ssh, 'mkdir -p {}'.format(dname)) try: - sftp.put(from_info['path'], to_info['path'], callback=create_cb(name)) + sftp.put(from_info['path'], + to_info['path'], + callback=create_cb(name)) except Exception as exc: - Logger.error("Failed to upload '{}' to '{}/{}'".format(from_info['path'], - to_info['host'], - to_info['path']), exc) + msg = "Failed to upload '{}' to '{}/{}'" + Logger.error(msg.format(from_info['path'], + to_info['host'], + to_info['path'], exc)) continue progress.finish_target(name) @@ -262,21 +278,23 @@ class RemoteSSH(RemoteBase): def _all_md5s(self): ssh = self.ssh(host=self.host, user=self.user) - stdout = self._exec(ssh, 'find {} -type f -follow -print'.format(self.prefix)) + cmd = 'find {} -type f -follow -print'.format(self.prefix) + stdout = self._exec(ssh, cmd) flist = stdout.split() ssh.close() return [self._path_to_md5(path) for path in flist] - def gc(self, checksum_infos): - used_md5s = [info[self.PARAM_MD5] for info in checksum_infos['ssh']] - used_md5s += [info[RemoteLOCAL.PARAM_MD5] for info in checksum_infos['local']] + def gc(self, cinfos): + used = [info[self.PARAM_MD5] for info in cinfos['ssh']] + used += [info[RemoteLOCAL.PARAM_MD5] for info in cinfos['local']] for md5 in self._all_md5s(): - if md5 in used_md5s: + if md5 in used: continue path_info = {'scheme': 'ssh', 'user': self.user, 'host': self.host, - 'path': posixpath.join(self.prefix, md5[0:2], md5[2:])} + 'path': posixpath.join(self.prefix, + md5[0:2], md5[2:])} self.remove(path_info) diff --git a/dvc/scm.py b/dvc/scm.py index 7b68c1b64..19779b77b 100644 --- a/dvc/scm.py +++ b/dvc/scm.py @@ -113,7 +113,8 @@ class Git(Base): ignore_list = [] if os.path.exists(gitignore): ignore_list = open(gitignore, 'r').readlines() - filtered = list(filter(lambda x: x.strip() == entry.strip(), ignore_list)) + filtered = list(filter(lambda x: x.strip() == entry.strip(), + ignore_list)) if len(filtered) != 0: return @@ -143,8 +144,10 @@ class Git(Base): try: self.repo.index.add(paths) except AssertionError as exc: - msg = 'Failed to add \'{}\' to git. You can add those files manually using \'git add\'. ' - msg += 'See \'https://github.com/iterative/dvc/issues/610\' for more details.' + msg = 'Failed to add \'{}\' to git. You can add those files ' + msg += 'manually using \'git add\'. ' + msg += 'See \'https://github.com/iterative/dvc/issues/610\' ' + msg += 'for more details.' Logger.error(msg.format(str(paths)), exc) def commit(self, msg): @@ -173,9 +176,13 @@ class Git(Base): return [h.name for h in self.repo.heads] def install(self): - hook = os.path.join(self.root_dir, self.GIT_DIR, 'hooks', 'post-checkout') + hook = os.path.join(self.root_dir, + self.GIT_DIR, + 'hooks', + 'post-checkout') if os.path.isfile(hook): - raise SCMError('Git hook \'{}\' already exists.'.format(os.path.relpath(hook))) + msg = 'Git hook \'{}\' already exists.' + raise SCMError(msg.format(os.path.relpath(hook))) with open(hook, 'w+') as fd: fd.write('#!/bin/sh\nexec dvc checkout\n') os.chmod(hook, 0o777) diff --git a/dvc/stage.py b/dvc/stage.py index a4a207a59..139cc5379 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -2,7 +2,7 @@ import os import yaml import itertools import subprocess -import schema +from schema import Schema, SchemaError, Optional, Or, And import posixpath import dvc.dependency as dependency @@ -46,14 +46,22 @@ class Stage(object): PARAM_LOCKED = 'locked' SCHEMA = { - schema.Optional(PARAM_MD5): schema.Or(str, None), - schema.Optional(PARAM_CMD): schema.Or(str, None), - schema.Optional(PARAM_DEPS): schema.Or(schema.And(list, schema.Schema([dependency.SCHEMA])), None), - schema.Optional(PARAM_OUTS): schema.Or(schema.And(list, schema.Schema([output.SCHEMA])), None), - schema.Optional(PARAM_LOCKED): bool, + Optional(PARAM_MD5): Or(str, None), + Optional(PARAM_CMD): Or(str, None), + Optional(PARAM_DEPS): Or(And(list, Schema([dependency.SCHEMA])), None), + Optional(PARAM_OUTS): Or(And(list, Schema([output.SCHEMA])), None), + Optional(PARAM_LOCKED): bool, } - def __init__(self, project, path=None, cmd=None, cwd=os.curdir, deps=[], outs=[], md5=None, locked=False): + def __init__(self, + project, + path=None, + cmd=None, + cwd=os.curdir, + deps=[], + outs=[], + md5=None, + locked=False): self.project = project self.path = path self.cmd = cmd @@ -76,7 +84,8 @@ class Stage(object): if not os.path.isfile(path): return False - if not path.endswith(Stage.STAGE_FILE_SUFFIX) and os.path.basename(path) != Stage.STAGE_FILE: + if not path.endswith(Stage.STAGE_FILE_SUFFIX) \ + and os.path.basename(path) != Stage.STAGE_FILE: return False return True @@ -85,7 +94,7 @@ class Stage(object): md5 = self.dumpd().get(self.PARAM_MD5, None) # backward compatibility - if self.md5 == None: + if self.md5 is None: return False if self.md5 and md5 and self.md5 == md5: @@ -124,9 +133,11 @@ class Stage(object): ret = True if ret: - self.project.logger.debug(u'Dvc file \'{}\' changed'.format(self.relpath)) + msg = u'Dvc file \'{}\' changed'.format(self.relpath) else: - self.project.logger.debug(u'Dvc file \'{}\' didn\'t change'.format(self.relpath)) + msg = u'Dvc file \'{}\' didn\'t change'.format(self.relpath) + + self.project.logger.debug(msg) return ret @@ -150,15 +161,16 @@ class Stage(object): self.run() - self.project.logger.debug(u'\'{}\' was reproduced'.format(self.relpath)) + msg = u'\'{}\' was reproduced'.format(self.relpath) + self.project.logger.debug(msg) return self @staticmethod def validate(d): try: - schema.Schema(Stage.SCHEMA).validate(d) - except schema.SchemaError as exc: + Schema(Stage.SCHEMA).validate(d) + except SchemaError as exc: Logger.debug(str(exc)) raise StageFileFormatError() @@ -198,7 +210,9 @@ class Stage(object): else: path = posixpath - fname = fname if fname else path.basename(out.path) + cls.STAGE_FILE_SUFFIX + if not fname: + fname = path.basename(out.path) + cls.STAGE_FILE_SUFFIX + cwd = path.dirname(out.path) if not cwd or add else cwd return (fname, cwd) @@ -221,7 +235,8 @@ class Stage(object): stage.outs = output.loads_from(stage, outs, use_cache=True) stage.outs += output.loads_from(stage, outs_no_cache, use_cache=False) - stage.outs += output.loads_from(stage, metrics_no_cache, use_cache=False, metric=True) + stage.outs += output.loads_from(stage, metrics_no_cache, + use_cache=False, metric=True) stage.deps = dependency.loads_from(stage, deps) fname, cwd = Stage._stage_fname_cwd(fname, cwd, stage.outs, add=add) @@ -244,7 +259,7 @@ class Stage(object): outs = [x.dumpd() for x in self.outs] ret = {} - if self.cmd != None: + if self.cmd is not None: ret[Stage.PARAM_CMD] = self.cmd if len(deps): @@ -276,18 +291,22 @@ class Stage(object): def run(self): if self.locked: - self.project.logger.info(u'Verifying outputs in locked stage \'{}\''.format(self.relpath)) + msg = u'Verifying outputs in locked stage \'{}\'' + self.project.logger.info(msg.format(self.relpath)) self.check_missing_outputs() elif self.is_import: msg = u'Importing \'{}\' -> \'{}\'' - self.project.logger.info(msg.format(self.deps[0].path, self.outs[0].path)) + self.project.logger.info(msg.format(self.deps[0].path, + self.outs[0].path)) self.deps[0].download(self.outs[0].path_info) elif self.is_data_source: - self.project.logger.info(u'Verifying data sources in \'{}\''.format(self.relpath)) + msg = u'Verifying data sources in \'{}\''.format(self.relpath) + self.project.logger.info(msg) self.check_missing_outputs() else: - self.project.logger.info(u'Running command:\n\t{}'.format(self.cmd)) + msg = u'Running command:\n\t{}'.format(self.cmd) + self.project.logger.info(msg) p = subprocess.Popen(self.cmd, cwd=self.cwd, @@ -296,13 +315,14 @@ class Stage(object): executable=os.getenv('SHELL')) p.communicate() if p.returncode != 0: - raise StageCmdFailedError(self) + raise StageCmdFailedError(self) self.save() def check_missing_outputs(self): outs = [out for out in self.outs if not out.exists] - paths = [out.path if out.path_info['scheme'] != 'local' else out.rel_path for out in outs] + paths = [out.path if out.path_info['scheme'] != 'local' else + out.rel_path for out in outs] if paths: raise MissingDataSource(paths) diff --git a/dvc/state.py b/dvc/state.py index 689740efe..1e668d031 100644 --- a/dvc/state.py +++ b/dvc/state.py @@ -82,7 +82,8 @@ class State(object): try: return json.load(fd) except ValueError as exc: - Logger.error('Failed to load \'{}\''.format(self.state_file), exc) + Logger.error('Failed to load \'{}\''.format(self.state_file), + exc) return {} def dump(self): diff --git a/dvc/system.py b/dvc/system.py index 666731f6b..0bc454e94 100644 --- a/dvc/system.py +++ b/dvc/system.py @@ -4,7 +4,6 @@ import reflink if os.name == 'nt': import ntfsutils.hardlink as winlink - from ntfsutils.fs import getdirinfo class System(object): @@ -44,11 +43,18 @@ class System(object): # https://github.com/sunshowers/ntfs/pull/11 @staticmethod def getdirinfo(path): - from ntfsutils.fs import FILE_FLAG_BACKUP_SEMANTICS, FILE_SHARE_READ, OPEN_EXISTING - from ntfsutils.fs import CreateFile, WinError, BY_HANDLE_FILE_INFORMATION + from ntfsutils.fs import FILE_FLAG_BACKUP_SEMANTICS, FILE_SHARE_READ + from ntfsutils.fs import OPEN_EXISTING, BY_HANDLE_FILE_INFORMATION + from ntfsutils.fs import CreateFile, WinError from ntfsutils.fs import GetFileInformationByHandle, CloseHandle flags = FILE_FLAG_BACKUP_SEMANTICS - hfile = CreateFile(path, 0, FILE_SHARE_READ, None, OPEN_EXISTING, flags, None) + hfile = CreateFile(path, + 0, + FILE_SHARE_READ, + None, + OPEN_EXISTING, + flags, + None) if hfile is None: raise WinError() info = BY_HANDLE_FILE_INFORMATION() @@ -65,4 +71,6 @@ class System(object): # getdirinfo from ntfsutils works on both files and dirs info = System.getdirinfo(path) - return hash((info.dwVolumeSerialNumber, info.nFileIndexHigh, info.nFileIndexLow)) + return hash((info.dwVolumeSerialNumber, + info.nFileIndexHigh, + info.nFileIndexLow)) diff --git a/dvc/updater.py b/dvc/updater.py index 10f859d7e..fb6585458 100644 --- a/dvc/updater.py +++ b/dvc/updater.py @@ -5,10 +5,12 @@ import requests from dvc import VERSION_BASE from dvc.logger import Logger + class Updater(object): - URL = 'https://4ki8820rsf.execute-api.us-east-2.amazonaws.com/prod/latest-version' + URL = 'https://4ki8820rsf.execute-api.us-east-2.amazonaws.com/' \ + 'prod/latest-version' UPDATER_FILE = 'updater' - TIMEOUT = 7 * 24 * 60 * 60 #every week + TIMEOUT = 7 * 24 * 60 * 60 # every week TIMEOUT_GET = 10 def __init__(self, dvc_dir): @@ -42,7 +44,8 @@ class Updater(object): latest = j['version'] open(self.updater_file, 'w+').close() except Exception as exc: - Logger.debug('Failed to obtain latest version: {}'.format(str(exc))) + msg = 'Failed to obtain latest version: {}'.format(str(exc)) + Logger.debug(msg) return l_major, l_minor, l_patch = [int(x) for x in latest.split('.')] @@ -51,7 +54,8 @@ class Updater(object): if l_major <= c_major and \ l_minor <= c_minor and \ l_patch <= c_patch: - return + return - msg = 'You are using dvc version {}, however version {} is available. Consider upgrading.' + msg = 'You are using dvc version {}, however version {} is ' \ + 'available. Consider upgrading.' Logger.warn(msg.format(current, latest)) diff --git a/dvc/utils.py b/dvc/utils.py index 51882d19a..0eb9c8ace 100644 --- a/dvc/utils.py +++ b/dvc/utils.py @@ -2,14 +2,10 @@ Helpers for other modules. """ import os -import re -import sys import math import json -import stat import shutil import hashlib -from multiprocessing.pool import ThreadPool from dvc.istextfile import istextfile from dvc.progress import progress From 66f1777d9c105f94140ef8e254d522fa6b893930 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Wed, 25 Jul 2018 17:28:13 +0300 Subject: [PATCH 3/3] test: ssh: don't try to ask for a password Signed-off-by: Ruslan Kuprieiev --- tests/test_data_cloud.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index 011213693..fecb09f81 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -78,7 +78,7 @@ def _should_test_ssh(): return False try: - check_output(['ssh', '127.0.0.1', 'ls']) + check_output(['ssh', '-o', 'BatchMode=yes', '127.0.0.1', 'ls']) except (CalledProcessError, IOError): return False