Merge pull request #1759 from pared/1214

[WIP] run: add --outs-persist and --outs-persist-no-cache options

Author: efiop

dvc/command/run.py

@@ -43,6 +43,8 @@ def run(self):
                 ignore_build_cache=self.args.ignore_build_cache,
                 remove_outs=self.args.remove_outs,
                 no_commit=self.args.no_commit,
+                outs_persist=self.args.outs_persist,
+                outs_persist_no_cache=self.args.outs_persist_no_cache,
             )
         except DvcException:
             logger.error("failed to run command")
@@ -175,6 +177,20 @@ def add_parser(subparsers, parent_parser):
         default=False,
         help="Don't put files/directories into cache.",
     )
+    run_parser.add_argument(
+        "--outs-persist",
+        action="append",
+        default=[],
+        help="Declare output file or directory that will not be "
+        "removed upon repro.",
+    )
+    run_parser.add_argument(
+        "--outs-persist-no-cache",
+        action="append",
+        default=[],
+        help="Declare output file or directory that will not be "
+        "removed upon repro (do not put into DVC cache).",
+    )
     run_parser.add_argument(
         "command", nargs=argparse.REMAINDER, help="Command to execute."
     )

dvc/output/__init__.py

@@ -48,23 +48,38 @@
     schema.Optional(RemoteHDFS.PARAM_CHECKSUM): schema.Or(str, None),
     schema.Optional(OutputBase.PARAM_CACHE): bool,
     schema.Optional(OutputBase.PARAM_METRIC): OutputBase.METRIC_SCHEMA,
+    schema.Optional(OutputBase.PARAM_PERSIST): bool,
 }
 
 
-def _get(stage, p, info, cache, metric):
+def _get(stage, p, info, cache, metric, persist):
     parsed = urlparse(p)
     if parsed.scheme == "remote":
         name = Config.SECTION_REMOTE_FMT.format(parsed.netloc)
         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
+            stage,
+            p,
+            info,
+            cache=cache,
+            remote=remote,
+            metric=metric,
+            persist=persist,
         )
 
     for o in OUTS:
         if o.supported(p):
             return o(stage, p, info, cache=cache, remote=None, metric=metric)
-    return OutputLOCAL(stage, p, info, cache=cache, remote=None, metric=metric)
+    return OutputLOCAL(
+        stage,
+        p,
+        info,
+        cache=cache,
+        remote=None,
+        metric=metric,
+        persist=persist,
+    )
 
 
 def loadd_from(stage, d_list):
@@ -73,12 +88,24 @@ def loadd_from(stage, d_list):
         p = d.pop(OutputBase.PARAM_PATH)
         cache = d.pop(OutputBase.PARAM_CACHE, True)
         metric = d.pop(OutputBase.PARAM_METRIC, False)
-        ret.append(_get(stage, p, info=d, cache=cache, metric=metric))
+        persist = d.pop(OutputBase.PARAM_PERSIST, False)
+        ret.append(
+            _get(stage, p, info=d, cache=cache, metric=metric, persist=persist)
+        )
     return ret
 
 
-def loads_from(stage, s_list, use_cache=True, metric=False):
+def loads_from(stage, s_list, use_cache=True, metric=False, persist=False):
     ret = []
     for s in s_list:
-        ret.append(_get(stage, s, info={}, cache=use_cache, metric=metric))
+        ret.append(
+            _get(
+                stage,
+                s,
+                info={},
+                cache=use_cache,
+                metric=metric,
+                persist=persist,
+            )
+        )
     return ret

dvc/output/base.py

@@ -36,6 +36,7 @@ class OutputBase(object):
     PARAM_METRIC = "metric"
     PARAM_METRIC_TYPE = "type"
     PARAM_METRIC_XPATH = "xpath"
+    PARAM_PERSIST = "persist"
 
     METRIC_SCHEMA = Or(
         None,
@@ -50,7 +51,14 @@ class OutputBase(object):
     IsNotFileOrDirError = OutputIsNotFileOrDirError
 
     def __init__(
-        self, stage, path, info=None, remote=None, cache=True, metric=False
+        self,
+        stage,
+        path,
+        info=None,
+        remote=None,
+        cache=True,
+        metric=False,
+        persist=False,
     ):
         self.stage = stage
         self.repo = stage.repo
@@ -59,6 +67,7 @@ def __init__(
         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
+        self.persist = persist
 
         if (
             self.use_cache
@@ -186,6 +195,7 @@ def dumpd(self):
                 del self.metric[self.PARAM_METRIC_XPATH]
 
         ret[self.PARAM_METRIC] = self.metric
+        ret[self.PARAM_PERSIST] = self.persist
 
         return ret
 
@@ -231,3 +241,7 @@ def get_files_number(self):
             return 1
 
         return 0
+
+    def unprotect(self):
+        if self.exists:
+            self.remote.unprotect(self.path_info)

dvc/output/hdfs.py

@@ -11,10 +11,23 @@ class OutputHDFS(OutputBase):
     REMOTE = RemoteHDFS
 
     def __init__(
-        self, stage, path, info=None, remote=None, cache=True, metric=False
+        self,
+        stage,
+        path,
+        info=None,
+        remote=None,
+        cache=True,
+        metric=False,
+        persist=False,
     ):
         super(OutputHDFS, self).__init__(
-            stage, path, info=info, remote=remote, cache=cache, metric=metric
+            stage,
+            path,
+            info=info,
+            remote=remote,
+            cache=cache,
+            metric=metric,
+            persist=persist,
         )
         if remote:
             path = posixpath.join(remote.url, urlparse(path).path.lstrip("/"))

dvc/output/local.py

@@ -14,10 +14,23 @@ class OutputLOCAL(OutputBase):
     REMOTE = RemoteLOCAL
 
     def __init__(
-        self, stage, path, info=None, remote=None, cache=True, metric=False
+        self,
+        stage,
+        path,
+        info=None,
+        remote=None,
+        cache=True,
+        metric=False,
+        persist=False,
     ):
         super(OutputLOCAL, self).__init__(
-            stage, path, info, remote=remote, cache=cache, metric=metric
+            stage,
+            path,
+            info,
+            remote=remote,
+            cache=cache,
+            metric=metric,
+            persist=persist,
         )
         if remote:
             p = os.path.join(

dvc/output/s3.py

@@ -11,10 +11,23 @@ class OutputS3(OutputBase):
     REMOTE = RemoteS3
 
     def __init__(
-        self, stage, path, info=None, remote=None, cache=True, metric=False
+        self,
+        stage,
+        path,
+        info=None,
+        remote=None,
+        cache=True,
+        metric=False,
+        persist=False,
     ):
         super(OutputS3, self).__init__(
-            stage, path, info=info, remote=remote, cache=cache, metric=metric
+            stage,
+            path,
+            info=info,
+            remote=remote,
+            cache=cache,
+            metric=metric,
+            persist=persist,
         )
         bucket = remote.bucket if remote else urlparse(path).netloc
         path = urlparse(path).path.lstrip("/")

dvc/output/ssh.py

@@ -12,10 +12,23 @@ class OutputSSH(OutputBase):
     REMOTE = RemoteSSH
 
     def __init__(
-        self, stage, path, info=None, remote=None, cache=True, metric=False
+        self,
+        stage,
+        path,
+        info=None,
+        remote=None,
+        cache=True,
+        metric=False,
+        persist=False,
     ):
         super(OutputSSH, self).__init__(
-            stage, path, info=info, remote=remote, cache=cache, metric=metric
+            stage,
+            path,
+            info=info,
+            remote=remote,
+            cache=cache,
+            metric=metric,
+            persist=persist,
         )
         parsed = urlparse(path)
         host = remote.host if remote else parsed.hostname

dvc/remote/base.py

@@ -362,3 +362,7 @@ def checkout(self, output, force=False, progress_callback=None):
         self.do_checkout(
             output, force=force, progress_callback=progress_callback
         )
+
+    @staticmethod
+    def unprotect(path_info):
+        pass

dvc/remote/local.py

@@ -743,3 +743,44 @@ def _log_missing_caches(self, checksum_info_dict):
                 "nor on remote. Missing cache files: {}".format(missing_desc)
             )
             logger.warning(msg)
+
+    @staticmethod
+    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()))
+
+            # The operations order is important here - if some application
+            # would access the file during the process of copyfile then it
+            # would get only the part of file. So, at first, the file should be
+            # copied with the temporary name, and then original file should be
+            # replaced by new.
+            copyfile(path, tmp)
+            remove(path)
+            os.rename(tmp, path)
+
+        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)
+
+    @staticmethod
+    def _unprotect_dir(path):
+        for path in walk_files(path):
+            RemoteLOCAL._unprotect_file(path)
+
+    @staticmethod
+    def unprotect(path_info):
+        path = path_info["path"]
+        if not os.path.exists(path):
+            raise DvcException(
+                "can't unprotect non-existing data '{}'".format(path)
+            )
+
+        if os.path.isdir(path):
+            RemoteLOCAL._unprotect_dir(path)
+        else:
+            RemoteLOCAL._unprotect_file(path)

dvc/repo/__init__.py

@@ -101,11 +101,9 @@ def init(root_dir=os.curdir, no_scm=False, force=False):
         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 unprotect(self, target):
+        path_info = {"schema": "local", "path": target}
+        return self.cache.local.unprotect(path_info)
 
     def _ignore(self):
         flist = [

dvc/repo/run.py

@@ -20,6 +20,8 @@ def run(
     ignore_build_cache=False,
     remove_outs=False,
     no_commit=False,
+    outs_persist=None,
+    outs_persist_no_cache=None,
 ):
     from dvc.stage import Stage
 
@@ -33,6 +35,10 @@ def run(
         metrics = []
     if metrics_no_cache is None:
         metrics_no_cache = []
+    if outs_persist is None:
+        outs_persist = []
+    if outs_persist_no_cache is None:
+        outs_persist_no_cache = []
 
     with self.state:
         stage = Stage.create(
@@ -49,6 +55,8 @@ def run(
             overwrite=overwrite,
             ignore_build_cache=ignore_build_cache,
             remove_outs=remove_outs,
+            outs_persist=outs_persist,
+            outs_persist_no_cache=outs_persist_no_cache,
         )
 
     if stage is None: