From ba8d83e9b191f721afa2b6c75202ed41679bcbb4 Mon Sep 17 00:00:00 2001 From: Ruslan Kuprieiev Date: Tue, 12 Feb 2019 05:02:15 +0200 Subject: [PATCH] dvc: refactor dvc project class 1) split Project's methods into separate files; 2) rename Project into Repo(similar to gitpython's Repo) Signed-off-by: Ruslan Kuprieiev --- dvc/analytics.py | 16 +- dvc/cache.py | 25 +- dvc/command/add.py | 2 +- dvc/command/base.py | 8 +- dvc/command/checkout.py | 4 +- dvc/command/config.py | 8 +- dvc/command/daemon.py | 6 +- dvc/command/data_sync.py | 8 +- dvc/command/destroy.py | 4 +- dvc/command/gc.py | 15 +- dvc/command/imp.py | 2 +- dvc/command/init.py | 7 +- dvc/command/install.py | 2 +- dvc/command/lock.py | 2 +- dvc/command/metrics.py | 8 +- dvc/command/move.py | 2 +- dvc/command/pipeline.py | 14 +- dvc/command/remote.py | 2 +- dvc/command/remove.py | 2 +- dvc/command/repro.py | 8 +- dvc/command/root.py | 4 +- dvc/command/run.py | 4 +- dvc/command/status.py | 2 +- dvc/command/unprotect.py | 2 +- dvc/config.py | 12 +- dvc/data_cloud.py | 20 +- dvc/dependency/__init__.py | 4 +- dvc/exceptions.py | 18 +- dvc/lock.py | 8 +- dvc/logger.py | 2 +- dvc/main.py | 4 +- dvc/output/__init__.py | 4 +- dvc/output/base.py | 20 +- dvc/output/local.py | 8 +- dvc/progress.py | 2 +- dvc/project.py | 1395 ---------------------------------- dvc/remote/__init__.py | 4 +- dvc/remote/azure.py | 4 +- dvc/remote/base.py | 4 +- dvc/remote/gs.py | 4 +- dvc/remote/hdfs.py | 4 +- dvc/remote/http.py | 4 +- dvc/remote/local.py | 6 +- dvc/remote/s3.py | 4 +- dvc/remote/ssh.py | 4 +- dvc/repo/__init__.py | 410 ++++++++++ dvc/repo/add.py | 43 ++ dvc/repo/checkout.py | 39 + dvc/repo/destroy.py | 8 + dvc/repo/fetch.py | 28 + dvc/repo/gc.py | 111 +++ dvc/repo/imp.py | 20 + dvc/repo/init.py | 88 +++ dvc/repo/install.py | 2 + dvc/repo/lock.py | 8 + dvc/repo/metrics/__init__.py | 23 + dvc/repo/metrics/add.py | 7 + dvc/repo/metrics/modify.py | 64 ++ dvc/repo/metrics/remove.py | 5 + dvc/repo/metrics/show.py | 150 ++++ dvc/repo/move.py | 75 ++ dvc/repo/pull.py | 28 + dvc/repo/push.py | 28 + dvc/repo/remove.py | 10 + dvc/repo/reproduce.py | 134 ++++ dvc/repo/run.py | 62 ++ dvc/repo/status.py | 89 +++ dvc/repo/unprotect.py | 48 ++ dvc/scm/__init__.py | 13 +- dvc/scm/base.py | 12 +- dvc/scm/git.py | 42 +- dvc/stage.py | 28 +- dvc/state.py | 17 +- tests/basic_env.py | 4 +- tests/test_add.py | 4 +- tests/test_cache.py | 2 +- tests/test_checkout.py | 4 +- tests/test_cli.py | 4 +- tests/test_data_cloud.py | 2 +- tests/test_gc.py | 20 +- tests/test_init.py | 17 +- tests/test_metrics.py | 30 +- tests/test_move.py | 4 +- tests/test_pipeline.py | 2 +- tests/test_remote.py | 2 +- tests/test_repro.py | 20 +- tests/unit/remote/base.py | 4 +- 87 files changed, 1752 insertions(+), 1651 deletions(-) delete mode 100644 dvc/project.py create mode 100644 dvc/repo/__init__.py create mode 100644 dvc/repo/add.py create mode 100644 dvc/repo/checkout.py create mode 100644 dvc/repo/destroy.py create mode 100644 dvc/repo/fetch.py create mode 100644 dvc/repo/gc.py create mode 100644 dvc/repo/imp.py create mode 100644 dvc/repo/init.py create mode 100644 dvc/repo/install.py create mode 100644 dvc/repo/lock.py create mode 100644 dvc/repo/metrics/__init__.py create mode 100644 dvc/repo/metrics/add.py create mode 100644 dvc/repo/metrics/modify.py create mode 100644 dvc/repo/metrics/remove.py create mode 100644 dvc/repo/metrics/show.py create mode 100644 dvc/repo/move.py create mode 100644 dvc/repo/pull.py create mode 100644 dvc/repo/push.py create mode 100644 dvc/repo/remove.py create mode 100644 dvc/repo/reproduce.py create mode 100644 dvc/repo/run.py create mode 100644 dvc/repo/status.py create mode 100644 dvc/repo/unprotect.py diff --git a/dvc/analytics.py b/dvc/analytics.py index a95dcbab3..d458fa11d 100644 --- a/dvc/analytics.py +++ b/dvc/analytics.py @@ -161,8 +161,8 @@ class Analytics(object): """Collect analytics report.""" from dvc.scm import SCM from dvc.utils import is_binary - from dvc.project import Project - from dvc.exceptions import NotDvcProjectError + from dvc.repo import Repo + from dvc.exceptions import NotDvcRepoError self.info[self.PARAM_DVC_VERSION] = VERSION self.info[self.PARAM_IS_BINARY] = is_binary() @@ -171,9 +171,9 @@ class Analytics(object): self.info[self.PARAM_SYSTEM_INFO] = self._collect_system_info() try: - scm = SCM(root_dir=Project.find_root()) + scm = SCM(root_dir=Repo.find_root()) self.info[self.PARAM_SCM_CLASS] = type(scm).__name__ - except NotDvcProjectError: + except NotDvcRepoError: pass def collect_cmd(self, args, ret): @@ -204,8 +204,8 @@ class Analytics(object): @staticmethod def _is_enabled(cmd=None): from dvc.config import Config - from dvc.project import Project - from dvc.exceptions import NotDvcProjectError + from dvc.repo import Repo + from dvc.exceptions import NotDvcRepoError from dvc.command.daemon import CmdDaemonBase if os.getenv("DVC_TEST"): @@ -216,10 +216,10 @@ class Analytics(object): if cmd is None or not hasattr(cmd, "config"): try: - dvc_dir = Project.find_dvc_dir() + dvc_dir = Repo.find_dvc_dir() config = Config(dvc_dir) assert config is not None - except NotDvcProjectError: + except NotDvcRepoError: config = Config(validate=False) assert config is not None else: diff --git a/dvc/cache.py b/dvc/cache.py index f5945f317..35ee9a084 100644 --- a/dvc/cache.py +++ b/dvc/cache.py @@ -1,4 +1,4 @@ -"""Manages cache of a dvc project.""" +"""Manages cache of a dvc repo.""" from __future__ import unicode_literals @@ -8,42 +8,41 @@ from dvc.config import Config class Cache(object): - """Class that manages cache locations of a dvc project. + """Class that manages cache locations of a dvc repo. Args: - project (dvc.project.Project): project instance that this cache - belongs to. + repo (dvc.repo.Repo): repo instance that this cache belongs to. """ CACHE_DIR = "cache" - def __init__(self, project): + def __init__(self, repo): from dvc.remote import Remote - self.project = project + self.repo = repo - config = project.config.config[Config.SECTION_CACHE] + config = repo.config.config[Config.SECTION_CACHE] local = config.get(Config.SECTION_CACHE_LOCAL) if local: name = Config.SECTION_REMOTE_FMT.format(local) - sect = project.config.config[name] + sect = repo.config.config[name] else: - default_cache_dir = os.path.join(project.dvc_dir, self.CACHE_DIR) + default_cache_dir = os.path.join(repo.dvc_dir, self.CACHE_DIR) cache_dir = config.get(Config.SECTION_CACHE_DIR, default_cache_dir) cache_type = config.get(Config.SECTION_CACHE_TYPE) protected = config.get(Config.SECTION_CACHE_PROTECTED) sect = { Config.PRIVATE_CWD: config.get( - Config.PRIVATE_CWD, project.dvc_dir + Config.PRIVATE_CWD, repo.dvc_dir ), Config.SECTION_REMOTE_URL: cache_dir, Config.SECTION_CACHE_TYPE: cache_type, Config.SECTION_CACHE_PROTECTED: protected, } - self._local = Remote(project, sect) + self._local = Remote(repo, sect) self._s3 = self._get_remote(config, Config.SECTION_CACHE_S3) self._gs = self._get_remote(config, Config.SECTION_CACHE_GS) @@ -89,5 +88,5 @@ class Cache(object): return None name = Config.SECTION_REMOTE_FMT.format(remote) - sect = self.project.config.config[name] - return Remote(self.project, sect) + sect = self.repo.config.config[name] + return Remote(self.repo, sect) diff --git a/dvc/command/add.py b/dvc/command/add.py index bfb9c45e5..79c523ace 100644 --- a/dvc/command/add.py +++ b/dvc/command/add.py @@ -9,7 +9,7 @@ class CmdAdd(CmdBase): def run(self): for target in self.args.targets: try: - self.project.add(target, recursive=self.args.recursive) + self.repo.add(target, recursive=self.args.recursive) except DvcException: logger.error("failed to add file") return 1 diff --git a/dvc/command/base.py b/dvc/command/base.py index 225d030a5..53177ea94 100644 --- a/dvc/command/base.py +++ b/dvc/command/base.py @@ -20,10 +20,10 @@ def fix_subparsers(subparsers): class CmdBase(object): def __init__(self, args): - from dvc.project import Project + from dvc.repo import Repo - self.project = Project() - self.config = self.project.config + self.repo = Repo() + self.config = self.repo.config self.args = args self.set_loglevel(args) @@ -49,7 +49,7 @@ class CmdBase(object): from dvc.lock import LockError try: - with self.project.lock: + with self.repo.lock: return self.run() except LockError: logger.error("failed to lock before running a command") diff --git a/dvc/command/checkout.py b/dvc/command/checkout.py index 1b0b87221..49a3efd21 100644 --- a/dvc/command/checkout.py +++ b/dvc/command/checkout.py @@ -6,10 +6,10 @@ from dvc.command.base import CmdBase class CmdCheckout(CmdBase): def run(self): if not self.args.targets: - self.project.checkout(force=self.args.force) + self.repo.checkout(force=self.args.force) else: for target in self.args.targets: - self.project.checkout( + self.repo.checkout( target=target, with_deps=self.args.with_deps, force=self.args.force, diff --git a/dvc/command/config.py b/dvc/command/config.py index b25087fa9..72ac8cc47 100644 --- a/dvc/command/config.py +++ b/dvc/command/config.py @@ -11,14 +11,14 @@ from dvc.exceptions import DvcException class CmdConfig(CmdBase): def __init__(self, args): - from dvc.project import Project, NotDvcProjectError + from dvc.repo import Repo, NotDvcRepoError self.args = args try: - dvc_dir = os.path.join(Project.find_root(), Project.DVC_DIR) + dvc_dir = os.path.join(Repo.find_root(), Repo.DVC_DIR) saved_exc = None - except NotDvcProjectError as exc: + except NotDvcRepoError as exc: dvc_dir = None saved_exc = exc @@ -34,7 +34,7 @@ class CmdConfig(CmdBase): else: if dvc_dir is None: raise saved_exc - self.configobj = self.config._project_config + self.configobj = self.config._repo_config def run_cmd(self): return self.run() diff --git a/dvc/command/daemon.py b/dvc/command/daemon.py index 43b89c96e..a4acf6ce0 100644 --- a/dvc/command/daemon.py +++ b/dvc/command/daemon.py @@ -16,11 +16,11 @@ class CmdDaemonBase(CmdBase): class CmdDaemonUpdater(CmdDaemonBase): def run(self): import os - from dvc.project import Project + from dvc.repo import Repo from dvc.updater import Updater - root_dir = Project.find_root() - dvc_dir = os.path.join(root_dir, Project.DVC_DIR) + root_dir = Repo.find_root() + dvc_dir = os.path.join(root_dir, Repo.DVC_DIR) updater = Updater(dvc_dir) updater.fetch(detach=False) diff --git a/dvc/command/data_sync.py b/dvc/command/data_sync.py index f0b6a7452..d52006d09 100644 --- a/dvc/command/data_sync.py +++ b/dvc/command/data_sync.py @@ -24,7 +24,7 @@ class CmdDataBase(CmdBase): class CmdDataPull(CmdDataBase): def do_run(self, target=None): try: - self.project.pull( + self.repo.pull( target=target, jobs=self.args.jobs, remote=self.args.remote, @@ -44,7 +44,7 @@ class CmdDataPull(CmdDataBase): class CmdDataPush(CmdDataBase): def do_run(self, target=None): try: - self.project.push( + self.repo.push( target=target, jobs=self.args.jobs, remote=self.args.remote, @@ -63,7 +63,7 @@ class CmdDataPush(CmdDataBase): class CmdDataFetch(CmdDataBase): def do_run(self, target=None): try: - self.project.fetch( + self.repo.fetch( target=target, jobs=self.args.jobs, remote=self.args.remote, @@ -240,7 +240,7 @@ def add_parser(subparsers, _parent_parser): fetch_parser.set_defaults(func=CmdDataFetch) # Status - STATUS_HELP = "Show the project status." + STATUS_HELP = "Show the repo status." status_parser = subparsers.add_parser( "status", parents=[shared_parent_parser()], diff --git a/dvc/command/destroy.py b/dvc/command/destroy.py index 7a95e922d..3ef49f664 100644 --- a/dvc/command/destroy.py +++ b/dvc/command/destroy.py @@ -22,7 +22,7 @@ class CmdDestroy(CmdBase): " Use '-f' to force." ) - self.project.destroy() + self.repo.destroy() except Exception: logger.error("failed to destroy DVC") return 1 @@ -31,7 +31,7 @@ class CmdDestroy(CmdBase): def add_parser(subparsers, parent_parser): DESTROY_HELP = ( - "Destroy dvc. Will remove all project's information, " + "Destroy dvc. Will remove all repo's information, " "data files and cache." ) destroy_parser = subparsers.add_parser( diff --git a/dvc/command/gc.py b/dvc/command/gc.py index 7112d8f98..1b42e0f10 100644 --- a/dvc/command/gc.py +++ b/dvc/command/gc.py @@ -19,13 +19,13 @@ class CmdGC(CmdBase): else: msg += "all git branches and all git tags" - if self.args.projects is not None and len(self.args.projects) > 0: - msg += " of the current and the following projects:" + if self.args.repos is not None and len(self.args.repos) > 0: + msg += " of the current and the following repos:" - for project_path in self.args.projects: - msg += "\n - %s" % os.path.abspath(project_path) + for repo_path in self.args.repos: + msg += "\n - %s" % os.path.abspath(repo_path) else: - msg += " of the current project." + msg += " of the current repo." logger.warning(msg) @@ -33,14 +33,14 @@ class CmdGC(CmdBase): if not self.args.force and not prompt.confirm(msg): return 1 - self.project.gc( + self.repo.gc( all_branches=self.args.all_branches, all_tags=self.args.all_tags, cloud=self.args.cloud, remote=self.args.remote, force=self.args.force, jobs=self.args.jobs, - projects=self.args.projects, + repos=self.args.repos, ) return 0 @@ -91,6 +91,7 @@ def add_parser(subparsers, parent_parser): gc_parser.add_argument( "-p", "--projects", + dest="repos", type=str, nargs="*", default=None, diff --git a/dvc/command/imp.py b/dvc/command/imp.py index a0292bb05..82b96b22c 100644 --- a/dvc/command/imp.py +++ b/dvc/command/imp.py @@ -15,7 +15,7 @@ class CmdImport(CmdBase): out = self.args.out or default_out - self.project.imp(self.args.url, out, self.args.resume) + self.repo.imp(self.args.url, out, self.args.resume) except DvcException: logger.error( "failed to import {}. You could also try downloading " diff --git a/dvc/command/init.py b/dvc/command/init.py index bb311549b..3864c6f6d 100644 --- a/dvc/command/init.py +++ b/dvc/command/init.py @@ -8,13 +8,14 @@ class CmdInit(object): self.args = args def run_cmd(self): - from dvc.project import Project, InitError + from dvc.repo import Repo + from dvc.exceptions import InitError try: - self.project = Project.init( + self.repo = Repo.init( ".", no_scm=self.args.no_scm, force=self.args.force ) - self.config = self.project.config + self.config = self.repo.config except InitError: logger.error("failed to initiate dvc") return 1 diff --git a/dvc/command/install.py b/dvc/command/install.py index 818ba5ea7..f587b734d 100644 --- a/dvc/command/install.py +++ b/dvc/command/install.py @@ -7,7 +7,7 @@ from dvc.command.base import CmdBase class CmdInstall(CmdBase): def run_cmd(self): try: - self.project.install() + self.repo.install() except Exception: logger.error("failed to install dvc hooks") return 1 diff --git a/dvc/command/lock.py b/dvc/command/lock.py index 79742e8d7..79f105a1f 100644 --- a/dvc/command/lock.py +++ b/dvc/command/lock.py @@ -9,7 +9,7 @@ class CmdLockBase(CmdBase): def run(self, unlock=False): for target in self.args.targets: try: - self.project.lock_stage(target, unlock=unlock) + self.repo.lock_stage(target, unlock=unlock) except DvcException: logger.error( "failed to {}lock '{}'".format( diff --git a/dvc/command/metrics.py b/dvc/command/metrics.py index 53baa7214..522195dab 100644 --- a/dvc/command/metrics.py +++ b/dvc/command/metrics.py @@ -28,7 +28,7 @@ class CmdMetricsShow(CmdBase): typ = self.args.type xpath = self.args.xpath - self.project.metrics_show( + self.repo.metrics.show( self.args.path, typ=typ, xpath=xpath, @@ -46,7 +46,7 @@ class CmdMetricsShow(CmdBase): class CmdMetricsModify(CmdBase): def run(self): try: - self.project.metrics_modify( + self.repo.metrics.modify( self.args.path, typ=self.args.type, xpath=self.args.xpath ) except DvcException: @@ -59,7 +59,7 @@ class CmdMetricsModify(CmdBase): class CmdMetricsAdd(CmdBase): def run(self): try: - self.project.metrics_add( + self.repo.metrics.add( self.args.path, self.args.type, self.args.xpath ) except DvcException: @@ -73,7 +73,7 @@ class CmdMetricsAdd(CmdBase): class CmdMetricsRemove(CmdBase): def run(self): try: - self.project.metrics_remove(self.args.path) + self.repo.metrics.remove(self.args.path) except DvcException: msg = "failed to remove metric file '{}'".format(self.args.path) logger.error(msg) diff --git a/dvc/command/move.py b/dvc/command/move.py index dca6f58a7..c44a0aba1 100644 --- a/dvc/command/move.py +++ b/dvc/command/move.py @@ -8,7 +8,7 @@ from dvc.command.base import CmdBase class CmdMove(CmdBase): def run(self): try: - self.project.move(self.args.src, self.args.dst) + self.repo.move(self.args.src, self.args.dst) except DvcException: msg = "failed to move '{}' -> '{}'".format( self.args.src, self.args.dst diff --git a/dvc/command/pipeline.py b/dvc/command/pipeline.py index eeaf2607f..e9e3e6b0e 100644 --- a/dvc/command/pipeline.py +++ b/dvc/command/pipeline.py @@ -14,10 +14,10 @@ class CmdPipelineShow(CmdBase): import networkx from dvc.stage import Stage - stage = Stage.load(self.project, target) - G = self.project.graph()[0] + stage = Stage.load(self.repo, target) + G = self.repo.graph()[0] stages = networkx.get_node_attributes(G, "stage") - node = os.path.relpath(stage.path, self.project.root_dir) + node = os.path.relpath(stage.path, self.repo.root_dir) for n in networkx.dfs_postorder_nodes(G, node): if commands: @@ -32,11 +32,11 @@ class CmdPipelineShow(CmdBase): import networkx from dvc.stage import Stage - stage = Stage.load(self.project, target) - node = os.path.relpath(stage.path, self.project.root_dir) + stage = Stage.load(self.repo, target) + node = os.path.relpath(stage.path, self.repo.root_dir) pipelines = list( - filter(lambda g: node in g.nodes(), self.project.pipelines()) + filter(lambda g: node in g.nodes(), self.repo.pipelines()) ) assert len(pipelines) == 1 @@ -124,7 +124,7 @@ class CmdPipelineList(CmdBase): def run(self): import networkx - pipelines = self.project.pipelines() + pipelines = self.repo.pipelines() for p in pipelines: stages = networkx.get_node_attributes(p, "stage") for stage in stages: diff --git a/dvc/command/remote.py b/dvc/command/remote.py index 67a7f2452..f108a1383 100644 --- a/dvc/command/remote.py +++ b/dvc/command/remote.py @@ -76,7 +76,7 @@ class CmdRemoteRemove(CmdConfig): for configobj in [ self.config._local_config, - self.config._project_config, + self.config._repo_config, self.config._global_config, self.config._system_config, ]: diff --git a/dvc/command/remove.py b/dvc/command/remove.py index 67e17eb8f..51ba2e40d 100644 --- a/dvc/command/remove.py +++ b/dvc/command/remove.py @@ -30,7 +30,7 @@ class CmdRemove(CmdBase): for target in self.args.targets: try: outs_only = self._is_outs_only(target) - self.project.remove(target, outs_only=outs_only) + self.repo.remove(target, outs_only=outs_only) except DvcException: logger.error("failed to remove {}".format(target)) return 1 diff --git a/dvc/command/repro.py b/dvc/command/repro.py index 4aeea1299..59727c3d8 100644 --- a/dvc/command/repro.py +++ b/dvc/command/repro.py @@ -24,7 +24,7 @@ class CmdRepro(CmdBase): ret = 0 for target in self.args.targets: try: - stages = self.project.reproduce( + stages = self.repo.reproduce( target, recursive=recursive, force=self.args.force, @@ -39,7 +39,7 @@ class CmdRepro(CmdBase): logger.info(CmdDataStatus.UP_TO_DATE_MSG) if self.args.metrics: - self.project.metrics_show() + self.repo.metrics.show() except DvcException: logger.error() ret = 1 @@ -79,7 +79,7 @@ def add_parser(subparsers, parent_parser): "-c", "--cwd", default=os.path.curdir, - help="Directory within your project to reroduce from.", + help="Directory within your repo to reroduce from.", ) repro_parser.add_argument( "-m", @@ -115,7 +115,7 @@ def add_parser(subparsers, parent_parser): "--all-pipelines", action="store_true", default=False, - help="Reproduce all pipelines in the project.", + help="Reproduce all pipelines in the repo.", ) repro_parser.add_argument( "--ignore-build-cache", diff --git a/dvc/command/root.py b/dvc/command/root.py index 8710f141d..f43dce9e3 100644 --- a/dvc/command/root.py +++ b/dvc/command/root.py @@ -11,12 +11,12 @@ class CmdRoot(CmdBase): return self.run() def run(self): - logger.info(os.path.relpath(self.project.root_dir)) + logger.info(os.path.relpath(self.repo.root_dir)) return 0 def add_parser(subparsers, parent_parser): - ROOT_HELP = "Relative path to project's directory." + ROOT_HELP = "Relative path to repo's directory." root_parser = subparsers.add_parser( "root", parents=[parent_parser], description=ROOT_HELP, help=ROOT_HELP ) diff --git a/dvc/command/run.py b/dvc/command/run.py index c070f2708..cae569654 100644 --- a/dvc/command/run.py +++ b/dvc/command/run.py @@ -29,7 +29,7 @@ class CmdRun(CmdBase): return 1 try: - self.project.run( + self.repo.run( cmd=self._parsed_cmd(), outs=self.args.outs, outs_no_cache=self.args.outs_no_cache, @@ -130,7 +130,7 @@ def add_parser(subparsers, parent_parser): "-c", "--cwd", default=os.path.curdir, - help="Directory within your project to run your command and place " + help="Directory within your repo to run your command and place " "stage file in.", ) run_parser.add_argument( diff --git a/dvc/command/status.py b/dvc/command/status.py index 5419e7166..75bc4f2e4 100644 --- a/dvc/command/status.py +++ b/dvc/command/status.py @@ -28,7 +28,7 @@ class CmdDataStatus(CmdDataBase): def do_run(self, target=None): indent = 1 if self.args.cloud else 0 try: - st = self.project.status( + st = self.repo.status( target=target, jobs=self.args.jobs, cloud=self.args.cloud, diff --git a/dvc/command/unprotect.py b/dvc/command/unprotect.py index e603ad786..2cc9d931d 100644 --- a/dvc/command/unprotect.py +++ b/dvc/command/unprotect.py @@ -9,7 +9,7 @@ class CmdUnprotect(CmdBase): def run(self): for target in self.args.targets: try: - self.project.unprotect(target) + self.repo.unprotect(target) except DvcException: msg = "failed to unprotect '{}'".format(target) logger.error(msg) diff --git a/dvc/config.py b/dvc/config.py index 603ee5b78..e733f5d19 100644 --- a/dvc/config.py +++ b/dvc/config.py @@ -108,11 +108,11 @@ def is_percent(val): class Config(object): # pylint: disable=too-many-instance-attributes - """Class that manages configuration files for a dvc project. + """Class that manages configuration files for a dvc repo. Args: dvc_dir (str): optional path to `.dvc` directory, that is used to - access project-specific configs like .dvc/config and + access repo-specific configs like .dvc/config and .dvc/config.local. validate (bool): optional flag to tell dvc if it should validate the config or just load it as is. 'True' by default. @@ -295,7 +295,7 @@ class Config(object): # pylint: disable=too-many-instance-attributes self._system_config = None self._global_config = None - self._project_config = None + self._repo_config = None self._local_config = None self.config = None @@ -347,9 +347,9 @@ class Config(object): # pylint: disable=too-many-instance-attributes self._global_config = configobj.ConfigObj(self.global_config_file) if self.config_file is not None: - self._project_config = configobj.ConfigObj(self.config_file) + self._repo_config = configobj.ConfigObj(self.config_file) else: - self._project_config = configobj.ConfigObj() + self._repo_config = configobj.ConfigObj() if self.config_local_file is not None: self._local_config = configobj.ConfigObj(self.config_local_file) @@ -454,7 +454,7 @@ class Config(object): # pylint: disable=too-many-instance-attributes clist = [ self._system_config, self._global_config, - self._project_config, + self._repo_config, self._local_config, ] diff --git a/dvc/data_cloud.py b/dvc/data_cloud.py index bc2433a23..e78a0d916 100644 --- a/dvc/data_cloud.py +++ b/dvc/data_cloud.py @@ -19,9 +19,9 @@ class DataCloud(object): """Class that manages dvc remotes. Args: - project (dvc.project.Project): project instance that belongs to the - project that we are working on. - config (dict): config of the project. + repo (dvc.repo.Repo): repo instance that belongs to the repo that + we are working on. + config (dict): config of the repo. Raises: config.ConfigError: thrown when config has invalid format. @@ -38,8 +38,8 @@ class DataCloud(object): "https": RemoteHTTP, } - def __init__(self, project, config=None): - self.project = project + def __init__(self, repo, config=None): + self.repo = repo self._config = config self._core = self._config[Config.SECTION_CORE] @@ -64,7 +64,7 @@ class DataCloud(object): msg = "can't find remote section '{}' in config" raise ConfigError(msg.format(section)) - return Remote(self.project, cloud_config) + return Remote(self.repo, cloud_config) def _init_compat(self): name = self._core.get(Config.SECTION_CORE_CLOUD, "").strip().lower() @@ -94,7 +94,7 @@ class DataCloud(object): if global_storage_path: logger.warning("using obsoleted config format. Consider updating.") - cloud = cloud_type(self.project, cloud_config) + cloud = cloud_type(self.repo, cloud_config) return cloud def _get_cloud(self, remote, cmd): @@ -122,7 +122,7 @@ class DataCloud(object): show_checksums (bool): show checksums instead of file names in information messages. """ - return self.project.cache.local.push( + return self.repo.cache.local.push( targets, jobs=jobs, remote=self._get_cloud(remote, "push"), @@ -140,7 +140,7 @@ class DataCloud(object): show_checksums (bool): show checksums instead of file names in information messages. """ - return self.project.cache.local.pull( + return self.repo.cache.local.pull( targets, jobs=jobs, remote=self._get_cloud(remote, "pull"), @@ -160,6 +160,6 @@ class DataCloud(object): information messages. """ cloud = self._get_cloud(remote, "status") - return self.project.cache.local.status( + return self.repo.cache.local.status( targets, jobs=jobs, remote=cloud, show_checksums=show_checksums ) diff --git a/dvc/dependency/__init__.py b/dvc/dependency/__init__.py index 47e29dc10..0d372ce44 100644 --- a/dvc/dependency/__init__.py +++ b/dvc/dependency/__init__.py @@ -49,8 +49,8 @@ def _get(stage, p, info): parsed = urlparse(p) if parsed.scheme == "remote": name = Config.SECTION_REMOTE_FMT.format(parsed.netloc) - sect = stage.project.config.config[name] - remote = Remote(stage.project, sect) + sect = stage.repo.config.config[name] + remote = Remote(stage.repo, sect) return DEP_MAP[remote.scheme](stage, p, info, remote=remote) for d in DEPS: diff --git a/dvc/exceptions.py b/dvc/exceptions.py index a152355fd..74d41e93d 100644 --- a/dvc/exceptions.py +++ b/dvc/exceptions.py @@ -125,8 +125,8 @@ class MoveNotDataSourceError(DvcException): super(MoveNotDataSourceError, self).__init__(msg.format(path=path)) -class NotDvcProjectError(DvcException): - """Thrown if a directory is not a dvc project. +class NotDvcRepoError(DvcException): + """Thrown if a directory is not a dvc repo. Args: root (str): path to the directory. @@ -134,7 +134,7 @@ class NotDvcProjectError(DvcException): def __init__(self, root): msg = "not a dvc repository (checked up to mount point '{}')" - super(NotDvcProjectError, self).__init__(msg.format(root)) + super(NotDvcRepoError, self).__init__(msg.format(root)) class DvcParserError(DvcException): @@ -163,3 +163,15 @@ class ConfirmRemoveError(DvcException): "unable to remove '{}' without a confirmation from the user. Use " "'-f' to force.".format(path) ) + + +class InitError(DvcException): + def __init__(self, msg): + super(InitError, self).__init__(msg) + + +class ReproductionError(DvcException): + def __init__(self, dvc_file_name, ex): + self.path = dvc_file_name + msg = "failed to reproduce '{}'".format(dvc_file_name) + super(ReproductionError, self).__init__(msg, cause=ex) diff --git a/dvc/lock.py b/dvc/lock.py index 8bd2a0fd9..81356a070 100644 --- a/dvc/lock.py +++ b/dvc/lock.py @@ -10,11 +10,11 @@ from dvc.exceptions import DvcException class LockError(DvcException): - """Thrown when unable to acquire the lock for dvc project.""" + """Thrown when unable to acquire the lock for dvc repo.""" class Lock(object): - """Class for dvc project lock. + """Class for dvc repo lock. Args: dvc_dir (str): path to the directory that the lock should be created @@ -39,7 +39,7 @@ class Lock(object): ) def lock(self): - """Acquire lock for dvc project.""" + """Acquire lock for dvc repo.""" try: self._do_lock() return @@ -49,7 +49,7 @@ class Lock(object): self._do_lock() def unlock(self): - """Release lock for dvc project.""" + """Release lock for dvc repo.""" self._lock.close() self._lock = None diff --git a/dvc/logger.py b/dvc/logger.py index fcc0663fd..2836bcffa 100644 --- a/dvc/logger.py +++ b/dvc/logger.py @@ -1,4 +1,4 @@ -"""Manages logger for dvc project.""" +"""Manages logger for dvc repo.""" from __future__ import unicode_literals diff --git a/dvc/main.py b/dvc/main.py index 7ac37bcdb..18a952463 100644 --- a/dvc/main.py +++ b/dvc/main.py @@ -6,7 +6,7 @@ import dvc.logger as logger from dvc.cli import parse_args from dvc.command.base import CmdBase from dvc.analytics import Analytics -from dvc.exceptions import NotDvcProjectError, DvcParserError +from dvc.exceptions import NotDvcRepoError, DvcParserError def main(argv=None): @@ -33,7 +33,7 @@ def main(argv=None): except KeyboardInterrupt: logger.error("interrupted by the user") ret = 252 - except NotDvcProjectError: + except NotDvcRepoError: logger.error() ret = 253 except DvcParserError: diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index e17b425b6..ede71b562 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -55,8 +55,8 @@ def _get(stage, p, info, cache, metric): parsed = urlparse(p) if parsed.scheme == "remote": name = Config.SECTION_REMOTE_FMT.format(parsed.netloc) - sect = stage.project.config.config[name] - remote = Remote(stage.project, sect) + sect = stage.repo.config.config[name] + remote = Remote(stage.repo, sect) return OUTS_MAP[remote.scheme]( stage, p, info, cache=cache, remote=remote, metric=metric ) diff --git a/dvc/output/base.py b/dvc/output/base.py index 71c9e7e7a..abbcaf402 100644 --- a/dvc/output/base.py +++ b/dvc/output/base.py @@ -52,16 +52,16 @@ class OutputBase(object): self, stage, path, info=None, remote=None, cache=True, metric=False ): self.stage = stage - self.project = stage.project + self.repo = stage.repo self.url = path self.info = info - self.remote = remote or self.REMOTE(self.project, {}) + self.remote = remote or self.REMOTE(self.repo, {}) self.use_cache = False if self.IS_DEPENDENCY else cache self.metric = False if self.IS_DEPENDENCY else metric if ( self.use_cache - and getattr(self.project.cache, self.REMOTE.scheme) is None + and getattr(self.repo.cache, self.REMOTE.scheme) is None ): raise DvcException( "no cache location setup for '{}' outputs.".format( @@ -118,7 +118,7 @@ class OutputBase(object): if not self.use_cache: return self.info != self.remote.save_info(self.path_info) - return getattr(self.project.cache, self.scheme).changed( + return getattr(self.repo.cache, self.scheme).changed( self.path_info, self.info ) @@ -126,7 +126,7 @@ class OutputBase(object): if not self.use_cache or not self.checksum: return True - cache = self.project.cache.__getattribute__(self.scheme) + cache = self.repo.cache.__getattribute__(self.scheme) return cache.changed_cache(self.checksum) @@ -140,7 +140,7 @@ class OutputBase(object): if not self.use_cache: self.info = self.remote.save_info(self.path_info) else: - self.info = getattr(self.project.cache, self.scheme).save( + self.info = getattr(self.repo.cache, self.scheme).save( self.path_info ) @@ -171,7 +171,7 @@ class OutputBase(object): if not self.use_cache: return - getattr(self.project.cache, self.scheme).checkout( + getattr(self.repo.cache, self.scheme).checkout( self.path_info, self.info, force=force ) @@ -181,11 +181,11 @@ class OutputBase(object): return if ignore_remove and self.use_cache and self.is_local: - self.project.scm.ignore_remove(self.path) + self.repo.scm.ignore_remove(self.path) def move(self, out): if self.scheme == "local" and self.use_cache and self.is_local: - self.project.scm.ignore_remove(self.path) + self.repo.scm.ignore_remove(self.path) self.remote.move(self.path_info, out.path_info) self.url = out.url @@ -193,4 +193,4 @@ class OutputBase(object): self.save() if self.scheme == "local" and self.use_cache and self.is_local: - self.project.scm.ignore(self.path) + self.repo.scm.ignore(self.path) diff --git a/dvc/output/local.py b/dvc/output/local.py index 756325afd..1324e1377 100644 --- a/dvc/output/local.py +++ b/dvc/output/local.py @@ -52,7 +52,7 @@ class OutputLOCAL(OutputBase): @property def cache(self): - return self.project.cache.local.get(self.checksum) + return self.repo.cache.local.get(self.checksum) def dumpd(self): ret = super(OutputLOCAL, self).dumpd() @@ -111,10 +111,10 @@ class OutputLOCAL(OutputBase): return if self.is_local: - if self.project.scm.is_tracked(self.path): + if self.repo.scm.is_tracked(self.path): raise OutputAlreadyTrackedError(self.rel_path) if self.use_cache: - self.project.scm.ignore(self.path) + self.repo.scm.ignore(self.path) - self.info = self.project.cache.local.save(self.path_info) + self.info = self.repo.cache.local.save(self.path_info) diff --git a/dvc/progress.py b/dvc/progress.py index 3851bf8a8..7b6544238 100644 --- a/dvc/progress.py +++ b/dvc/progress.py @@ -1,4 +1,4 @@ -"""Manages progress bars for dvc project.""" +"""Manages progress bars for dvc repo.""" from __future__ import print_function from __future__ import unicode_literals diff --git a/dvc/project.py b/dvc/project.py deleted file mode 100644 index f808a31e3..000000000 --- a/dvc/project.py +++ /dev/null @@ -1,1395 +0,0 @@ -from __future__ import unicode_literals - -from dvc.utils.compat import str, builtin_str, open - -import collections -import os -import dvc.prompt as prompt -import dvc.logger as logger - -from dvc.exceptions import ( - DvcException, - MoveNotDataSourceError, - NotDvcProjectError, -) - - -class InitError(DvcException): - def __init__(self, msg): - super(InitError, self).__init__(msg) - - -class ReproductionError(DvcException): - def __init__(self, dvc_file_name, ex): - self.path = dvc_file_name - msg = "failed to reproduce '{}'".format(dvc_file_name) - super(ReproductionError, self).__init__(msg, cause=ex) - - -class Project(object): - DVC_DIR = ".dvc" - - def __init__(self, root_dir=None): - from dvc.config import Config - from dvc.state import State - from dvc.lock import Lock - from dvc.scm import SCM - from dvc.cache import Cache - from dvc.data_cloud import DataCloud - from dvc.updater import Updater - - root_dir = self.find_root(root_dir) - - self.root_dir = os.path.abspath(os.path.realpath(root_dir)) - self.dvc_dir = os.path.join(self.root_dir, self.DVC_DIR) - - self.config = Config(self.dvc_dir) - - self.scm = SCM(self.root_dir, project=self) - self.lock = Lock(self.dvc_dir) - # NOTE: storing state and link_state in the repository itself to avoid - # any possible state corruption in 'shared cache dir' scenario. - self.state = State(self, self.config.config) - - core = self.config.config[Config.SECTION_CORE] - - logger.set_level(core.get(Config.SECTION_CORE_LOGLEVEL)) - - self.cache = Cache(self) - self.cloud = DataCloud(self, config=self.config.config) - self.updater = Updater(self.dvc_dir) - - self.files_to_git_add = [] - - self._ignore() - - self.updater.check() - - def __repr__(self): - return "Project: '{root_dir}'".format(root_dir=self.root_dir) - - @staticmethod - def find_root(root=None): - if root is None: - root = os.getcwd() - else: - root = os.path.abspath(os.path.realpath(root)) - - while True: - dvc_dir = os.path.join(root, Project.DVC_DIR) - if os.path.isdir(dvc_dir): - return root - if os.path.ismount(root): - break - root = os.path.dirname(root) - raise NotDvcProjectError(root) - - @staticmethod - def find_dvc_dir(root=None): - root_dir = Project.find_root(root) - return os.path.join(root_dir, Project.DVC_DIR) - - def _remind_to_git_add(self): - if not self.files_to_git_add: - return - - logger.info( - "\n" - "To track the changes with git run:\n" - "\n" - "\tgit add {files}".format(files=" ".join(self.files_to_git_add)) - ) - - @staticmethod - def init(root_dir=os.curdir, no_scm=False, force=False): - """ - Creates an empty project on the given directory -- basically a - `.dvc` directory with subdirectories for configuration and cache. - - It should be tracked by a SCM or use the `--no-scm` flag. - - If the given directory is not empty, you must use the `--force` - flag to override it. - - Args: - root_dir: Path to project's root directory. - - Returns: - Project instance. - - Raises: - KeyError: Raises an exception. - """ - import shutil - from dvc.scm import SCM, Base - from dvc.config import Config - - root_dir = os.path.abspath(root_dir) - dvc_dir = os.path.join(root_dir, Project.DVC_DIR) - scm = SCM(root_dir) - if type(scm) == Base and not no_scm: - raise InitError( - "{project} is not tracked by any supported scm tool" - " (e.g. git). Use '--no-scm' if you don't want to use any scm.".format( - project=root_dir - ) - ) - - if os.path.isdir(dvc_dir): - if not force: - raise InitError( - "'{project}' exists. Use '-f' to force.".format( - project=os.path.relpath(dvc_dir) - ) - ) - - shutil.rmtree(dvc_dir) - - os.mkdir(dvc_dir) - - config = Config.init(dvc_dir) - proj = Project(root_dir) - - scm.add([config.config_file]) - - if scm.ignore_file: - scm.add([os.path.join(dvc_dir, scm.ignore_file)]) - logger.info("\nYou can now commit the changes to git.\n") - - proj._welcome_message() - - return proj - - def destroy(self): - import shutil - - for stage in self.stages(): - stage.remove() - - shutil.rmtree(self.dvc_dir) - - def _ignore(self): - flist = [ - self.state.state_file, - self.lock.lock_file, - self.config.config_local_file, - self.updater.updater_file, - self.updater.lock.lock_file, - ] + self.state.temp_files - - if self.cache.local.cache_dir.startswith(self.root_dir): - flist += [self.cache.local.cache_dir] - - self.scm.ignore_list(flist) - - def install(self): - self.scm.install() - - @staticmethod - def _check_cyclic_graph(graph): - import networkx as nx - from dvc.exceptions import CyclicGraphError - - cycles = list(nx.simple_cycles(graph)) - - if cycles: - raise CyclicGraphError(cycles[0]) - - def add(self, fname, recursive=False): - from dvc.stage import Stage - - fnames = [] - if recursive and os.path.isdir(fname): - for root, dirs, files in os.walk(str(fname)): - for f in files: - path = os.path.join(root, f) - if Stage.is_stage_file(path): - continue - if os.path.basename(path) == self.scm.ignore_file: - continue - if self.scm.is_tracked(path): - continue - fnames.append(path) - else: - fnames = [fname] - - stages = [] - self.files_to_git_add = [] - with self.state: - for f in fnames: - stage = Stage.create(project=self, outs=[f], add=True) - - if stage is None: - stages.append(stage) - continue - - stage.save() - stages.append(stage) - - self._check_dag(self.stages() + stages) - - for stage in stages: - if stage is not None: - stage.dump() - - self._remind_to_git_add() - - return stages - - def _check_dag(self, stages): - """Generate graph including the new stage to check for errors""" - self.graph(stages=stages) - - def remove(self, target, outs_only=False): - from dvc.stage import Stage - - stage = Stage.load(self, target) - if outs_only: - stage.remove_outs() - else: - stage.remove() - - return stage - - def lock_stage(self, target, unlock=False): - from dvc.stage import Stage - - stage = Stage.load(self, target) - stage.locked = False if unlock else True - stage.dump() - - return stage - - def move(self, from_path, to_path): - """ - Renames an output file and modifies the stage associated - to reflect the change on the pipeline. - - If the output has the same name as its stage, it would - also rename the corresponding stage file. - - E.g. - Having: (hello, hello.dvc) - - $ dvc move hello greetings - - Result: (greeting, greeting.dvc) - - It only works with outputs generated by `add` or `import`, - also known as data sources. - """ - import dvc.output as Output - from dvc.stage import Stage - - from_out = Output.loads_from(Stage(self, cwd=os.curdir), [from_path])[ - 0 - ] - - to_path = self._expand_target_path(from_path, to_path) - - try: - stage, out = next( - (stage, out) - for stage in self.stages() - for out in stage.outs - if from_out.path == out.path - ) - except StopIteration: - raise DvcException( - "unable to find stage file with output '{path}'".format( - path=from_path - ) - ) - - if not stage.is_data_source: - raise MoveNotDataSourceError(stage.relpath) - - stage_name = os.path.splitext(os.path.basename(stage.path))[0] - from_name = os.path.basename(from_out.path) - if stage_name == from_name: - os.unlink(stage.path) - - stage.path = os.path.join( - os.path.dirname(to_path), - os.path.basename(to_path) + Stage.STAGE_FILE_SUFFIX, - ) - - stage.cwd = os.path.join(self.root_dir, os.path.dirname(to_path)) - - to_out = Output.loads_from( - stage, [os.path.basename(to_path)], out.cache, out.metric - )[0] - - with self.state: - out.move(to_out) - - stage.dump() - - self._remind_to_git_add() - - def _unprotect_file(self, path): - import stat - import uuid - from dvc.system import System - from dvc.utils import copyfile, move, remove - - if System.is_symlink(path) or System.is_hardlink(path): - logger.debug("Unprotecting '{}'".format(path)) - - tmp = os.path.join(os.path.dirname(path), "." + str(uuid.uuid4())) - move(path, tmp) - - copyfile(tmp, path) - - remove(tmp) - else: - logger.debug( - "Skipping copying for '{}', since it is not " - "a symlink or a hardlink.".format(path) - ) - - os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE) - - def _unprotect_dir(self, path): - for root, dirs, files in os.walk(str(path)): - for f in files: - path = os.path.join(root, f) - self._unprotect_file(path) - - def unprotect(self, path): - if not os.path.exists(path): - raise DvcException( - "can't unprotect non-existing data '{}'".format(path) - ) - - if os.path.isdir(path): - self._unprotect_dir(path) - else: - self._unprotect_file(path) - - def run( - self, - cmd=None, - deps=None, - outs=None, - outs_no_cache=None, - metrics=None, - metrics_no_cache=None, - fname=None, - cwd=os.curdir, - no_exec=False, - overwrite=False, - ignore_build_cache=False, - remove_outs=False, - ): - from dvc.stage import Stage - - if outs is None: - outs = [] - if deps is None: - deps = [] - if outs_no_cache is None: - outs_no_cache = [] - if metrics is None: - metrics = [] - if metrics_no_cache is None: - metrics_no_cache = [] - - with self.state: - stage = Stage.create( - project=self, - fname=fname, - cmd=cmd, - cwd=cwd, - outs=outs, - outs_no_cache=outs_no_cache, - metrics=metrics, - metrics_no_cache=metrics_no_cache, - deps=deps, - overwrite=overwrite, - ignore_build_cache=ignore_build_cache, - remove_outs=remove_outs, - ) - - if stage is None: - return None - - self._check_dag(self.stages() + [stage]) - - self.files_to_git_add = [] - with self.state: - if not no_exec: - stage.run() - - stage.dump() - - self._remind_to_git_add() - - return stage - - def imp(self, url, out, resume=False): - from dvc.stage import Stage - - stage = Stage.create(project=self, cmd=None, deps=[url], outs=[out]) - - if stage is None: - return None - - self._check_dag(self.stages() + [stage]) - - self.files_to_git_add = [] - with self.state: - stage.run(resume=resume) - - stage.dump() - - self._remind_to_git_add() - - return stage - - def _reproduce_stage(self, stages, node, force, dry, interactive): - stage = stages[node] - - if stage.locked: - logger.warning( - "DVC file '{path}' is locked. Its dependencies are" - " not going to be reproduced.".format(path=stage.relpath) - ) - - stage = stage.reproduce(force=force, dry=dry, interactive=interactive) - if not stage: - return [] - - if not dry: - stage.dump() - - return [stage] - - def reproduce( - self, - target=None, - recursive=True, - force=False, - dry=False, - interactive=False, - pipeline=False, - all_pipelines=False, - ignore_build_cache=False, - ): - from dvc.stage import Stage - - if not target and not all_pipelines: - raise ValueError() - - if not interactive: - config = self.config - core = config.config[config.SECTION_CORE] - interactive = core.get(config.SECTION_CORE_INTERACTIVE, False) - - targets = [] - if pipeline or all_pipelines: - if pipeline: - stage = Stage.load(self, target) - node = os.path.relpath(stage.path, self.root_dir) - pipelines = [self._get_pipeline(node)] - else: - pipelines = self.pipelines() - - for G in pipelines: - for node in G.nodes(): - if G.in_degree(node) == 0: - targets.append(os.path.join(self.root_dir, node)) - else: - targets.append(target) - - self.files_to_git_add = [] - - ret = [] - with self.state: - for target in targets: - stages = self._reproduce( - target, - recursive=recursive, - force=force, - dry=dry, - interactive=interactive, - ignore_build_cache=ignore_build_cache, - ) - ret.extend(stages) - - self._remind_to_git_add() - - return ret - - def _reproduce( - self, - target, - recursive=True, - force=False, - dry=False, - interactive=False, - ignore_build_cache=False, - ): - import networkx as nx - from dvc.stage import Stage - - stage = Stage.load(self, target) - G = self.graph()[1] - stages = nx.get_node_attributes(G, "stage") - node = os.path.relpath(stage.path, self.root_dir) - - if recursive: - ret = self._reproduce_stages( - G, stages, node, force, dry, interactive, ignore_build_cache - ) - else: - ret = self._reproduce_stage(stages, node, force, dry, interactive) - - return ret - - def _reproduce_stages( - self, G, stages, node, force, dry, interactive, ignore_build_cache - ): - import networkx as nx - - result = [] - for n in nx.dfs_postorder_nodes(G, node): - try: - ret = self._reproduce_stage(stages, n, force, dry, interactive) - - if len(ret) == 0 and ignore_build_cache: - # NOTE: we are walking our pipeline from the top to the - # bottom. If one stage is changed, it will be reproduced, - # which tells us that we should force reproducing all of - # the other stages down below, even if their direct - # dependencies didn't change. - force = True - - result += ret - except Exception as ex: - raise ReproductionError(stages[n].relpath, ex) - return result - - def _cleanup_unused_links(self, all_stages): - used = [] - for stage in all_stages: - for out in stage.outs: - used.append(out.path) - self.state.remove_unused_links(used) - - def checkout( - self, target=None, with_deps=False, force=False, recursive=False - ): - if target and not recursive: - from dvc.stage import ( - StageFileDoesNotExistError, - StageFileBadNameError, - ) - - all_stages = self.active_stages() - try: - stages = self._collect(target, with_deps=with_deps) - except (StageFileDoesNotExistError, StageFileBadNameError) as exc: - raise DvcException( - str(exc) - + " Did you mean 'git checkout {}'?".format(target) - ) - else: - all_stages = self.active_stages(target) - stages = all_stages - - with self.state: - self._cleanup_unused_links(all_stages) - - for stage in stages: - if stage.locked: - logger.warning( - "DVC file '{path}' is locked. Its dependencies are" - " not going to be checked out.".format( - path=stage.relpath - ) - ) - - stage.checkout(force=force) - - def _get_pipeline(self, node): - pipelines = list(filter(lambda g: node in g.nodes(), self.pipelines())) - assert len(pipelines) == 1 - return pipelines[0] - - def _collect(self, target, with_deps=False): - import networkx as nx - from dvc.stage import Stage - - stage = Stage.load(self, target) - if not with_deps: - return [stage] - - node = os.path.relpath(stage.path, self.root_dir) - G = self._get_pipeline(node) - stages = nx.get_node_attributes(G, "stage") - - ret = [stage] - for n in nx.dfs_postorder_nodes(G, node): - ret.append(stages[n]) - - return ret - - def _collect_dir_cache( - self, out, branch=None, remote=None, force=False, jobs=None - ): - info = out.dumpd() - ret = [info] - r = out.remote - md5 = info[r.PARAM_CHECKSUM] - - if self.cache.local.changed_cache_file(md5): - try: - self.cloud.pull( - ret, jobs=jobs, remote=remote, show_checksums=False - ) - except DvcException as exc: - msg = "Failed to pull cache for '{}': {}" - logger.debug(msg.format(out, exc)) - - if self.cache.local.changed_cache_file(md5): - msg = ( - "Missing cache for directory '{}'. " - "Cache for files inside will be lost. " - "Would you like to continue? Use '-f' to force. " - ) - if not force and not prompt.confirm(msg): - raise DvcException( - "unable to fully collect used cache" - " without cache for directory '{}'".format(out) - ) - else: - return ret - - for i in self.cache.local.load_dir_cache(md5): - i["branch"] = branch - i[r.PARAM_PATH] = os.path.join( - info[r.PARAM_PATH], i[r.PARAM_RELPATH] - ) - ret.append(i) - - return ret - - def _collect_used_cache( - self, out, branch=None, remote=None, force=False, jobs=None - ): - if not out.use_cache or not out.info: - if not out.info: - logger.warning( - "Output '{}'({}) is missing version " - "info. Cache for it will not be collected. " - "Use dvc repro to get your pipeline up to " - "date.".format(out, out.stage) - ) - return [] - - info = out.dumpd() - info["branch"] = branch - ret = [info] - - if out.scheme != "local": - return ret - - md5 = info[out.remote.PARAM_CHECKSUM] - cache = self.cache.local.get(md5) - if not out.remote.is_dir_cache(cache): - return ret - - return self._collect_dir_cache( - out, branch=branch, remote=remote, force=force, jobs=jobs - ) - - def _used_cache( - self, - target=None, - all_branches=False, - active=True, - with_deps=False, - all_tags=False, - remote=None, - force=False, - jobs=None, - recursive=False, - ): - cache = {} - cache["local"] = [] - cache["s3"] = [] - cache["gs"] = [] - cache["hdfs"] = [] - cache["ssh"] = [] - cache["azure"] = [] - - for branch in self.scm.brancher( - all_branches=all_branches, all_tags=all_tags - ): - if target: - if recursive: - stages = self.stages(target) - else: - stages = self._collect(target, with_deps=with_deps) - elif active: - stages = self.active_stages() - else: - stages = self.stages() - - for stage in stages: - if active and not target and stage.locked: - logger.warning( - "DVC file '{path}' is locked. Its dependencies are" - " not going to be pushed/pulled/fetched.".format( - path=stage.relpath - ) - ) - - for out in stage.outs: - scheme = out.path_info["scheme"] - cache[scheme] += self._collect_used_cache( - out, - branch=branch, - remote=remote, - force=force, - jobs=jobs, - ) - - return cache - - @staticmethod - def merge_cache_lists(clists): - merged_cache = collections.defaultdict(list) - - for cache_list in clists: - for scheme, cache in cache_list.items(): - for item in cache: - if item not in merged_cache[scheme]: - merged_cache[scheme].append(item) - - return merged_cache - - @staticmethod - def load_all_used_cache( - projects, - target=None, - all_branches=False, - active=True, - with_deps=False, - all_tags=False, - remote=None, - force=False, - jobs=None, - ): - clists = [] - - for project in projects: - with project.state: - project_clist = project._used_cache( - target=None, - all_branches=all_branches, - active=False, - with_deps=with_deps, - all_tags=all_tags, - remote=remote, - force=force, - jobs=jobs, - ) - - clists.append(project_clist) - - return clists - - def _do_gc(self, typ, func, clist): - removed = func(clist) - if not removed: - logger.info("No unused {} cache to remove.".format(typ)) - - def gc( - self, - all_branches=False, - cloud=False, - remote=None, - with_deps=False, - all_tags=False, - force=False, - jobs=None, - projects=None, - ): - - all_projects = [self] - - if projects: - all_projects.extend(Project(path) for path in projects) - - all_clists = Project.load_all_used_cache( - all_projects, - target=None, - all_branches=all_branches, - active=False, - with_deps=with_deps, - all_tags=all_tags, - remote=remote, - force=force, - jobs=jobs, - ) - - if len(all_clists) > 1: - clist = Project.merge_cache_lists(all_clists) - else: - clist = all_clists[0] - - with self.state: - self._do_gc("local", self.cache.local.gc, clist) - - if self.cache.s3: - self._do_gc("s3", self.cache.s3.gc, clist) - - if self.cache.gs: - self._do_gc("gs", self.cache.gs.gc, clist) - - if self.cache.ssh: - self._do_gc("ssh", self.cache.ssh.gc, clist) - - if self.cache.hdfs: - self._do_gc("hdfs", self.cache.hdfs.gc, clist) - - if self.cache.azure: - self._do_gc("azure", self.cache.azure.gc, clist) - - if cloud: - self._do_gc( - "remote", self.cloud._get_cloud(remote, "gc -c").gc, clist - ) - - def push( - self, - target=None, - jobs=1, - remote=None, - all_branches=False, - show_checksums=False, - with_deps=False, - all_tags=False, - recursive=False, - ): - with self.state: - used = self._used_cache( - target, - all_branches=all_branches, - all_tags=all_tags, - with_deps=with_deps, - force=True, - remote=remote, - jobs=jobs, - recursive=recursive, - )["local"] - self.cloud.push( - used, jobs, remote=remote, show_checksums=show_checksums - ) - - def fetch( - self, - target=None, - jobs=1, - remote=None, - all_branches=False, - show_checksums=False, - with_deps=False, - all_tags=False, - recursive=False, - ): - with self.state: - used = self._used_cache( - target, - all_branches=all_branches, - all_tags=all_tags, - with_deps=with_deps, - force=True, - remote=remote, - jobs=jobs, - recursive=recursive, - )["local"] - self.cloud.pull( - used, jobs, remote=remote, show_checksums=show_checksums - ) - - def pull( - self, - target=None, - jobs=1, - remote=None, - all_branches=False, - show_checksums=False, - with_deps=False, - all_tags=False, - force=False, - recursive=False, - ): - self.fetch( - target, - jobs, - remote=remote, - all_branches=all_branches, - all_tags=all_tags, - show_checksums=show_checksums, - with_deps=with_deps, - recursive=recursive, - ) - self.checkout( - target=target, - with_deps=with_deps, - force=force, - recursive=recursive, - ) - - def _local_status(self, target=None, with_deps=False): - status = {} - - if target: - stages = self._collect(target, with_deps=with_deps) - else: - stages = self.active_stages() - - for stage in stages: - if stage.locked: - logger.warning( - "DVC file '{path}' is locked. Its dependencies are" - " not going to be shown in the status output.".format( - path=stage.relpath - ) - ) - - status.update(stage.status()) - - return status - - def _cloud_status( - self, - target=None, - jobs=1, - remote=None, - show_checksums=False, - all_branches=False, - with_deps=False, - all_tags=False, - ): - import dvc.remote.base as cloud - - used = self._used_cache( - target, - all_branches=all_branches, - all_tags=all_tags, - with_deps=with_deps, - force=True, - remote=remote, - jobs=jobs, - )["local"] - - ret = {} - status_info = self.cloud.status( - used, jobs, remote=remote, show_checksums=show_checksums - ) - for md5, info in status_info.items(): - name = info["name"] - status = info["status"] - if status == cloud.STATUS_OK: - continue - - prefix_map = { - cloud.STATUS_DELETED: "deleted", - cloud.STATUS_NEW: "new", - } - - ret[name] = prefix_map[status] - - return ret - - def status( - self, - target=None, - jobs=1, - cloud=False, - remote=None, - show_checksums=False, - all_branches=False, - with_deps=False, - all_tags=False, - ): - with self.state: - if cloud: - return self._cloud_status( - target, - jobs, - remote=remote, - show_checksums=show_checksums, - all_branches=all_branches, - with_deps=with_deps, - all_tags=all_tags, - ) - return self._local_status(target, with_deps=with_deps) - - @staticmethod - def _read_metric_json(fd, json_path): - import json - from jsonpath_rw import parse - - parser = parse(json_path) - return [x.value for x in parser.find(json.load(fd))] - - @staticmethod - def _do_read_metric_xsv(reader, row, col): - if col is not None and row is not None: - return [reader[row][col]] - elif col is not None: - return [r[col] for r in reader] - elif row is not None: - return reader[row] - return None - - @classmethod - def _read_metric_hxsv(cls, fd, hxsv_path, delimiter): - import csv - - col, row = hxsv_path.split(",") - row = int(row) - reader = list(csv.DictReader(fd, delimiter=builtin_str(delimiter))) - return cls._do_read_metric_xsv(reader, row, col) - - @classmethod - def _read_metric_xsv(cls, fd, xsv_path, delimiter): - import csv - - col, row = xsv_path.split(",") - row = int(row) - col = int(col) - reader = list(csv.reader(fd, delimiter=builtin_str(delimiter))) - return cls._do_read_metric_xsv(reader, row, col) - - def _read_metric(self, path, typ=None, xpath=None): - ret = None - - if not os.path.exists(path): - return ret - - try: - with open(path, "r") as fd: - if typ == "json": - ret = self._read_metric_json(fd, xpath) - elif typ == "csv": - ret = self._read_metric_xsv(fd, xpath, ",") - elif typ == "tsv": - ret = self._read_metric_xsv(fd, xpath, "\t") - elif typ == "hcsv": - ret = self._read_metric_hxsv(fd, xpath, ",") - elif typ == "htsv": - ret = self._read_metric_hxsv(fd, xpath, "\t") - else: - ret = fd.read().strip() - except Exception: - logger.error("unable to read metric in '{}'".format(path)) - - return ret - - def _find_output_by_path(self, path, outs=None, recursive=False): - from dvc.exceptions import OutputDuplicationError - - if not outs: - astages = self.active_stages() - outs = [out for stage in astages for out in stage.outs] - - abs_path = os.path.abspath(path) - if os.path.isdir(abs_path) and recursive: - matched = [ - out - for out in outs - if os.path.abspath(out.path).startswith(abs_path) - ] - else: - matched = [out for out in outs if out.path == abs_path] - stages = [out.stage.relpath for out in matched] - if len(stages) > 1: - raise OutputDuplicationError(path, stages) - - return matched if matched else [] - - def metrics_show( - self, - path=None, - typ=None, - xpath=None, - all_branches=False, - all_tags=False, - recursive=False, - ): - res = {} - for branch in self.scm.brancher( - all_branches=all_branches, all_tags=all_tags - ): - astages = self.active_stages() - outs = [out for stage in astages for out in stage.outs] - - if path: - outs = self._find_output_by_path( - path, outs=outs, recursive=recursive - ) - - metrics = filter(lambda o: o.metric, outs) - stages = set() - entries = [] - - for o in metrics: - if not typ and isinstance(o.metric, dict): - t = o.metric.get(o.PARAM_METRIC_TYPE, typ) - x = o.metric.get(o.PARAM_METRIC_XPATH, xpath) - else: - t = typ - x = xpath - entries.append((o.path, t, x)) - stages.add(o.stage.path) - - if path and not entries: - if os.path.isdir(path): - logger.warning( - "Path '{path}' is a directory. " - "Consider running with '-R'.".format(path=path) - ) - return {} - - for fname, t, x in entries: - if stages: - for stage in stages: - self.checkout(stage, force=True) - - rel = os.path.relpath(fname) - metric = self._read_metric(fname, typ=t, xpath=x) - if not metric: - continue - - if branch not in res: - res[branch] = {} - - res[branch][rel] = metric - - for branch, val in res.items(): - if all_branches or all_tags: - logger.info("{}:".format(branch)) - for fname, metric in val.items(): - logger.info("\t{}: {}".format(fname, metric)) - - if res: - return res - - if path and os.path.isdir(path): - return res - - if path: - msg = "file '{}' does not exist, not a metric file or is malformed".format( - path - ) - else: - msg = ( - "no metric files in this repository." - " use 'dvc metrics add' to add a metric file to track." - ) - - raise DvcException(msg) - - def _metrics_modify(self, path, typ=None, xpath=None, delete=False): - outs = self._find_output_by_path(path) - - if not outs: - msg = "unable to find file '{}' in the pipeline".format(path) - raise DvcException(msg) - - if len(outs) != 1: - msg = ( - "-R not yet supported for metrics modify. " - "Make sure only one metric is referred to by '{}'".format(path) - ) - raise DvcException(msg) - - out = outs[0] - - if out.scheme != "local": - msg = "output '{}' scheme '{}' is not supported for metrics" - raise DvcException(msg.format(out.path, out.path_info["scheme"])) - - if typ: - if not isinstance(out.metric, dict): - out.metric = {} - out.metric[out.PARAM_METRIC_TYPE] = typ - - if xpath: - if not isinstance(out.metric, dict): - out.metric = {} - out.metric[out.PARAM_METRIC_XPATH] = xpath - - if delete: - out.metric = None - - out._verify_metric() - - out.stage.dump() - - def metrics_modify(self, path=None, typ=None, xpath=None): - self._metrics_modify(path, typ, xpath) - - def metrics_add(self, path, typ=None, xpath=None): - if not typ: - typ = "raw" - self._metrics_modify(path, typ, xpath) - - def metrics_remove(self, path): - self._metrics_modify(path, delete=True) - - def graph(self, stages=None, from_directory=None): - import networkx as nx - from dvc.exceptions import ( - OutputDuplicationError, - WorkingDirectoryAsOutputError, - ) - - G = nx.DiGraph() - G_active = nx.DiGraph() - stages = stages or self.stages(from_directory) - stages = [stage for stage in stages if stage] - outs = [] - - for stage in stages: - for out in stage.outs: - existing = [o.stage for o in outs if o.path == out.path] - - if existing: - stages = [stage.relpath, existing[0].relpath] - raise OutputDuplicationError(out.path, stages) - - outs.append(out) - - for stage in stages: - for out in outs: - overlaps = stage.cwd == out.path or stage.cwd.startswith( - out.path + os.sep - ) - - if overlaps: - raise WorkingDirectoryAsOutputError( - stage.cwd, stage.relpath - ) - - # collect the whole DAG - for stage in stages: - node = os.path.relpath(stage.path, self.root_dir) - - G.add_node(node, stage=stage) - G_active.add_node(node, stage=stage) - - for dep in stage.deps: - for out in outs: - if ( - out.path != dep.path - and not dep.path.startswith(out.path + out.sep) - and not out.path.startswith(dep.path + dep.sep) - ): - continue - - dep_stage = out.stage - dep_node = os.path.relpath(dep_stage.path, self.root_dir) - G.add_node(dep_node, stage=dep_stage) - G.add_edge(node, dep_node) - if not stage.locked: - G_active.add_node(dep_node, stage=dep_stage) - G_active.add_edge(node, dep_node) - - self._check_cyclic_graph(G) - - return G, G_active - - def pipelines(self, from_directory=None): - import networkx as nx - - G, G_active = self.graph(from_directory=from_directory) - - return [ - G.subgraph(c).copy() for c in nx.weakly_connected_components(G) - ] - - def stages(self, from_directory=None): - """ - Walks down the root directory looking for Dvcfiles, - skipping the directories that are related with - any SCM (e.g. `.git`), DVC itself (`.dvc`), or directories - tracked by DVC (e.g. `dvc add data` would skip `data/`) - - NOTE: For large projects, this could be an expensive - operation. Consider using some memoization. - """ - from dvc.stage import Stage - - if not from_directory: - from_directory = self.root_dir - - stages = [] - outs = [] - for root, dirs, files in os.walk(str(from_directory)): - for fname in files: - path = os.path.join(root, fname) - if not Stage.is_stage_file(path): - continue - stage = Stage.load(self, path) - for out in stage.outs: - outs.append(out.path + out.sep) - stages.append(stage) - - def filter_dirs(dname, root=root): - path = os.path.join(root, dname) - if path in (self.dvc_dir, self.scm.dir): - return False - for out in outs: - if path == os.path.normpath(out) or path.startswith(out): - return False - return True - - dirs[:] = list(filter(filter_dirs, dirs)) - - return stages - - def active_stages(self, from_directory=None): - import networkx as nx - - stages = [] - for G in self.pipelines(from_directory): - stages.extend(list(nx.get_node_attributes(G, "stage").values())) - return stages - - @staticmethod - def _welcome_message(): - import colorama - - logger.box( - "DVC has enabled anonymous aggregate usage analytics.\n" - "Read the analytics documentation (and how to opt-out) here:\n" - "{blue}https://dvc.org/doc/user-guide/analytics{nc}".format( - blue=colorama.Fore.BLUE, nc=colorama.Fore.RESET - ), - border_color="red", - ) - - logger.info( - "{yellow}What's next?{nc}\n" - "{yellow}------------{nc}\n" - "- Check out the documentation: {blue}https://dvc.org/doc{nc}\n" - "- Get help and share ideas: {blue}https://dvc.org/chat{nc}\n" - "- Star us on GitHub: {blue}https://github.com/iterative/dvc{nc}".format( - yellow=colorama.Fore.YELLOW, - blue=colorama.Fore.BLUE, - nc=colorama.Fore.RESET, - ) - ) - - @staticmethod - def _expand_target_path(from_path, to_path): - if os.path.isdir(to_path) and not os.path.isdir(from_path): - return os.path.join(to_path, os.path.basename(from_path)) - return to_path diff --git a/dvc/remote/__init__.py b/dvc/remote/__init__.py index 7b1bdcb92..94b274f4a 100644 --- a/dvc/remote/__init__.py +++ b/dvc/remote/__init__.py @@ -27,5 +27,5 @@ def _get(config): return RemoteLOCAL -def Remote(project, config): - return _get(config)(project, config) +def Remote(repo, config): + return _get(config)(repo, config) diff --git a/dvc/remote/azure.py b/dvc/remote/azure.py index 077f0f742..d73d431be 100644 --- a/dvc/remote/azure.py +++ b/dvc/remote/azure.py @@ -36,8 +36,8 @@ class RemoteAzure(RemoteBase): PARAM_CHECKSUM = "etag" COPY_POLL_SECONDS = 5 - def __init__(self, project, config): - super(RemoteAzure, self).__init__(project, config) + def __init__(self, repo, config): + super(RemoteAzure, self).__init__(repo, config) self.url = config.get(Config.SECTION_REMOTE_URL, "azure://") match = re.match(self.REGEX, self.url) # backward compatibility diff --git a/dvc/remote/base.py b/dvc/remote/base.py index e3b35b21c..b08b83986 100644 --- a/dvc/remote/base.py +++ b/dvc/remote/base.py @@ -59,8 +59,8 @@ class RemoteBase(object): REQUIRES = {} JOBS = 4 * cpu_count() - def __init__(self, project, config): - self.project = project + def __init__(self, repo, config): + self.repo = repo deps_ok = all(self.REQUIRES.values()) if not deps_ok: missing = [k for k, v in self.REQUIRES.items() if v is None] diff --git a/dvc/remote/gs.py b/dvc/remote/gs.py index ac5d1f2a6..c2d4c3dbe 100644 --- a/dvc/remote/gs.py +++ b/dvc/remote/gs.py @@ -22,8 +22,8 @@ class RemoteGS(RemoteBase): REQUIRES = {"google.cloud.storage": storage} PARAM_CHECKSUM = "md5" - def __init__(self, project, config): - super(RemoteGS, self).__init__(project, config) + def __init__(self, repo, config): + super(RemoteGS, self).__init__(repo, config) storagepath = "gs://" storagepath += config.get(Config.SECTION_AWS_STORAGEPATH, "/") storagepath.lstrip("/") diff --git a/dvc/remote/hdfs.py b/dvc/remote/hdfs.py index 2aa0a25ad..5d36500cb 100644 --- a/dvc/remote/hdfs.py +++ b/dvc/remote/hdfs.py @@ -17,8 +17,8 @@ class RemoteHDFS(RemoteBase): REGEX = r"^hdfs://((?P.*)@)?.*$" PARAM_CHECKSUM = "checksum" - def __init__(self, project, config): - super(RemoteHDFS, self).__init__(project, config) + def __init__(self, repo, config): + super(RemoteHDFS, self).__init__(repo, config) self.url = config.get(Config.SECTION_REMOTE_URL, "/") self.prefix = self.url self.user = self.group("user") diff --git a/dvc/remote/http.py b/dvc/remote/http.py index c44bcedc0..7e2b3855b 100644 --- a/dvc/remote/http.py +++ b/dvc/remote/http.py @@ -32,8 +32,8 @@ class RemoteHTTP(RemoteBase): CHUNK_SIZE = 2 ** 16 PARAM_CHECKSUM = "etag" - def __init__(self, project, config): - super(RemoteHTTP, self).__init__(project, config) + def __init__(self, repo, config): + super(RemoteHTTP, self).__init__(repo, config) self.cache_dir = config.get(Config.SECTION_REMOTE_URL) self.url = self.cache_dir self.path_info = {"scheme": "http"} diff --git a/dvc/remote/local.py b/dvc/remote/local.py index 58f71dc32..da1cdf724 100644 --- a/dvc/remote/local.py +++ b/dvc/remote/local.py @@ -38,9 +38,9 @@ class RemoteLOCAL(RemoteBase): "reflink": System.reflink, } - def __init__(self, project, config): - super(RemoteLOCAL, self).__init__(project, config) - self.state = self.project.state if self.project else None + def __init__(self, repo, config): + super(RemoteLOCAL, self).__init__(repo, config) + self.state = self.repo.state if self.repo else None self.protected = config.get(Config.SECTION_CACHE_PROTECTED, False) storagepath = config.get(Config.SECTION_AWS_STORAGEPATH, None) self.cache_dir = config.get(Config.SECTION_REMOTE_URL, storagepath) diff --git a/dvc/remote/s3.py b/dvc/remote/s3.py index 0c05c047a..d7c3d4f19 100644 --- a/dvc/remote/s3.py +++ b/dvc/remote/s3.py @@ -35,8 +35,8 @@ class RemoteS3(RemoteBase): REQUIRES = {"boto3": boto3} PARAM_CHECKSUM = "etag" - def __init__(self, project, config): - super(RemoteS3, self).__init__(project, config) + def __init__(self, repo, config): + super(RemoteS3, self).__init__(repo, config) storagepath = "s3://{}".format( config.get(Config.SECTION_AWS_STORAGEPATH, "").lstrip("/") diff --git a/dvc/remote/ssh.py b/dvc/remote/ssh.py index 859b2daac..a9072318f 100644 --- a/dvc/remote/ssh.py +++ b/dvc/remote/ssh.py @@ -59,8 +59,8 @@ class RemoteSSH(RemoteBase): DEFAULT_PORT = 22 TIMEOUT = 1800 - def __init__(self, project, config): - super(RemoteSSH, self).__init__(project, config) + def __init__(self, repo, config): + super(RemoteSSH, self).__init__(repo, config) self.url = config.get(Config.SECTION_REMOTE_URL, "ssh://") parsed = urlparse(self.url) diff --git a/dvc/repo/__init__.py b/dvc/repo/__init__.py new file mode 100644 index 000000000..fe2911f4f --- /dev/null +++ b/dvc/repo/__init__.py @@ -0,0 +1,410 @@ +from __future__ import unicode_literals + +from dvc.utils.compat import str, builtin_str, open + +import collections +import os +import dvc.prompt as prompt +import dvc.logger as logger + +from dvc.exceptions import DvcException, NotDvcRepoError + + +class Repo(object): + DVC_DIR = ".dvc" + + from dvc.repo.destroy import destroy + from dvc.repo.install import install + from dvc.repo.add import add + from dvc.repo.remove import remove + from dvc.repo.lock import lock as lock_stage + from dvc.repo.move import move + from dvc.repo.run import run + from dvc.repo.imp import imp + from dvc.repo.reproduce import reproduce + from dvc.repo.checkout import checkout + from dvc.repo.push import push + from dvc.repo.fetch import fetch + from dvc.repo.pull import pull + from dvc.repo.status import status + from dvc.repo.gc import gc + + def __init__(self, root_dir=None): + from dvc.config import Config + from dvc.state import State + from dvc.lock import Lock + from dvc.scm import SCM + from dvc.cache import Cache + from dvc.data_cloud import DataCloud + from dvc.updater import Updater + from dvc.repo.metrics import Metrics + + root_dir = self.find_root(root_dir) + + self.root_dir = os.path.abspath(os.path.realpath(root_dir)) + self.dvc_dir = os.path.join(self.root_dir, self.DVC_DIR) + + self.config = Config(self.dvc_dir) + + self.scm = SCM(self.root_dir, repo=self) + self.lock = Lock(self.dvc_dir) + # NOTE: storing state and link_state in the repository itself to avoid + # any possible state corruption in 'shared cache dir' scenario. + self.state = State(self, self.config.config) + + core = self.config.config[Config.SECTION_CORE] + + logger.set_level(core.get(Config.SECTION_CORE_LOGLEVEL)) + + self.cache = Cache(self) + self.cloud = DataCloud(self, config=self.config.config) + self.updater = Updater(self.dvc_dir) + + self.metrics = Metrics(self) + + self.files_to_git_add = [] + + self._ignore() + + self.updater.check() + + def __repr__(self): + return "Repo: '{root_dir}'".format(root_dir=self.root_dir) + + @staticmethod + def find_root(root=None): + if root is None: + root = os.getcwd() + else: + root = os.path.abspath(os.path.realpath(root)) + + while True: + dvc_dir = os.path.join(root, Repo.DVC_DIR) + if os.path.isdir(dvc_dir): + return root + if os.path.ismount(root): + break + root = os.path.dirname(root) + raise NotDvcRepoError(root) + + @staticmethod + def find_dvc_dir(root=None): + root_dir = Repo.find_root(root) + return os.path.join(root_dir, Repo.DVC_DIR) + + def remind_to_git_add(self): + if not self.files_to_git_add: + return + + logger.info( + "\n" + "To track the changes with git run:\n" + "\n" + "\tgit add {files}".format(files=" ".join(self.files_to_git_add)) + ) + + @staticmethod + def init(root_dir=os.curdir, no_scm=False, force=False): + from dvc.repo.init import init + + init(root_dir=root_dir, no_scm=no_scm, force=force) + return Repo(root_dir) + + @staticmethod + def unprotect(target): + from dvc.repo.unprotect import unprotect + + return unprotect(target) + + def _ignore(self): + flist = [ + self.state.state_file, + self.lock.lock_file, + self.config.config_local_file, + self.updater.updater_file, + self.updater.lock.lock_file, + ] + self.state.temp_files + + if self.cache.local.cache_dir.startswith(self.root_dir): + flist += [self.cache.local.cache_dir] + + self.scm.ignore_list(flist) + + def check_dag(self, stages): + """Generate graph including the new stage to check for errors""" + self.graph(stages=stages) + + @staticmethod + def _check_cyclic_graph(graph): + import networkx as nx + from dvc.exceptions import CyclicGraphError + + cycles = list(nx.simple_cycles(graph)) + + if cycles: + raise CyclicGraphError(cycles[0]) + + def _get_pipeline(self, node): + pipelines = list(filter(lambda g: node in g.nodes(), self.pipelines())) + assert len(pipelines) == 1 + return pipelines[0] + + def collect(self, target, with_deps=False): + import networkx as nx + from dvc.stage import Stage + + stage = Stage.load(self, target) + if not with_deps: + return [stage] + + node = os.path.relpath(stage.path, self.root_dir) + G = self._get_pipeline(node) + stages = nx.get_node_attributes(G, "stage") + + ret = [stage] + for n in nx.dfs_postorder_nodes(G, node): + ret.append(stages[n]) + + return ret + + def _collect_dir_cache( + self, out, branch=None, remote=None, force=False, jobs=None + ): + info = out.dumpd() + ret = [info] + r = out.remote + md5 = info[r.PARAM_CHECKSUM] + + if self.cache.local.changed_cache_file(md5): + try: + self.cloud.pull( + ret, jobs=jobs, remote=remote, show_checksums=False + ) + except DvcException as exc: + msg = "Failed to pull cache for '{}': {}" + logger.debug(msg.format(out, exc)) + + if self.cache.local.changed_cache_file(md5): + msg = ( + "Missing cache for directory '{}'. " + "Cache for files inside will be lost. " + "Would you like to continue? Use '-f' to force. " + ) + if not force and not prompt.confirm(msg): + raise DvcException( + "unable to fully collect used cache" + " without cache for directory '{}'".format(out) + ) + else: + return ret + + for i in self.cache.local.load_dir_cache(md5): + i["branch"] = branch + i[r.PARAM_PATH] = os.path.join( + info[r.PARAM_PATH], i[r.PARAM_RELPATH] + ) + ret.append(i) + + return ret + + def _collect_used_cache( + self, out, branch=None, remote=None, force=False, jobs=None + ): + if not out.use_cache or not out.info: + if not out.info: + logger.warning( + "Output '{}'({}) is missing version " + "info. Cache for it will not be collected. " + "Use dvc repro to get your pipeline up to " + "date.".format(out, out.stage) + ) + return [] + + info = out.dumpd() + info["branch"] = branch + ret = [info] + + if out.scheme != "local": + return ret + + md5 = info[out.remote.PARAM_CHECKSUM] + cache = self.cache.local.get(md5) + if not out.remote.is_dir_cache(cache): + return ret + + return self._collect_dir_cache( + out, branch=branch, remote=remote, force=force, jobs=jobs + ) + + def used_cache( + self, + target=None, + all_branches=False, + active=True, + with_deps=False, + all_tags=False, + remote=None, + force=False, + jobs=None, + recursive=False, + ): + cache = {} + cache["local"] = [] + cache["s3"] = [] + cache["gs"] = [] + cache["hdfs"] = [] + cache["ssh"] = [] + cache["azure"] = [] + + for branch in self.scm.brancher( + all_branches=all_branches, all_tags=all_tags + ): + if target: + if recursive: + stages = self.stages(target) + else: + stages = self.collect(target, with_deps=with_deps) + elif active: + stages = self.active_stages() + else: + stages = self.stages() + + for stage in stages: + if active and not target and stage.locked: + logger.warning( + "DVC file '{path}' is locked. Its dependencies are" + " not going to be pushed/pulled/fetched.".format( + path=stage.relpath + ) + ) + + for out in stage.outs: + scheme = out.path_info["scheme"] + cache[scheme] += self._collect_used_cache( + out, + branch=branch, + remote=remote, + force=force, + jobs=jobs, + ) + + return cache + + def graph(self, stages=None, from_directory=None): + import networkx as nx + from dvc.exceptions import ( + OutputDuplicationError, + WorkingDirectoryAsOutputError, + ) + + G = nx.DiGraph() + G_active = nx.DiGraph() + stages = stages or self.stages(from_directory) + stages = [stage for stage in stages if stage] + outs = [] + + for stage in stages: + for out in stage.outs: + existing = [o.stage for o in outs if o.path == out.path] + + if existing: + stages = [stage.relpath, existing[0].relpath] + raise OutputDuplicationError(out.path, stages) + + outs.append(out) + + for stage in stages: + for out in outs: + overlaps = stage.cwd == out.path or stage.cwd.startswith( + out.path + os.sep + ) + + if overlaps: + raise WorkingDirectoryAsOutputError( + stage.cwd, stage.relpath + ) + + # collect the whole DAG + for stage in stages: + node = os.path.relpath(stage.path, self.root_dir) + + G.add_node(node, stage=stage) + G_active.add_node(node, stage=stage) + + for dep in stage.deps: + for out in outs: + if ( + out.path != dep.path + and not dep.path.startswith(out.path + out.sep) + and not out.path.startswith(dep.path + dep.sep) + ): + continue + + dep_stage = out.stage + dep_node = os.path.relpath(dep_stage.path, self.root_dir) + G.add_node(dep_node, stage=dep_stage) + G.add_edge(node, dep_node) + if not stage.locked: + G_active.add_node(dep_node, stage=dep_stage) + G_active.add_edge(node, dep_node) + + self._check_cyclic_graph(G) + + return G, G_active + + def pipelines(self, from_directory=None): + import networkx as nx + + G, G_active = self.graph(from_directory=from_directory) + + return [ + G.subgraph(c).copy() for c in nx.weakly_connected_components(G) + ] + + def stages(self, from_directory=None): + """ + Walks down the root directory looking for Dvcfiles, + skipping the directories that are related with + any SCM (e.g. `.git`), DVC itself (`.dvc`), or directories + tracked by DVC (e.g. `dvc add data` would skip `data/`) + + NOTE: For large repos, this could be an expensive + operation. Consider using some memoization. + """ + from dvc.stage import Stage + + if not from_directory: + from_directory = self.root_dir + + stages = [] + outs = [] + for root, dirs, files in os.walk(str(from_directory)): + for fname in files: + path = os.path.join(root, fname) + if not Stage.is_stage_file(path): + continue + stage = Stage.load(self, path) + for out in stage.outs: + outs.append(out.path + out.sep) + stages.append(stage) + + def filter_dirs(dname, root=root): + path = os.path.join(root, dname) + if path in (self.dvc_dir, self.scm.dir): + return False + for out in outs: + if path == os.path.normpath(out) or path.startswith(out): + return False + return True + + dirs[:] = list(filter(filter_dirs, dirs)) + + return stages + + def active_stages(self, from_directory=None): + import networkx as nx + + stages = [] + for G in self.pipelines(from_directory): + stages.extend(list(nx.get_node_attributes(G, "stage").values())) + return stages diff --git a/dvc/repo/add.py b/dvc/repo/add.py new file mode 100644 index 000000000..3e727c176 --- /dev/null +++ b/dvc/repo/add.py @@ -0,0 +1,43 @@ +import os + + +def add(self, fname, recursive=False): + from dvc.stage import Stage + + fnames = [] + if recursive and os.path.isdir(fname): + for root, dirs, files in os.walk(str(fname)): + for f in files: + path = os.path.join(root, f) + if Stage.is_stage_file(path): + continue + if os.path.basename(path) == self.scm.ignore_file: + continue + if self.scm.is_tracked(path): + continue + fnames.append(path) + else: + fnames = [fname] + + stages = [] + self.files_to_git_add = [] + with self.state: + for f in fnames: + stage = Stage.create(repo=self, outs=[f], add=True) + + if stage is None: + stages.append(stage) + continue + + stage.save() + stages.append(stage) + + self.check_dag(self.stages() + stages) + + for stage in stages: + if stage is not None: + stage.dump() + + self.remind_to_git_add() + + return stages diff --git a/dvc/repo/checkout.py b/dvc/repo/checkout.py new file mode 100644 index 000000000..0b7adc5fa --- /dev/null +++ b/dvc/repo/checkout.py @@ -0,0 +1,39 @@ +from __future__ import unicode_literals + +import dvc.logger as logger +from dvc.stage import StageFileDoesNotExistError, StageFileBadNameError +from dvc.exceptions import DvcException + + +def _cleanup_unused_links(self, all_stages): + used = [] + for stage in all_stages: + for out in stage.outs: + used.append(out.path) + self.state.remove_unused_links(used) + + +def checkout(self, target=None, with_deps=False, force=False, recursive=False): + if target and not recursive: + all_stages = self.active_stages() + try: + stages = self.collect(target, with_deps=with_deps) + except (StageFileDoesNotExistError, StageFileBadNameError) as exc: + raise DvcException( + str(exc) + " Did you mean 'git checkout {}'?".format(target) + ) + else: + all_stages = self.active_stages(target) + stages = all_stages + + with self.state: + _cleanup_unused_links(self, all_stages) + + for stage in stages: + if stage.locked: + logger.warning( + "DVC file '{path}' is locked. Its dependencies are" + " not going to be checked out.".format(path=stage.relpath) + ) + + stage.checkout(force=force) diff --git a/dvc/repo/destroy.py b/dvc/repo/destroy.py new file mode 100644 index 000000000..399d29a51 --- /dev/null +++ b/dvc/repo/destroy.py @@ -0,0 +1,8 @@ +import shutil + + +def destroy(self): + for stage in self.stages(): + stage.remove() + + shutil.rmtree(self.dvc_dir) diff --git a/dvc/repo/fetch.py b/dvc/repo/fetch.py new file mode 100644 index 000000000..4cdc03dca --- /dev/null +++ b/dvc/repo/fetch.py @@ -0,0 +1,28 @@ +from __future__ import unicode_literals + + +def fetch( + self, + target=None, + jobs=1, + remote=None, + all_branches=False, + show_checksums=False, + with_deps=False, + all_tags=False, + recursive=False, +): + with self.state: + used = self.used_cache( + target, + all_branches=all_branches, + all_tags=all_tags, + with_deps=with_deps, + force=True, + remote=remote, + jobs=jobs, + recursive=recursive, + )["local"] + self.cloud.pull( + used, jobs, remote=remote, show_checksums=show_checksums + ) diff --git a/dvc/repo/gc.py b/dvc/repo/gc.py new file mode 100644 index 000000000..b99c39ac8 --- /dev/null +++ b/dvc/repo/gc.py @@ -0,0 +1,111 @@ +from __future__ import unicode_literals + +import collections + +import dvc.logger as logger + + +def _merge_cache_lists(clists): + merged_cache = collections.defaultdict(list) + + for cache_list in clists: + for scheme, cache in cache_list.items(): + for item in cache: + if item not in merged_cache[scheme]: + merged_cache[scheme].append(item) + + return merged_cache + + +def _load_all_used_cache( + repos, + target=None, + all_branches=False, + active=True, + with_deps=False, + all_tags=False, + remote=None, + force=False, + jobs=None, +): + clists = [] + + for repo in repos: + with repo.state: + repo_clist = repo.used_cache( + target=None, + all_branches=all_branches, + active=False, + with_deps=with_deps, + all_tags=all_tags, + remote=remote, + force=force, + jobs=jobs, + ) + + clists.append(repo_clist) + + return clists + + +def _do_gc(typ, func, clist): + removed = func(clist) + if not removed: + logger.info("No unused {} cache to remove.".format(typ)) + + +def gc( + self, + all_branches=False, + cloud=False, + remote=None, + with_deps=False, + all_tags=False, + force=False, + jobs=None, + repos=None, +): + from dvc.repo import Repo + + all_repos = [self] + + if repos: + all_repos.extend(Repo(path) for path in repos) + + all_clists = _load_all_used_cache( + all_repos, + target=None, + all_branches=all_branches, + active=False, + with_deps=with_deps, + all_tags=all_tags, + remote=remote, + force=force, + jobs=jobs, + ) + + if len(all_clists) > 1: + clist = _merge_cache_lists(all_clists) + else: + clist = all_clists[0] + + with self.state: + _do_gc("local", self.cache.local.gc, clist) + + if self.cache.s3: + _do_gc("s3", self.cache.s3.gc, clist) + + if self.cache.gs: + _do_gc("gs", self.cache.gs.gc, clist) + + if self.cache.ssh: + _do_gc("ssh", self.cache.ssh.gc, clist) + + if self.cache.hdfs: + _do_gc("hdfs", self.cache.hdfs.gc, clist) + + if self.cache.azure: + _do_gc("azure", self.cache.azure.gc, clist) + + if cloud: + _do_gc("remote", self.cloud._get_cloud(remote, "gc -c").gc, clist) diff --git a/dvc/repo/imp.py b/dvc/repo/imp.py new file mode 100644 index 000000000..0d40cf499 --- /dev/null +++ b/dvc/repo/imp.py @@ -0,0 +1,20 @@ +from dvc.stage import Stage + + +def imp(self, url, out, resume=False): + stage = Stage.create(repo=self, cmd=None, deps=[url], outs=[out]) + + if stage is None: + return None + + self.check_dag(self.stages() + [stage]) + + self.files_to_git_add = [] + with self.state: + stage.run(resume=resume) + + stage.dump() + + self.remind_to_git_add() + + return stage diff --git a/dvc/repo/init.py b/dvc/repo/init.py new file mode 100644 index 000000000..d0bc62cea --- /dev/null +++ b/dvc/repo/init.py @@ -0,0 +1,88 @@ +import os +import shutil +import colorama + +import dvc.logger as logger +from dvc.repo import Repo +from dvc.scm import SCM, Base +from dvc.config import Config +from dvc.exceptions import InitError + + +def _welcome_message(): + logger.box( + "DVC has enabled anonymous aggregate usage analytics.\n" + "Read the analytics documentation (and how to opt-out) here:\n" + "{blue}https://dvc.org/doc/user-guide/analytics{nc}".format( + blue=colorama.Fore.BLUE, nc=colorama.Fore.RESET + ), + border_color="red", + ) + + logger.info( + "{yellow}What's next?{nc}\n" + "{yellow}------------{nc}\n" + "- Check out the documentation: {blue}https://dvc.org/doc{nc}\n" + "- Get help and share ideas: {blue}https://dvc.org/chat{nc}\n" + "- Star us on GitHub: {blue}https://github.com/iterative/dvc{nc}".format( + yellow=colorama.Fore.YELLOW, + blue=colorama.Fore.BLUE, + nc=colorama.Fore.RESET, + ) + ) + + +def init(root_dir=os.curdir, no_scm=False, force=False): + """ + Creates an empty repo on the given directory -- basically a + `.dvc` directory with subdirectories for configuration and cache. + + It should be tracked by a SCM or use the `--no-scm` flag. + + If the given directory is not empty, you must use the `--force` + flag to override it. + + Args: + root_dir: Path to repo's root directory. + + Returns: + Repo instance. + + Raises: + KeyError: Raises an exception. + """ + root_dir = os.path.abspath(root_dir) + dvc_dir = os.path.join(root_dir, Repo.DVC_DIR) + scm = SCM(root_dir) + if type(scm) == Base and not no_scm: + raise InitError( + "{repo} is not tracked by any supported scm tool" + " (e.g. git). Use '--no-scm' if you don't want to use any scm.".format( + repo=root_dir + ) + ) + + if os.path.isdir(dvc_dir): + if not force: + raise InitError( + "'{repo}' exists. Use '-f' to force.".format( + repo=os.path.relpath(dvc_dir) + ) + ) + + shutil.rmtree(dvc_dir) + + os.mkdir(dvc_dir) + + config = Config.init(dvc_dir) + proj = Repo(root_dir) + + scm.add([config.config_file]) + + if scm.ignore_file: + scm.add([os.path.join(dvc_dir, scm.ignore_file)]) + logger.info("\nYou can now commit the changes to git.\n") + + _welcome_message() + + return proj diff --git a/dvc/repo/install.py b/dvc/repo/install.py new file mode 100644 index 000000000..cff23a15f --- /dev/null +++ b/dvc/repo/install.py @@ -0,0 +1,2 @@ +def install(self): + self.scm.install() diff --git a/dvc/repo/lock.py b/dvc/repo/lock.py new file mode 100644 index 000000000..618fb7aa2 --- /dev/null +++ b/dvc/repo/lock.py @@ -0,0 +1,8 @@ +def lock(self, target, unlock=False): + from dvc.stage import Stage + + stage = Stage.load(self, target) + stage.locked = False if unlock else True + stage.dump() + + return stage diff --git a/dvc/repo/metrics/__init__.py b/dvc/repo/metrics/__init__.py new file mode 100644 index 000000000..8904a00b9 --- /dev/null +++ b/dvc/repo/metrics/__init__.py @@ -0,0 +1,23 @@ +class Metrics(object): + def __init__(self, repo): + self.repo = repo + + def add(self, *args, **kwargs): + from dvc.repo.metrics.add import add + + return add(self.repo, *args, **kwargs) + + def modify(self, *args, **kwargs): + from dvc.repo.metrics.modify import modify + + return modify(self.repo, *args, **kwargs) + + def show(self, *args, **kwargs): + from dvc.repo.metrics.show import show + + return show(self.repo, *args, **kwargs) + + def remove(self, *args, **kwargs): + from dvc.repo.metrics.remove import remove + + return remove(self.repo, *args, **kwargs) diff --git a/dvc/repo/metrics/add.py b/dvc/repo/metrics/add.py new file mode 100644 index 000000000..45a172dac --- /dev/null +++ b/dvc/repo/metrics/add.py @@ -0,0 +1,7 @@ +from dvc.repo.metrics.modify import modify + + +def add(repo, path, typ=None, xpath=None): + if not typ: + typ = "raw" + modify(repo, path, typ, xpath) diff --git a/dvc/repo/metrics/modify.py b/dvc/repo/metrics/modify.py new file mode 100644 index 000000000..a45a93d68 --- /dev/null +++ b/dvc/repo/metrics/modify.py @@ -0,0 +1,64 @@ +from __future__ import unicode_literals + +import os + +from dvc.exceptions import DvcException, OutputDuplicationError + + +def find_output_by_path(repo, path, outs=None, recursive=False): + if not outs: + astages = repo.active_stages() + outs = [out for stage in astages for out in stage.outs] + + abs_path = os.path.abspath(path) + if os.path.isdir(abs_path) and recursive: + matched = [ + out + for out in outs + if os.path.abspath(out.path).startswith(abs_path) + ] + else: + matched = [out for out in outs if out.path == abs_path] + stages = [out.stage.relpath for out in matched] + if len(stages) > 1: + raise OutputDuplicationError(path, stages) + + return matched if matched else [] + + +def modify(repo, path, typ=None, xpath=None, delete=False): + outs = find_output_by_path(repo, path) + + if not outs: + msg = "unable to find file '{}' in the pipeline".format(path) + raise DvcException(msg) + + if len(outs) != 1: + msg = ( + "-R not yet supported for metrics modify. " + "Make sure only one metric is referred to by '{}'".format(path) + ) + raise DvcException(msg) + + out = outs[0] + + if out.scheme != "local": + msg = "output '{}' scheme '{}' is not supported for metrics" + raise DvcException(msg.format(out.path, out.path_info["scheme"])) + + if typ: + if not isinstance(out.metric, dict): + out.metric = {} + out.metric[out.PARAM_METRIC_TYPE] = typ + + if xpath: + if not isinstance(out.metric, dict): + out.metric = {} + out.metric[out.PARAM_METRIC_XPATH] = xpath + + if delete: + out.metric = None + + out._verify_metric() + + out.stage.dump() diff --git a/dvc/repo/metrics/remove.py b/dvc/repo/metrics/remove.py new file mode 100644 index 000000000..4576c447b --- /dev/null +++ b/dvc/repo/metrics/remove.py @@ -0,0 +1,5 @@ +from dvc.repo.metrics.modify import modify + + +def remove(repo, path): + modify(repo, path, delete=True) diff --git a/dvc/repo/metrics/show.py b/dvc/repo/metrics/show.py new file mode 100644 index 000000000..84f8b5cc1 --- /dev/null +++ b/dvc/repo/metrics/show.py @@ -0,0 +1,150 @@ +from __future__ import unicode_literals + +import os +import csv +import json +from jsonpath_rw import parse + +import dvc.logger as logger +from dvc.exceptions import OutputDuplicationError, DvcException +from dvc.repo.metrics.modify import find_output_by_path +from dvc.utils.compat import str, builtin_str, open + + +def _read_metric_json(fd, json_path): + parser = parse(json_path) + return [x.value for x in parser.find(json.load(fd))] + + +def _do_read_metric_xsv(reader, row, col): + if col is not None and row is not None: + return [reader[row][col]] + elif col is not None: + return [r[col] for r in reader] + elif row is not None: + return reader[row] + return None + + +def _read_metric_hxsv(fd, hxsv_path, delimiter): + col, row = hxsv_path.split(",") + row = int(row) + reader = list(csv.DictReader(fd, delimiter=builtin_str(delimiter))) + return _do_read_metric_xsv(reader, row, col) + + +def _read_metric_xsv(fd, xsv_path, delimiter): + col, row = xsv_path.split(",") + row = int(row) + col = int(col) + reader = list(csv.reader(fd, delimiter=builtin_str(delimiter))) + return _do_read_metric_xsv(reader, row, col) + + +def _read_metric(path, typ=None, xpath=None): + ret = None + + if not os.path.exists(path): + return ret + + try: + with open(path, "r") as fd: + if typ == "json": + ret = _read_metric_json(fd, xpath) + elif typ == "csv": + ret = _read_metric_xsv(fd, xpath, ",") + elif typ == "tsv": + ret = _read_metric_xsv(fd, xpath, "\t") + elif typ == "hcsv": + ret = _read_metric_hxsv(fd, xpath, ",") + elif typ == "htsv": + ret = _read_metric_hxsv(fd, xpath, "\t") + else: + ret = fd.read().strip() + except Exception: + logger.error("unable to read metric in '{}'".format(path)) + + return ret + + +def show( + self, + path=None, + typ=None, + xpath=None, + all_branches=False, + all_tags=False, + recursive=False, +): + res = {} + for branch in self.scm.brancher( + all_branches=all_branches, all_tags=all_tags + ): + astages = self.active_stages() + outs = [out for stage in astages for out in stage.outs] + + if path: + outs = find_output_by_path( + self, path, outs=outs, recursive=recursive + ) + + metrics = filter(lambda o: o.metric, outs) + stages = set() + entries = [] + + for o in metrics: + if not typ and isinstance(o.metric, dict): + t = o.metric.get(o.PARAM_METRIC_TYPE, typ) + x = o.metric.get(o.PARAM_METRIC_XPATH, xpath) + else: + t = typ + x = xpath + entries.append((o.path, t, x)) + stages.add(o.stage.path) + + if path and not entries: + if os.path.isdir(path): + logger.warning( + "Path '{path}' is a directory. " + "Consider running with '-R'.".format(path=path) + ) + return {} + + for fname, t, x in entries: + if stages: + for stage in stages: + self.checkout(stage, force=True) + + rel = os.path.relpath(fname) + metric = _read_metric(fname, typ=t, xpath=x) + if not metric: + continue + + if branch not in res: + res[branch] = {} + + res[branch][rel] = metric + + for branch, val in res.items(): + if all_branches or all_tags: + logger.info("{}:".format(branch)) + for fname, metric in val.items(): + logger.info("\t{}: {}".format(fname, metric)) + + if res: + return res + + if path and os.path.isdir(path): + return res + + if path: + msg = "file '{}' does not exist, not a metric file or is malformed".format( + path + ) + else: + msg = ( + "no metric files in this repository." + " use 'dvc metrics add' to add a metric file to track." + ) + + raise DvcException(msg) diff --git a/dvc/repo/move.py b/dvc/repo/move.py new file mode 100644 index 000000000..e35e16ef8 --- /dev/null +++ b/dvc/repo/move.py @@ -0,0 +1,75 @@ +import os + +from dvc.exceptions import MoveNotDataSourceError, DvcException + + +def _expand_target_path(from_path, to_path): + if os.path.isdir(to_path) and not os.path.isdir(from_path): + return os.path.join(to_path, os.path.basename(from_path)) + return to_path + + +def move(self, from_path, to_path): + """ + Renames an output file and modifies the stage associated + to reflect the change on the pipeline. + + If the output has the same name as its stage, it would + also rename the corresponding stage file. + + E.g. + Having: (hello, hello.dvc) + + $ dvc move hello greetings + + Result: (greeting, greeting.dvc) + + It only works with outputs generated by `add` or `import`, + also known as data sources. + """ + import dvc.output as Output + from dvc.stage import Stage + + from_out = Output.loads_from(Stage(self, cwd=os.curdir), [from_path])[0] + + to_path = _expand_target_path(from_path, to_path) + + try: + stage, out = next( + (stage, out) + for stage in self.stages() + for out in stage.outs + if from_out.path == out.path + ) + except StopIteration: + raise DvcException( + "unable to find stage file with output '{path}'".format( + path=from_path + ) + ) + + if not stage.is_data_source: + raise MoveNotDataSourceError(stage.relpath) + + stage_name = os.path.splitext(os.path.basename(stage.path))[0] + from_name = os.path.basename(from_out.path) + if stage_name == from_name: + os.unlink(stage.path) + + stage.path = os.path.join( + os.path.dirname(to_path), + os.path.basename(to_path) + Stage.STAGE_FILE_SUFFIX, + ) + + stage.cwd = os.path.join(self.root_dir, os.path.dirname(to_path)) + + to_out = Output.loads_from( + stage, [os.path.basename(to_path)], out.cache, out.metric + )[0] + + with self.state: + out.move(to_out) + + stage.dump() + + self.remind_to_git_add() diff --git a/dvc/repo/pull.py b/dvc/repo/pull.py new file mode 100644 index 000000000..e0f94f4d1 --- /dev/null +++ b/dvc/repo/pull.py @@ -0,0 +1,28 @@ +from __future__ import unicode_literals + + +def pull( + self, + target=None, + jobs=1, + remote=None, + all_branches=False, + show_checksums=False, + with_deps=False, + all_tags=False, + force=False, + recursive=False, +): + self.fetch( + target, + jobs, + remote=remote, + all_branches=all_branches, + all_tags=all_tags, + show_checksums=show_checksums, + with_deps=with_deps, + recursive=recursive, + ) + self.checkout( + target=target, with_deps=with_deps, force=force, recursive=recursive + ) diff --git a/dvc/repo/push.py b/dvc/repo/push.py new file mode 100644 index 000000000..96b7263a5 --- /dev/null +++ b/dvc/repo/push.py @@ -0,0 +1,28 @@ +from __future__ import unicode_literals + + +def push( + self, + target=None, + jobs=1, + remote=None, + all_branches=False, + show_checksums=False, + with_deps=False, + all_tags=False, + recursive=False, +): + with self.state: + used = self.used_cache( + target, + all_branches=all_branches, + all_tags=all_tags, + with_deps=with_deps, + force=True, + remote=remote, + jobs=jobs, + recursive=recursive, + )["local"] + self.cloud.push( + used, jobs, remote=remote, show_checksums=show_checksums + ) diff --git a/dvc/repo/remove.py b/dvc/repo/remove.py new file mode 100644 index 000000000..2977ee82c --- /dev/null +++ b/dvc/repo/remove.py @@ -0,0 +1,10 @@ +def remove(self, target, outs_only=False): + from dvc.stage import Stage + + stage = Stage.load(self, target) + if outs_only: + stage.remove_outs() + else: + stage.remove() + + return stage diff --git a/dvc/repo/reproduce.py b/dvc/repo/reproduce.py new file mode 100644 index 000000000..30464caea --- /dev/null +++ b/dvc/repo/reproduce.py @@ -0,0 +1,134 @@ +from __future__ import unicode_literals + +import os + +import dvc.logger as logger +from dvc.exceptions import ReproductionError + + +def _reproduce_stage(stages, node, force, dry, interactive): + stage = stages[node] + + if stage.locked: + logger.warning( + "DVC file '{path}' is locked. Its dependencies are" + " not going to be reproduced.".format(path=stage.relpath) + ) + + stage = stage.reproduce(force=force, dry=dry, interactive=interactive) + if not stage: + return [] + + if not dry: + stage.dump() + + return [stage] + + +def reproduce( + self, + target=None, + recursive=True, + force=False, + dry=False, + interactive=False, + pipeline=False, + all_pipelines=False, + ignore_build_cache=False, +): + from dvc.stage import Stage + + if not target and not all_pipelines: + raise ValueError() + + if not interactive: + config = self.config + core = config.config[config.SECTION_CORE] + interactive = core.get(config.SECTION_CORE_INTERACTIVE, False) + + targets = [] + if pipeline or all_pipelines: + if pipeline: + stage = Stage.load(self, target) + node = os.path.relpath(stage.path, self.root_dir) + pipelines = [self._get_pipeline(node)] + else: + pipelines = self.pipelines() + + for G in pipelines: + for node in G.nodes(): + if G.in_degree(node) == 0: + targets.append(os.path.join(self.root_dir, node)) + else: + targets.append(target) + + self.files_to_git_add = [] + + ret = [] + with self.state: + for target in targets: + stages = _reproduce( + self, + target, + recursive=recursive, + force=force, + dry=dry, + interactive=interactive, + ignore_build_cache=ignore_build_cache, + ) + ret.extend(stages) + + self.remind_to_git_add() + + return ret + + +def _reproduce( + self, + target, + recursive=True, + force=False, + dry=False, + interactive=False, + ignore_build_cache=False, +): + import networkx as nx + from dvc.stage import Stage + + stage = Stage.load(self, target) + G = self.graph()[1] + stages = nx.get_node_attributes(G, "stage") + node = os.path.relpath(stage.path, self.root_dir) + + if recursive: + ret = _reproduce_stages( + G, stages, node, force, dry, interactive, ignore_build_cache + ) + else: + ret = _reproduce_stage(stages, node, force, dry, interactive) + + return ret + + +def _reproduce_stages( + G, stages, node, force, dry, interactive, ignore_build_cache +): + import networkx as nx + + result = [] + for n in nx.dfs_postorder_nodes(G, node): + try: + ret = _reproduce_stage(stages, n, force, dry, interactive) + + if len(ret) == 0 and ignore_build_cache: + # NOTE: we are walking our pipeline from the top to the + # bottom. If one stage is changed, it will be reproduced, + # which tells us that we should force reproducing all of + # the other stages down below, even if their direct + # dependencies didn't change. + force = True + + result += ret + except Exception as ex: + raise ReproductionError(stages[n].relpath, ex) + return result diff --git a/dvc/repo/run.py b/dvc/repo/run.py new file mode 100644 index 000000000..a2c142655 --- /dev/null +++ b/dvc/repo/run.py @@ -0,0 +1,62 @@ +import os + + +def run( + self, + cmd=None, + deps=None, + outs=None, + outs_no_cache=None, + metrics=None, + metrics_no_cache=None, + fname=None, + cwd=os.curdir, + no_exec=False, + overwrite=False, + ignore_build_cache=False, + remove_outs=False, +): + from dvc.stage import Stage + + if outs is None: + outs = [] + if deps is None: + deps = [] + if outs_no_cache is None: + outs_no_cache = [] + if metrics is None: + metrics = [] + if metrics_no_cache is None: + metrics_no_cache = [] + + with self.state: + stage = Stage.create( + repo=self, + fname=fname, + cmd=cmd, + cwd=cwd, + outs=outs, + outs_no_cache=outs_no_cache, + metrics=metrics, + metrics_no_cache=metrics_no_cache, + deps=deps, + overwrite=overwrite, + ignore_build_cache=ignore_build_cache, + remove_outs=remove_outs, + ) + + if stage is None: + return None + + self.check_dag(self.stages() + [stage]) + + self.files_to_git_add = [] + with self.state: + if not no_exec: + stage.run() + + stage.dump() + + self.remind_to_git_add() + + return stage diff --git a/dvc/repo/status.py b/dvc/repo/status.py new file mode 100644 index 000000000..b341d78e0 --- /dev/null +++ b/dvc/repo/status.py @@ -0,0 +1,89 @@ +from __future__ import unicode_literals + +import dvc.logger as logger +import dvc.remote.base as cloud + + +def _local_status(self, target=None, with_deps=False): + status = {} + + if target: + stages = self.collect(target, with_deps=with_deps) + else: + stages = self.active_stages() + + for stage in stages: + if stage.locked: + logger.warning( + "DVC file '{path}' is locked. Its dependencies are" + " not going to be shown in the status output.".format( + path=stage.relpath + ) + ) + + status.update(stage.status()) + + return status + + +def _cloud_status( + self, + target=None, + jobs=1, + remote=None, + show_checksums=False, + all_branches=False, + with_deps=False, + all_tags=False, +): + used = self.used_cache( + target, + all_branches=all_branches, + all_tags=all_tags, + with_deps=with_deps, + force=True, + remote=remote, + jobs=jobs, + )["local"] + + ret = {} + status_info = self.cloud.status( + used, jobs, remote=remote, show_checksums=show_checksums + ) + for md5, info in status_info.items(): + name = info["name"] + status = info["status"] + if status == cloud.STATUS_OK: + continue + + prefix_map = {cloud.STATUS_DELETED: "deleted", cloud.STATUS_NEW: "new"} + + ret[name] = prefix_map[status] + + return ret + + +def status( + self, + target=None, + jobs=1, + cloud=False, + remote=None, + show_checksums=False, + all_branches=False, + with_deps=False, + all_tags=False, +): + with self.state: + if cloud: + return _cloud_status( + self, + target, + jobs, + remote=remote, + show_checksums=show_checksums, + all_branches=all_branches, + with_deps=with_deps, + all_tags=all_tags, + ) + return _local_status(self, target, with_deps=with_deps) diff --git a/dvc/repo/unprotect.py b/dvc/repo/unprotect.py new file mode 100644 index 000000000..ddb69809d --- /dev/null +++ b/dvc/repo/unprotect.py @@ -0,0 +1,48 @@ +from __future__ import unicode_literals + +import os +import stat +import uuid + +import dvc.logger as logger +from dvc.system import System +from dvc.utils import copyfile, move, remove +from dvc.exceptions import DvcException + + +def _unprotect_file(path): + if System.is_symlink(path) or System.is_hardlink(path): + logger.debug("Unprotecting '{}'".format(path)) + + tmp = os.path.join(os.path.dirname(path), "." + str(uuid.uuid4())) + move(path, tmp) + + copyfile(tmp, path) + + remove(tmp) + else: + logger.debug( + "Skipping copying for '{}', since it is not " + "a symlink or a hardlink.".format(path) + ) + + os.chmod(path, os.stat(path).st_mode | stat.S_IWRITE) + + +def _unprotect_dir(path): + for root, dirs, files in os.walk(str(path)): + for f in files: + path = os.path.join(root, f) + _unprotect_file(path) + + +def unprotect(path): + if not os.path.exists(path): + raise DvcException( + "can't unprotect non-existing data '{}'".format(path) + ) + + if os.path.isdir(path): + _unprotect_dir(path) + else: + _unprotect_file(path) diff --git a/dvc/scm/__init__.py b/dvc/scm/__init__.py index ef5d66b77..3ebd2e67b 100644 --- a/dvc/scm/__init__.py +++ b/dvc/scm/__init__.py @@ -6,19 +6,18 @@ from dvc.scm.base import Base from dvc.scm.git import Git -def SCM(root_dir, project=None): # pylint: disable=invalid-name - """Returns SCM instance that corresponds to a project at the specified +def SCM(root_dir, repo=None): # pylint: disable=invalid-name + """Returns SCM instance that corresponds to a repo at the specified path. Args: - root_dir (str): path to a root directory of the project. - project (dvc.project.Project): dvc project instance that root_dir - belongs to. + root_dir (str): path to a root directory of the repo. + repo (dvc.repo.Repo): dvc repo instance that root_dir belongs to. Returns: dvc.scm.base.Base: SCM instance. """ if Git.is_repo(root_dir) or Git.is_submodule(root_dir): - return Git(root_dir, project=project) + return Git(root_dir, repo=repo) - return Base(root_dir, project=project) + return Base(root_dir, repo=repo) diff --git a/dvc/scm/base.py b/dvc/scm/base.py index b02309fd1..59d082a1e 100644 --- a/dvc/scm/base.py +++ b/dvc/scm/base.py @@ -20,8 +20,8 @@ class FileNotInRepoError(DvcException): class Base(object): """Base class for source control management driver implementations.""" - def __init__(self, root_dir=os.curdir, project=None): - self.project = project + def __init__(self, root_dir=os.curdir, repo=None): + self.repo = repo self.root_dir = root_dir def __repr__(self): @@ -126,16 +126,16 @@ class Base(object): return False def active_branch(self): # pylint: disable=no-self-use - """Returns current branch in the project.""" + """Returns current branch in the repo.""" return "" def list_branches(self): # pylint: disable=no-self-use - """Returns a list of available branches in the project.""" + """Returns a list of available branches in the repo.""" return [] def list_tags(self): # pylint: disable=no-self-use - """Returns a list of available tags in the project.""" + """Returns a list of available tags in the repo.""" return [] def install(self): - """Adds dvc commands to SCM hooks for the project.""" + """Adds dvc commands to SCM hooks for the repo.""" diff --git a/dvc/scm/git.py b/dvc/scm/git.py index 2a7cff144..0c9ea835c 100644 --- a/dvc/scm/git.py +++ b/dvc/scm/git.py @@ -21,14 +21,14 @@ class Git(Base): GITIGNORE = ".gitignore" GIT_DIR = ".git" - def __init__(self, root_dir=os.curdir, project=None): - super(Git, self).__init__(root_dir, project=project) + def __init__(self, root_dir=os.curdir, repo=None): + super(Git, self).__init__(root_dir, repo=repo) import git from git.exc import InvalidGitRepositoryError try: - self.repo = git.Repo(root_dir) + self.git = git.Repo(root_dir) except InvalidGitRepositoryError: msg = "{} is not a git repository" raise SCMError(msg.format(root_dir)) @@ -37,7 +37,7 @@ class Git(Base): # http://pyinstaller.readthedocs.io/en/stable/runtime-information.html env = fix_env(None) libpath = env.get("LD_LIBRARY_PATH", None) - self.repo.git.update_environment(LD_LIBRARY_PATH=libpath) + self.git.git.update_environment(LD_LIBRARY_PATH=libpath) @staticmethod def is_repo(root_dir): @@ -53,7 +53,7 @@ class Git(Base): @property def dir(self): - return self.repo.git_dir + return self.git.git_dir @property def ignore_file(self): @@ -94,8 +94,8 @@ class Git(Base): with open(gitignore, "a") as fobj: fobj.write(content) - if self.project is not None: - self.project.files_to_git_add.append(os.path.relpath(gitignore)) + if self.repo is not None: + self.repo.files_to_git_add.append(os.path.relpath(gitignore)) def ignore_remove(self, path): entry, gitignore = self._get_gitignore(path) @@ -111,14 +111,14 @@ class Git(Base): with open(gitignore, "w") as fobj: fobj.writelines(filtered) - if self.project is not None: - self.project.files_to_git_add.append(os.path.relpath(gitignore)) + if self.repo is not None: + self.repo.files_to_git_add.append(os.path.relpath(gitignore)) def add(self, paths): # NOTE: GitPython is not currently able to handle index version >= 3. # See https://github.com/iterative/dvc/issues/610 for more details. try: - self.repo.index.add(paths) + self.git.index.add(paths) except AssertionError: msg = ( "failed to add '{}' to git. You can add those files" @@ -130,35 +130,35 @@ class Git(Base): logger.error(msg) def commit(self, msg): - self.repo.index.commit(msg) + self.git.index.commit(msg) def checkout(self, branch, create_new=False): if create_new: - self.repo.git.checkout("HEAD", b=branch) + self.git.git.checkout("HEAD", b=branch) else: - self.repo.git.checkout(branch) + self.git.git.checkout(branch) def branch(self, branch): - self.repo.git.branch(branch) + self.git.git.branch(branch) def tag(self, tag): - self.repo.git.tag(tag) + self.git.git.tag(tag) def untracked_files(self): - files = self.repo.untracked_files - return [os.path.join(self.repo.working_dir, fname) for fname in files] + files = self.git.untracked_files + return [os.path.join(self.git.working_dir, fname) for fname in files] def is_tracked(self, path): - return len(self.repo.git.ls_files(path)) != 0 + return len(self.git.git.ls_files(path)) != 0 def active_branch(self): - return self.repo.active_branch.name + return self.git.active_branch.name def list_branches(self): - return [h.name for h in self.repo.heads] + return [h.name for h in self.git.heads] def list_tags(self): - return [t.name for t in self.repo.tags] + return [t.name for t in self.git.tags] def _install_hook(self, name, cmd): hook = os.path.join(self.root_dir, self.GIT_DIR, "hooks", name) diff --git a/dvc/stage.py b/dvc/stage.py index 8e885befc..ff3b61692 100644 --- a/dvc/stage.py +++ b/dvc/stage.py @@ -63,7 +63,7 @@ class StageFileBadNameError(DvcException): class StageBadCwdError(DvcException): def __init__(self, cwd): - msg = "stage cwd '{}' is outside of the current dvc project" + msg = "stage cwd '{}' is outside of the current dvc repo" super(StageBadCwdError, self).__init__(msg.format(cwd)) @@ -112,7 +112,7 @@ class Stage(object): def __init__( self, - project, + repo, path=None, cmd=None, cwd=os.curdir, @@ -126,7 +126,7 @@ class Stage(object): if outs is None: outs = [] - self.project = project + self.repo = repo self.path = path self.cmd = cmd self.cwd = cwd @@ -245,7 +245,7 @@ class Stage(object): for out in self.outs: if out.scheme != "local" or not out.exists: continue - self.project.unprotect(out.path) + self.repo.unprotect(out.path) def remove(self): self.remove_outs(ignore_remove=True) @@ -306,9 +306,9 @@ class Stage(object): return (fname, cwd) @staticmethod - def _check_inside_project(project, cwd): - assert project is not None - proj_dir = os.path.realpath(project.root_dir) + def _check_inside_repo(repo, cwd): + assert repo is not None + proj_dir = os.path.realpath(repo.root_dir) if not os.path.realpath(cwd).startswith(proj_dir): raise StageBadCwdError(cwd) @@ -320,7 +320,7 @@ class Stage(object): from dvc.remote.local import RemoteLOCAL from dvc.remote.s3 import RemoteS3 - old = Stage.load(self.project, self.path) + old = Stage.load(self.repo, self.path) if old._changed_outs(): return False @@ -345,7 +345,7 @@ class Stage(object): @staticmethod def create( - project=None, + repo=None, cmd=None, deps=None, outs=None, @@ -372,7 +372,7 @@ class Stage(object): if metrics_no_cache is None: metrics_no_cache = [] - stage = Stage(project=project, cwd=cwd, cmd=cmd, locked=locked) + stage = Stage(repo=repo, cwd=cwd, cmd=cmd, locked=locked) stage.outs = output.loads_from(stage, outs, use_cache=True) stage.outs += output.loads_from( @@ -397,7 +397,7 @@ class Stage(object): fname, cwd = Stage._stage_fname_cwd(fname, cwd, stage.outs, add=add) - Stage._check_inside_project(project, cwd) + Stage._check_inside_repo(repo, cwd) cwd = os.path.abspath(cwd) path = os.path.join(cwd, fname) @@ -446,7 +446,7 @@ class Stage(object): raise StageFileDoesNotExistError(fname) @staticmethod - def load(project, fname): + def load(repo, fname): Stage._check_file_exists(fname) Stage._check_dvc_filename(fname) @@ -459,7 +459,7 @@ class Stage(object): Stage.validate(d, fname=os.path.relpath(fname)) stage = Stage( - project=project, + repo=repo, path=os.path.abspath(fname), cwd=os.path.dirname(os.path.abspath(fname)), cmd=d.get(Stage.PARAM_CMD), @@ -499,7 +499,7 @@ class Stage(object): with open(fname, "w") as fd: yaml.safe_dump(self.dumpd(), fd, default_flow_style=False) - self.project.files_to_git_add.append(os.path.relpath(fname)) + self.repo.files_to_git_add.append(os.path.relpath(fname)) def _compute_md5(self): from dvc.output.local import OutputLOCAL diff --git a/dvc/state.py b/dvc/state.py index 21bce0cfb..bf2ad8068 100644 --- a/dvc/state.py +++ b/dvc/state.py @@ -24,7 +24,7 @@ class StateVersionTooNewError(DvcException): "you are using an old version '{dvc_version}' of dvc that is " "using state file version '{expected}' which is not compatible " "with the state file version '{actual}' that is used in this " - "project. Please upgrade right now!".format( + "repo. Please upgrade right now!".format( dvc_version=dvc_version, expected=expected, actual=actual ) ) @@ -34,8 +34,7 @@ class State(object): # pylint: disable=too-many-instance-attributes """Class for the state database. Args: - project (dvc.project.Project): project instance that this state - belongs to. + repo (dvc.repo.Repo): repo instance that this state belongs to. config (configobj.ConfigObj): config for the state. Raises: @@ -71,10 +70,10 @@ class State(object): # pylint: disable=too-many-instance-attributes MAX_INT = 2 ** 63 - 1 MAX_UINT = 2 ** 64 - 2 - def __init__(self, project, config): - self.project = project - self.dvc_dir = project.dvc_dir - self.root_dir = project.root_dir + def __init__(self, repo, config): + self.repo = repo + self.dvc_dir = repo.dvc_dir + self.root_dir = repo.root_dir self.row_limit = 100 self.row_cleanup_quota = 50 @@ -112,7 +111,7 @@ class State(object): # pylint: disable=too-many-instance-attributes def _collect(self, path): if os.path.isdir(path): - return self.project.cache.local.collect_dir_cache(path) + return self.repo.cache.local.collect_dir_cache(path) return (file_md5(path)[0], None) def changed(self, path, md5): @@ -420,7 +419,7 @@ class State(object): # pylint: disable=too-many-instance-attributes """ md5, info = self._do_update(path) if not info: - info = self.project.cache.local.load_dir_cache(md5) + info = self.repo.cache.local.load_dir_cache(md5) return (md5, info) def update_link(self, path): diff --git a/tests/basic_env.py b/tests/basic_env.py index e308b9247..ab48e2c73 100644 --- a/tests/basic_env.py +++ b/tests/basic_env.py @@ -6,7 +6,7 @@ from git.exc import GitCommandNotFound from unittest import TestCase import dvc.logger as logger -from dvc.project import Project +from dvc.repo import Repo as DvcRepo class TestDir(TestCase): @@ -111,5 +111,5 @@ class TestGitSubmodule(TestGit): class TestDvc(TestGit): def setUp(self): super(TestDvc, self).setUp() - self.dvc = Project.init(self._root_dir) + self.dvc = DvcRepo.init(self._root_dir) logger.be_verbose() diff --git a/tests/test_add.py b/tests/test_add.py index 0b396750a..a4ca03c32 100644 --- a/tests/test_add.py +++ b/tests/test_add.py @@ -11,7 +11,7 @@ from dvc.utils import file_md5 from dvc.stage import Stage from dvc.exceptions import DvcException from dvc.output.base import OutputAlreadyTrackedError -from dvc.project import Project +from dvc.repo import Repo as DvcRepo from tests.basic_env import TestDvc @@ -172,7 +172,7 @@ class TestAddLocalRemoteFile(TestDvc): ret = main(["remote", "add", remote, cwd]) self.assertEqual(ret, 0) - self.dvc = Project() + self.dvc = DvcRepo() foo = "remote://{}/{}".format(remote, self.FOO) ret = main(["add", foo]) diff --git a/tests/test_cache.py b/tests/test_cache.py index b3bab74f7..5fb620225 100644 --- a/tests/test_cache.py +++ b/tests/test_cache.py @@ -142,7 +142,7 @@ class TestCmdCacheDir(TestDvc): ret = main(["cache", "dir", dname]) self.assertEqual(ret, 0) - # NOTE: we are in the project's root and config is in .dvc/, so + # NOTE: we are in the repo's root and config is in .dvc/, so # dir path written to config should be just one level above. rel = os.path.join("..", dname) config = configobj.ConfigObj(self.dvc.config.config_file) diff --git a/tests/test_checkout.py b/tests/test_checkout.py index 4f82a3bd3..37effe9a9 100644 --- a/tests/test_checkout.py +++ b/tests/test_checkout.py @@ -6,7 +6,7 @@ import filecmp import collections from dvc.main import main -from dvc.project import Project +from dvc.repo import Repo as DvcRepo from dvc.system import System from tests.basic_env import TestDvc from tests.test_repro import TestRepro @@ -73,7 +73,7 @@ class TestCheckoutCorruptedCacheDir(TestDvc): ret = main(["config", "cache.type", "copy"]) self.assertEqual(ret, 0) - self.dvc = Project(".") + self.dvc = DvcRepo(".") stages = self.dvc.add(self.DATA_DIR) self.assertEqual(len(stages), 1) self.assertEqual(len(stages[0].outs), 1) diff --git a/tests/test_cli.py b/tests/test_cli.py index 5ecb9e2cd..e823fa810 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -137,13 +137,13 @@ class TestGC(TestDvc): self.assertIsInstance(args.func(args), CmdGC) -class TestGCMultipleProjects(TestDvc): +class TestGCMultipleDvcRepos(TestDvc): def test(self): args = parse_args(["gc", "-p", "/tmp/asdf", "/tmp/xyz"]) self.assertIsInstance(args.func(args), CmdGC) - self.assertEqual(args.projects, ["/tmp/asdf", "/tmp/xyz"]) + self.assertEqual(args.repos, ["/tmp/asdf", "/tmp/xyz"]) class TestConfig(TestDvc): diff --git a/tests/test_data_cloud.py b/tests/test_data_cloud.py index baf3a938a..d7208ba96 100644 --- a/tests/test_data_cloud.py +++ b/tests/test_data_cloud.py @@ -258,7 +258,7 @@ class TestDataCloudBase(TestDvc): info_dir = stage_dir.outs[0].dumpd() md5_dir = info_dir["md5"] - with self.cloud.project.state: + with self.cloud.repo.state: # Check status status = self.cloud.status([info], show_checksums=True) expected = {md5: {"name": md5, "status": STATUS_NEW}} diff --git a/tests/test_gc.py b/tests/test_gc.py index cb7565852..7d07c8a6b 100644 --- a/tests/test_gc.py +++ b/tests/test_gc.py @@ -4,7 +4,7 @@ import configobj from git import Repo from dvc.main import main -from dvc.project import Project +from dvc.repo import Repo as DvcRepo from tests.basic_env import TestDvc from tests.basic_env import TestDir @@ -104,7 +104,7 @@ class TestGCBranchesTags(TestDvc): self._check_cache(1) -class TestGCMultipleProjects(TestDvc): +class TestGCMultipleDvcRepos(TestDvc): def _check_cache(self, num): total = 0 for root, dirs, files in os.walk(os.path.join(".dvc", "cache")): @@ -112,10 +112,10 @@ class TestGCMultipleProjects(TestDvc): self.assertEqual(total, num) def setUp(self): - super(TestGCMultipleProjects, self).setUp() + super(TestGCMultipleDvcRepos, self).setUp() self.additional_path = TestDir.mkdtemp() self.additional_git = Repo.init(self.additional_path) - self.additional_dvc = Project.init(self.additional_path) + self.additional_dvc = DvcRepo.init(self.additional_path) cache_path = os.path.join(self._root_dir, ".dvc", "cache") config_path = os.path.join( @@ -126,14 +126,14 @@ class TestGCMultipleProjects(TestDvc): cfg["cache"] = {"dir": cache_path} cfg.write() - self.additional_dvc = Project(self.additional_path) + self.additional_dvc = DvcRepo(self.additional_path) def test(self): # ADD FILE ONLY IN MAIN PROJECT fname = "only_in_first" with open(fname, "w+") as fobj: - fobj.write("only in main project") + fobj.write("only in main repo") stages = self.dvc.add(fname) self.assertEqual(len(stages), 1) @@ -141,7 +141,7 @@ class TestGCMultipleProjects(TestDvc): # ADD FILE IN MAIN PROJECT THAT IS ALSO IN SECOND PROJECT fname = "in_both" with open(fname, "w+") as fobj: - fobj.write("in both projects") + fobj.write("in both repos") stages = self.dvc.add(fname) self.assertEqual(len(stages), 1) @@ -151,7 +151,7 @@ class TestGCMultipleProjects(TestDvc): # ADD FILE ONLY IN SECOND PROJECT fname = os.path.join(self.additional_path, "only_in_second") with open(fname, "w+") as fobj: - fobj.write("only in additional project") + fobj.write("only in additional repo") stages = self.additional_dvc.add(fname) self.assertEqual(len(stages), 1) @@ -159,7 +159,7 @@ class TestGCMultipleProjects(TestDvc): # ADD FILE IN SECOND PROJECT THAT IS ALSO IN MAIN PROJECT fname = os.path.join(self.additional_path, "in_both") with open(fname, "w+") as fobj: - fobj.write("in both projects") + fobj.write("in both repos") stages = self.additional_dvc.add(fname) self.assertEqual(len(stages), 1) @@ -168,7 +168,7 @@ class TestGCMultipleProjects(TestDvc): self._check_cache(3) - self.dvc.gc(projects=[self.additional_path]) + self.dvc.gc(repos=[self.additional_path]) self._check_cache(3) self.dvc.gc() diff --git a/tests/test_init.py b/tests/test_init.py index 7c796e1e2..1acdf06bb 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -1,18 +1,19 @@ import os from dvc.main import main -from dvc.project import Project, InitError +from dvc.repo import Repo as DvcRepo +from dvc.exceptions import InitError from tests.basic_env import TestGit, TestDir class TestInit(TestGit): def _test_init(self): - self.assertTrue(os.path.exists(Project.DVC_DIR)) - self.assertTrue(os.path.isdir(Project.DVC_DIR)) + self.assertTrue(os.path.exists(DvcRepo.DVC_DIR)) + self.assertTrue(os.path.isdir(DvcRepo.DVC_DIR)) def test_api(self): - Project.init() + DvcRepo.init() self._test_init() @@ -41,7 +42,7 @@ class TestDoubleInit(TestInit): class TestInitNoSCMFail(TestDir): def test_api(self): with self.assertRaises(InitError): - Project.init() + DvcRepo.init() def test_cli(self): ret = main(["init"]) @@ -50,11 +51,11 @@ class TestInitNoSCMFail(TestDir): class TestInitNoSCM(TestDir): def _test_init(self): - self.assertTrue(os.path.exists(Project.DVC_DIR)) - self.assertTrue(os.path.isdir(Project.DVC_DIR)) + self.assertTrue(os.path.exists(DvcRepo.DVC_DIR)) + self.assertTrue(os.path.isdir(DvcRepo.DVC_DIR)) def test_api(self): - Project.init(no_scm=True) + DvcRepo.init(no_scm=True) self._test_init() diff --git a/tests/test_metrics.py b/tests/test_metrics.py index 577db3337..24e699f79 100644 --- a/tests/test_metrics.py +++ b/tests/test_metrics.py @@ -1,7 +1,7 @@ import os import json -from dvc.project import Project +from dvc.repo import Repo as DvcRepo from dvc.main import main from dvc.exceptions import DvcException from tests.basic_env import TestDvc @@ -53,13 +53,13 @@ class TestMetrics(TestDvc): self.dvc.scm.checkout("master") def test(self): - ret = self.dvc.metrics_show("metric", all_branches=True) + ret = self.dvc.metrics.show("metric", all_branches=True) self.assertEqual(len(ret), 3) self.assertTrue(ret["foo"]["metric"] == "foo") self.assertTrue(ret["bar"]["metric"] == "bar") self.assertTrue(ret["baz"]["metric"] == "baz") - ret = self.dvc.metrics_show( + ret = self.dvc.metrics.show( "metric_json", typ="json", xpath="branch", all_branches=True ) self.assertEqual(len(ret), 3) @@ -67,7 +67,7 @@ class TestMetrics(TestDvc): self.assertTrue(ret["bar"]["metric_json"] == ["bar"]) self.assertTrue(ret["baz"]["metric_json"] == ["baz"]) - ret = self.dvc.metrics_show( + ret = self.dvc.metrics.show( "metric_tsv", typ="tsv", xpath="0,0", all_branches=True ) self.assertEqual(len(ret), 3) @@ -75,7 +75,7 @@ class TestMetrics(TestDvc): self.assertTrue(ret["bar"]["metric_tsv"] == ["bar"]) self.assertTrue(ret["baz"]["metric_tsv"] == ["baz"]) - ret = self.dvc.metrics_show( + ret = self.dvc.metrics.show( "metric_htsv", typ="htsv", xpath="branch,0", all_branches=True ) self.assertEqual(len(ret), 3) @@ -83,7 +83,7 @@ class TestMetrics(TestDvc): self.assertTrue(ret["bar"]["metric_htsv"] == ["bar"]) self.assertTrue(ret["baz"]["metric_htsv"] == ["baz"]) - ret = self.dvc.metrics_show( + ret = self.dvc.metrics.show( "metric_csv", typ="csv", xpath="0,0", all_branches=True ) self.assertEqual(len(ret), 3) @@ -91,7 +91,7 @@ class TestMetrics(TestDvc): self.assertTrue(ret["bar"]["metric_csv"] == ["bar"]) self.assertTrue(ret["baz"]["metric_csv"] == ["baz"]) - ret = self.dvc.metrics_show( + ret = self.dvc.metrics.show( "metric_hcsv", typ="hcsv", xpath="branch,0", all_branches=True ) self.assertEqual(len(ret), 3) @@ -145,12 +145,12 @@ class TestMetricsRecursive(TestDvc): def test(self): - ret = self.dvc.metrics_show( + ret = self.dvc.metrics.show( "nested", all_branches=True, recursive=False ) self.assertEqual(len(ret), 0) - ret = self.dvc.metrics_show( + ret = self.dvc.metrics.show( "nested", all_branches=True, recursive=True ) self.assertEqual(len(ret), 1) @@ -334,7 +334,7 @@ class TestMetricsCLI(TestMetrics): with self.assertRaises(DvcException): self.dvc.run(outs_no_cache=["metrics_dir"]) - self.dvc.metrics_add("metrics_dir") + self.dvc.metrics.add("metrics_dir") def test_binary(self): with open("metrics_bin", "w+") as fd: @@ -342,7 +342,7 @@ class TestMetricsCLI(TestMetrics): with self.assertRaises(DvcException): self.dvc.run(outs_no_cache=["metrics_bin"]) - self.dvc.metrics_add("metrics_bin") + self.dvc.metrics.add("metrics_bin") def test_non_existing(self): ret = main(["metrics", "add", "non-existing"]) @@ -370,7 +370,7 @@ class TestCachedMetrics(TestDvc): json.dump({"metrics": branch}, fd) stages = self.dvc.add("metrics.json") - self.dvc.metrics_add("metrics.json", typ="json", xpath="metrics") + self.dvc.metrics.add("metrics.json", typ="json", xpath="metrics") self.assertEqual(len(stages), 1) stage = stages[0] self.assertIsNotNone(stage) @@ -413,9 +413,9 @@ class TestCachedMetrics(TestDvc): func("one") func("two") - self.dvc = Project(".") + self.dvc = DvcRepo(".") - res = self.dvc.metrics_show( + res = self.dvc.metrics.show( "metrics.json", all_branches=True, typ="json", xpath="metrics" ) @@ -428,7 +428,7 @@ class TestCachedMetrics(TestDvc): }, ) - res = self.dvc.metrics_show( + res = self.dvc.metrics.show( "", all_branches=True, typ="json", xpath="metrics" ) diff --git a/tests/test_move.py b/tests/test_move.py index 9705c063b..cff338801 100644 --- a/tests/test_move.py +++ b/tests/test_move.py @@ -51,9 +51,9 @@ class TestCmdMove(TestDvc): class TestMoveNotDataSource(TestRepro): def test(self): - from dvc.project import Project + from dvc.repo import Repo as DvcRepo - self.dvc = Project(self._root_dir) + self.dvc = DvcRepo(self._root_dir) with self.assertRaises(MoveNotDataSourceError): self.dvc.move(self.file1, "dst") diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index be2d0ccec..f59e6afed 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -234,7 +234,7 @@ class TestPipelineListSingle(TestPipelineShowDeep): self.assertEqual(ret, 0) -class TestProjectPipeline(TestDvc): +class TestDvcRepoPipeline(TestDvc): def test_no_stages(self): pipelines = self.dvc.pipelines() self.assertEqual(len(pipelines), 0) diff --git a/tests/test_remote.py b/tests/test_remote.py index ce2453055..c2e611732 100644 --- a/tests/test_remote.py +++ b/tests/test_remote.py @@ -51,7 +51,7 @@ class TestRemote(TestDvc): ret = main(["remote", "add", "mylocal", dname]) self.assertEqual(ret, 0) - # NOTE: we are in the project's root and config is in .dvc/, so + # NOTE: we are in the repo's root and config is in .dvc/, so # dir path written to config should be just one level above. rel = os.path.join("..", dname) config = configobj.ConfigObj(self.dvc.config.config_file) diff --git a/tests/test_repro.py b/tests/test_repro.py index 001588be8..95d70908c 100644 --- a/tests/test_repro.py +++ b/tests/test_repro.py @@ -15,13 +15,17 @@ from google.cloud import storage as gc from flaky.flaky_decorator import flaky from dvc.main import main -from dvc.project import Project, ReproductionError +from dvc.repo import Repo as DvcRepo from dvc.utils import file_md5 from dvc.utils.compat import urljoin from dvc.remote.local import RemoteLOCAL from dvc.stage import Stage, StageFileDoesNotExistError from dvc.system import System -from dvc.exceptions import CyclicGraphError, WorkingDirectoryAsOutputError +from dvc.exceptions import ( + CyclicGraphError, + WorkingDirectoryAsOutputError, + ReproductionError, +) from tests.basic_env import TestDvc from tests.test_data_cloud import _should_test_aws, TEST_AWS_REPO_BUCKET @@ -160,7 +164,7 @@ class TestReproWorkingDirectoryAsOutput(TestDvc): # NOTE: os.walk() walks in a sorted order and we need dir2 subdirs to # be processed before dir1 to load error.dvc first. - with patch.object(Project, "stages") as mock_stages: + with patch.object(DvcRepo, "stages") as mock_stages: mock_stages.return_value = [ nested_stage, Stage.load(self.dvc, error_stage_path), @@ -582,11 +586,11 @@ class TestReproMetricsAddUnchanged(TestDvc): stages = self.dvc.reproduce(file1_stage) self.assertEqual(len(stages), 0) - self.dvc.metrics_add(file1) + self.dvc.metrics.add(file1) stages = self.dvc.reproduce(file1_stage) self.assertEqual(len(stages), 0) - self.dvc.metrics_remove(file1) + self.dvc.metrics.remove(file1) stages = self.dvc.reproduce(file1_stage) self.assertEqual(len(stages), 0) @@ -856,7 +860,7 @@ class TestReproExternalBase(TestDvc): ret = main(["remote", "modify", remote_name, "type", "hardlink"]) self.assertEqual(ret, 0) - self.dvc = Project(".") + self.dvc = DvcRepo(".") foo_key = remote_key + self.sep + self.FOO bar_key = remote_key + self.sep + self.BAR @@ -1079,7 +1083,7 @@ class TestReproExternalLOCAL(TestReproExternalBase): self.tmpdir = TestDvc.mkdtemp() ret = main(["config", "cache.type", "hardlink"]) self.assertEqual(ret, 0) - self.dvc = Project(".") + self.dvc = DvcRepo(".") def should_test(self): return True @@ -1148,7 +1152,7 @@ class TestReproExternalHTTP(TestReproExternalBase): self.assertEqual(ret1, 0) self.assertEqual(ret2, 0) - self.dvc = Project(".") + self.dvc = DvcRepo(".") # Import with StaticFileServer(): diff --git a/tests/unit/remote/base.py b/tests/unit/remote/base.py index 1199df216..c21b8105d 100644 --- a/tests/unit/remote/base.py +++ b/tests/unit/remote/base.py @@ -18,7 +18,7 @@ class TestMissingDeps(TestCase): class TestCmdError(TestCase): def test(self): for remote_class in [RemoteSSH, RemoteHDFS]: - project = None + repo = None config = {} remote_name = remote_class.scheme @@ -32,4 +32,4 @@ class TestCmdError(TestCase): side_effect=RemoteCmdError(remote_name, cmd, ret, err), ): with self.assertRaises(RemoteCmdError): - remote_class(project, config).remove("file") + remote_class(repo, config).remove("file")