Merge pull request #2519 from efiop/755

dvc: move repo lock from CLI to API

dvc/command/base.py

@@ -55,16 +55,6 @@ class CmdBase(object):
         logger.warning(msg)
         return [Stage.STAGE_FILE]
 
-    def run_cmd(self):
-        from dvc.lock import LockError
-
-        try:
-            with self.repo.lock:
-                return self.run()
-        except LockError:
-            logger.exception("failed to lock before running a command")
-            return 1
-
     # Abstract methods that have to be implemented by any inheritance class
     def run(self):
         pass
@@ -73,6 +63,3 @@ class CmdBase(object):
 class CmdBaseNoRepo(CmdBase):
     def __init__(self, args):
         self.args = args
-
-    def run_cmd(self):
-        return self.run()

dvc/command/destroy.py

@@ -12,7 +12,7 @@ logger = logging.getLogger(__name__)
 
 
 class CmdDestroy(CmdBase):
-    def run_cmd(self):
+    def run(self):
         try:
             statement = (
                 "This will destroy all information about your pipelines,"

dvc/command/install.py

@@ -10,7 +10,7 @@ logger = logging.getLogger(__name__)
 
 
 class CmdInstall(CmdBase):
-    def run_cmd(self):
+    def run(self):
         try:
             self.repo.install()
         except Exception:

dvc/main.py

@@ -5,6 +5,7 @@ from __future__ import unicode_literals
 import logging
 
 from dvc.cli import parse_args
+from dvc.lock import LockError
 from dvc.config import ConfigError
 from dvc.analytics import Analytics
 from dvc.exceptions import NotDvcRepoError, DvcParserError
@@ -37,7 +38,10 @@ def main(argv=None):
             logger.setLevel(logging.DEBUG)
 
         cmd = args.func(args)
-        ret = cmd.run_cmd()
+        ret = cmd.run()
+    except LockError:
+        logger.exception("failed to lock before running a command")
+        ret = 250
     except ConfigError:
         logger.exception("configuration error")
         ret = 251

dvc/repo/__init__.py

@@ -5,6 +5,7 @@ import logging
 from contextlib import contextmanager
 from itertools import chain
 
+from functools import wraps
 from funcy import cached_property
 
 from dvc.config import Config
@@ -23,6 +24,15 @@ from dvc.utils import relpath
 logger = logging.getLogger(__name__)
 
 
+def locked(f):
+    @wraps(f)
+    def wrapper(repo, *args, **kwargs):
+        with repo.lock:
+            return f(repo, *args, **kwargs)
+
+    return wrapper
+
+
 class Repo(object):
     DVC_DIR = ".dvc"
 
@@ -36,9 +46,9 @@ class Repo(object):
     from dvc.repo.imp import imp
     from dvc.repo.imp_url import imp_url
     from dvc.repo.reproduce import reproduce
-    from dvc.repo.checkout import checkout
+    from dvc.repo.checkout import _checkout
     from dvc.repo.push import push
-    from dvc.repo.fetch import fetch
+    from dvc.repo.fetch import _fetch
     from dvc.repo.pull import pull
     from dvc.repo.status import status
     from dvc.repo.gc import gc
@@ -486,3 +496,11 @@ class Repo(object):
         git.close()
 
         return Repo(to_path)
+
+    @locked
+    def checkout(self, *args, **kwargs):
+        return self._checkout(*args, **kwargs)
+
+    @locked
+    def fetch(self, *args, **kwargs):
+        return self._fetch(*args, **kwargs)

dvc/repo/add.py

@@ -7,10 +7,13 @@ from dvc.stage import Stage
 from dvc.utils import walk_files, LARGE_DIR_SIZE
 from dvc.exceptions import RecursiveAddingWhileUsingFilename
 
+from . import locked
+
 
 logger = logging.getLogger(__name__)
 
 
+@locked
 @scm_context
 def add(repo, target, recursive=False, no_commit=False, fname=None):
     if recursive and fname:

dvc/repo/checkout.py

@@ -23,7 +23,9 @@ def get_all_files_numbers(stages):
     return sum(stage.get_all_files_number() for stage in stages)
 
 
-def checkout(self, target=None, with_deps=False, force=False, recursive=False):
+def _checkout(
+    self, target=None, with_deps=False, force=False, recursive=False
+):
     from dvc.stage import StageFileDoesNotExistError, StageFileBadNameError
 
     all_stages = self.stages()

dvc/repo/commit.py

@@ -1,3 +1,7 @@
+from . import locked
+
+
+@locked
 def commit(self, target, with_deps=False, recursive=False, force=False):
     stages = self.collect(target, with_deps=with_deps, recursive=recursive)
     with self.state:

dvc/repo/diff.py

@@ -14,6 +14,8 @@ from dvc.scm.git import (
     DIFF_EQUAL,
 )
 
+from . import locked
+
 
 DIFF_TARGET = "target"
 DIFF_IS_DIR = "is_dir"
@@ -220,6 +222,7 @@ def _diff_royal(self, target, diff_dct):
     return _diff_file(self, target, diff_dct)
 
 
+@locked
 def diff(self, a_ref, target=None, b_ref=None):
     """Gerenates diff message string output
 

dvc/repo/fetch.py

@@ -1,7 +1,7 @@
 from __future__ import unicode_literals
 
 
-def fetch(
+def _fetch(
     self,
     targets=None,
     jobs=None,

dvc/repo/gc.py

@@ -3,6 +3,8 @@ from __future__ import unicode_literals
 import collections
 import logging
 
+from . import locked
+
 
 logger = logging.getLogger(__name__)
 
@@ -53,6 +55,7 @@ def _do_gc(typ, func, clist):
         logger.info("No unused {} cache to remove.".format(typ))
 
 
+@locked
 def gc(
     self,
     all_branches=False,

dvc/repo/imp_url.py

@@ -1,7 +1,10 @@
 from dvc.utils.compat import pathlib
 from dvc.repo.scm_context import scm_context
 
+from . import locked as locked_repo
 
+
+@locked_repo
 @scm_context
 def imp_url(self, url, out=None, fname=None, erepo=None, locked=True):
     from dvc.stage import Stage

dvc/repo/lock.py

@@ -1,3 +1,7 @@
+from . import locked
+
+
+@locked
 def lock(self, target, unlock=False):
     from dvc.stage import Stage
 

dvc/repo/move.py

@@ -3,6 +3,8 @@ import os
 from dvc.exceptions import MoveNotDataSourceError
 from dvc.repo.scm_context import scm_context
 
+from . import locked
+
 
 def _expand_target_path(from_path, to_path):
     if os.path.isdir(to_path):
@@ -10,6 +12,7 @@ def _expand_target_path(from_path, to_path):
     return to_path
 
 
+@locked
 @scm_context
 def move(self, from_path, to_path):
     """

dvc/repo/pull.py

@@ -1,6 +1,9 @@
 from __future__ import unicode_literals
 
+from . import locked
 
+
+@locked
 def pull(
     self,
     targets=None,
@@ -12,7 +15,7 @@ def pull(
     force=False,
     recursive=False,
 ):
-    processed_files_count = self.fetch(
+    processed_files_count = self._fetch(
         targets,
         jobs,
         remote=remote,
@@ -22,7 +25,7 @@ def pull(
         recursive=recursive,
     )
     for target in targets or [None]:
-        self.checkout(
+        self._checkout(
             target=target,
             with_deps=with_deps,
             force=force,

dvc/repo/push.py

@@ -1,6 +1,9 @@
 from __future__ import unicode_literals
 
+from . import locked
 
+
+@locked
 def push(
     self,
     targets=None,

dvc/repo/remove.py

@@ -1,3 +1,7 @@
+from . import locked
+
+
+@locked
 def remove(self, target, outs_only=False):
     from dvc.stage import Stage
 

dvc/repo/reproduce.py

@@ -7,6 +7,9 @@ from dvc.exceptions import ReproductionError
 from dvc.repo.scm_context import scm_context
 from dvc.utils import relpath
 
+from . import locked
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -29,6 +32,7 @@ def _reproduce_stage(stages, node, **kwargs):
     return [stage]
 
 
+@locked
 @scm_context
 def reproduce(
     self,

dvc/repo/run.py

@@ -1,8 +1,10 @@
 from __future__ import unicode_literals
 
-from dvc.repo.scm_context import scm_context
+from . import locked
+from .scm_context import scm_context
 
 
+@locked
 @scm_context
 def run(self, no_exec=False, **kwargs):
     from dvc.stage import Stage

dvc/repo/status.py

@@ -3,6 +3,8 @@ from __future__ import unicode_literals
 import logging
 from funcy.py3 import cat
 
+from . import locked
+
 
 logger = logging.getLogger(__name__)
 
@@ -91,6 +93,7 @@ def _cloud_status(
     return ret
 
 
+@locked
 def status(
     self,
     targets=None,

dvc/repo/update.py

@@ -1,3 +1,7 @@
+from . import locked
+
+
+@locked
 def update(self, target):
     from dvc.stage import Stage