Merge pull request #1601 from efiop/919

Introduce `dvc commit`
This commit is contained in:
Ruslan Kuprieiev
2019-02-17 10:04:33 +02:00
committed by GitHub
24 changed files with 377 additions and 69 deletions
+2
View File
@@ -29,6 +29,7 @@ import dvc.command.root as root
import dvc.command.lock as lock
import dvc.command.pipeline as pipeline
import dvc.command.daemon as daemon
import dvc.command.commit as commit
from dvc.exceptions import DvcParserError
from dvc import VERSION
@@ -56,6 +57,7 @@ COMMANDS = [
lock,
pipeline,
daemon,
commit,
]
+11 -1
View File
@@ -9,7 +9,11 @@ class CmdAdd(CmdBase):
def run(self):
for target in self.args.targets:
try:
self.repo.add(target, recursive=self.args.recursive)
self.repo.add(
target,
recursive=self.args.recursive,
no_commit=self.args.no_commit,
)
except DvcException:
logger.error("failed to add file")
return 1
@@ -28,6 +32,12 @@ def add_parser(subparsers, parent_parser):
default=False,
help="Recursively add each file under the directory.",
)
add_parser.add_argument(
"--no-commit",
action="store_true",
default=False,
help="Don't put files/directories into cache.",
)
add_parser.add_argument(
"targets", nargs="+", help="Input files/directories."
)
+63
View File
@@ -0,0 +1,63 @@
from __future__ import unicode_literals
import dvc.logger as logger
from dvc.exceptions import DvcException
from dvc.command.base import CmdBase
class CmdCommit(CmdBase):
def run(self):
if not self.args.targets:
self.args.targets = [None]
for target in self.args.targets:
try:
self.repo.commit(
target,
with_deps=self.args.with_deps,
recursive=self.args.recursive,
force=self.args.force,
)
except DvcException:
logger.error(
"failed to commit{}".format(
(" " + target) if target else ""
)
)
return 1
return 0
def add_parser(subparsers, parent_parser):
COMMIT_HELP = "Record changes to the repository."
commit_parser = subparsers.add_parser(
"commit",
parents=[parent_parser],
description=COMMIT_HELP,
help=COMMIT_HELP,
)
commit_parser.add_argument(
"-f",
"--force",
action="store_true",
default=False,
help="Commit even if checksums for dependencies/outputs changed.",
)
commit_parser.add_argument(
"-d",
"--with-deps",
action="store_true",
default=False,
help="Commit all dependencies of the specified target.",
)
commit_parser.add_argument(
"-R",
"--recursive",
action="store_true",
default=False,
help="Commit cache for subdirectories of the specified directory.",
)
commit_parser.add_argument(
"targets", nargs="*", default=None, help="DVC files."
)
commit_parser.set_defaults(func=CmdCommit)
+7
View File
@@ -33,6 +33,7 @@ class CmdRepro(CmdBase):
pipeline=self.args.pipeline,
all_pipelines=self.args.all_pipelines,
ignore_build_cache=self.args.ignore_build_cache,
no_commit=self.args.no_commit,
)
if len(stages) == 0:
@@ -124,4 +125,10 @@ def add_parser(subparsers, parent_parser):
help="Reproduce all descendants of a changed stage even if their "
"direct dependencies didn't change.",
)
repro_parser.add_argument(
"--no-commit",
action="store_true",
default=False,
help="Don't put files/directories into cache.",
)
repro_parser.set_defaults(func=CmdRepro)
+7
View File
@@ -42,6 +42,7 @@ class CmdRun(CmdBase):
overwrite=overwrite,
ignore_build_cache=self.args.ignore_build_cache,
remove_outs=self.args.remove_outs,
no_commit=self.args.no_commit,
)
except DvcException:
logger.error("failed to run command")
@@ -167,6 +168,12 @@ def add_parser(subparsers, parent_parser):
default=False,
help="Remove outputs before running the command.",
)
run_parser.add_argument(
"--no-commit",
action="store_true",
default=False,
help="Don't put files/directories into cache.",
)
run_parser.add_argument(
"command", nargs=argparse.REMAINDER, help="Command to execute."
)
+17 -6
View File
@@ -2,10 +2,11 @@ from __future__ import unicode_literals
import dvc.logger as logger
from dvc.command.data_sync import CmdDataBase
from dvc.utils.compat import str
class CmdDataStatus(CmdDataBase):
STATUS_LEN = 10
STATUS_LEN = 20
STATUS_INDENT = "\t"
UP_TO_DATE_MSG = "Pipeline is up to date. Nothing to reproduce."
@@ -17,13 +18,23 @@ class CmdDataStatus(CmdDataBase):
def _show(self, status, indent=0):
ind = indent * self.STATUS_INDENT
if isinstance(status, str):
logger.info("{}{}".format(ind, status))
return
if isinstance(status, list):
for entry in status:
self._show(entry, indent)
return
assert isinstance(status, dict)
for key, value in status.items():
if isinstance(value, dict):
logger.info("{}{}".format(ind, key))
if isinstance(value, str):
logger.info("{}{}{}".format(ind, self._normalize(value), key))
elif value:
logger.info("{}{}:".format(ind, key))
self._show(value, indent + 1)
else:
msg = "{}{}{}".format(ind, self._normalize(value), key)
logger.info(msg)
def do_run(self, target=None):
indent = 1 if self.args.cloud else 0
+28 -17
View File
@@ -111,15 +111,12 @@ class OutputBase(object):
def exists(self):
return self.remote.exists(self.path_info)
def changed(self):
if not self.exists:
return True
if not self.use_cache:
return self.info != self.remote.save_info(self.path_info)
return getattr(self.repo.cache, self.scheme).changed(
self.path_info, self.info
def changed_checksum(self):
return (
self.checksum
!= self.remote.save_info(self.path_info)[
self.remote.PARAM_CHECKSUM
]
)
def changed_cache(self):
@@ -131,17 +128,30 @@ class OutputBase(object):
return cache.changed_cache(self.checksum)
def status(self):
if self.changed():
# FIXME better msgs
return {str(self): "changed"}
if self.checksum and self.use_cache and self.changed_cache():
return {str(self): "not in cache"}
if not self.exists:
return {str(self): "deleted"}
if self.changed_checksum():
return {str(self): "modified"}
if not self.checksum:
return {str(self): "new"}
return {}
def changed(self):
return bool(self.status())
def save(self):
if not self.use_cache:
self.info = self.remote.save_info(self.path_info)
else:
self.info = getattr(self.repo.cache, self.scheme).save(
self.path_info
self.info = self.remote.save_info(self.path_info)
def commit(self):
if self.use_cache:
getattr(self.repo.cache, self.scheme).save(
self.path_info, self.info
)
def dumpd(self):
@@ -191,6 +201,7 @@ class OutputBase(object):
self.url = out.url
self.path_info = out.path_info
self.save()
self.commit()
if self.scheme == "local" and self.use_cache and self.is_local:
self.repo.scm.ignore(self.path)
+1 -1
View File
@@ -117,4 +117,4 @@ class OutputLOCAL(OutputBase):
if self.use_cache:
self.repo.scm.ignore(self.path)
self.info = self.repo.cache.local.save(self.path_info)
self.info = self.remote.save_info(self.path_info)
+1 -1
View File
@@ -172,7 +172,7 @@ class RemoteBase(object):
logger.debug("'{}' hasn't changed.".format(path_info))
return False
def save(self, path_info):
def save(self, path_info, checksum_info):
raise RemoteActionNotImplemented("save", self.scheme)
def download(
+2 -4
View File
@@ -88,20 +88,18 @@ class RemoteHDFS(RemoteBase):
return {self.PARAM_CHECKSUM: self.checksum(path_info)}
def save(self, path_info):
def save(self, path_info, checksum_info):
if path_info["scheme"] != "hdfs":
raise NotImplementedError
assert path_info.get("path")
checksum = self.checksum(path_info)
checksum = checksum_info[self.PARAM_CHECKSUM]
dest = path_info.copy()
dest["path"] = self.checksum_to_path(checksum)
self.copy(path_info, dest)
return {self.PARAM_CHECKSUM: checksum}
def remove(self, path_info):
if path_info["scheme"] != "hdfs":
raise NotImplementedError
+7 -11
View File
@@ -370,9 +370,7 @@ class RemoteLOCAL(RemoteBase):
move(inp, tmp)
move(tmp, outp)
def _save_file(self, path_info):
path = path_info["path"]
md5 = self.state.update(path)
def _save_file(self, path, md5):
assert md5 is not None
cache = self.get(md5)
@@ -393,9 +391,8 @@ class RemoteLOCAL(RemoteBase):
return {self.PARAM_CHECKSUM: md5}
def _save_dir(self, path_info):
path = path_info["path"]
md5, dir_info = self.state.update_info(path)
def _save_dir(self, path, md5):
dir_info = self.load_dir_cache(md5)
dir_relpath = os.path.relpath(path)
dir_size = len(dir_info)
bar = dir_size > LARGE_DIR_SIZE
@@ -429,9 +426,7 @@ class RemoteLOCAL(RemoteBase):
if bar:
progress.finish_target(dir_relpath)
return {self.PARAM_CHECKSUM: md5}
def save(self, path_info):
def save(self, path_info, checksum_info):
if path_info["scheme"] != "local":
raise NotImplementedError
@@ -442,10 +437,11 @@ class RemoteLOCAL(RemoteBase):
msg.format(os.path.relpath(path), os.path.relpath(self.cache_dir))
)
md5 = checksum_info[self.PARAM_CHECKSUM]
if os.path.isdir(path):
return self._save_dir(path_info)
self._save_dir(path, md5)
else:
return self._save_file(path_info)
self._save_file(path, md5)
def save_info(self, path_info):
if path_info["scheme"] != "local":
+2 -4
View File
@@ -109,18 +109,16 @@ class RemoteS3(RemoteBase):
source = {"Bucket": from_info["bucket"], "Key": from_info["path"]}
self.s3.copy(source, to_info["bucket"], to_info["path"])
def save(self, path_info):
def save(self, path_info, checksum_info):
if path_info["scheme"] != "s3":
raise NotImplementedError
etag = self.get_etag(path_info["bucket"], path_info["path"])
etag = checksum_info[self.PARAM_CHECKSUM]
path = self.checksum_to_path(etag)
to_info = {"scheme": "s3", "bucket": self.bucket, "path": path}
self.copy(path_info, to_info)
return {self.PARAM_CHECKSUM: etag}
def remove(self, path_info):
if path_info["scheme"] != "s3":
raise NotImplementedError
+2 -4
View File
@@ -244,18 +244,16 @@ class RemoteSSH(RemoteBase):
return {self.PARAM_CHECKSUM: self.md5(path_info)}
def save(self, path_info):
def save(self, path_info, checksum_info):
if path_info["scheme"] != "ssh":
raise NotImplementedError
md5 = self.md5(path_info)
md5 = checksum_info[self.PARAM_CHECKSUM]
dest = path_info.copy()
dest["path"] = self.checksum_to_path(md5)
self.copy(path_info, dest)
return {self.PARAM_CHECKSUM: md5}
def remove(self, path_info):
if path_info["scheme"] != "ssh":
raise NotImplementedError
+1
View File
@@ -28,6 +28,7 @@ class Repo(object):
from dvc.repo.pull import pull
from dvc.repo.status import status
from dvc.repo.gc import gc
from dvc.repo.commit import commit
def __init__(self, root_dir=None):
from dvc.config import Config
+3 -1
View File
@@ -1,7 +1,7 @@
import os
def add(self, fname, recursive=False):
def add(self, fname, recursive=False, no_commit=False):
from dvc.stage import Stage
fnames = []
@@ -30,6 +30,8 @@ def add(self, fname, recursive=False):
continue
stage.save()
if not no_commit:
stage.commit()
stages.append(stage)
self.check_dag(self.stages() + stages)
+11
View File
@@ -0,0 +1,11 @@
def commit(self, target, with_deps=False, recursive=False, force=False):
if target and not recursive:
stages = self.collect(target, with_deps=with_deps)
else:
stages = self.active_stages(target)
with self.state:
for stage in stages:
stage.check_can_commit(force=force)
stage.commit()
stage.dump()
+22 -6
View File
@@ -6,7 +6,7 @@ import dvc.logger as logger
from dvc.exceptions import ReproductionError
def _reproduce_stage(stages, node, force, dry, interactive):
def _reproduce_stage(stages, node, force, dry, interactive, no_commit):
stage = stages[node]
if stage.locked:
@@ -15,7 +15,9 @@ def _reproduce_stage(stages, node, force, dry, interactive):
" not going to be reproduced.".format(path=stage.relpath)
)
stage = stage.reproduce(force=force, dry=dry, interactive=interactive)
stage = stage.reproduce(
force=force, dry=dry, interactive=interactive, no_commit=no_commit
)
if not stage:
return []
@@ -35,6 +37,7 @@ def reproduce(
pipeline=False,
all_pipelines=False,
ignore_build_cache=False,
no_commit=False,
):
from dvc.stage import Stage
@@ -75,6 +78,7 @@ def reproduce(
dry=dry,
interactive=interactive,
ignore_build_cache=ignore_build_cache,
no_commit=no_commit,
)
ret.extend(stages)
@@ -91,6 +95,7 @@ def _reproduce(
dry=False,
interactive=False,
ignore_build_cache=False,
no_commit=False,
):
import networkx as nx
from dvc.stage import Stage
@@ -102,23 +107,34 @@ def _reproduce(
if recursive:
ret = _reproduce_stages(
G, stages, node, force, dry, interactive, ignore_build_cache
G,
stages,
node,
force,
dry,
interactive,
ignore_build_cache,
no_commit,
)
else:
ret = _reproduce_stage(stages, node, force, dry, interactive)
ret = _reproduce_stage(
stages, node, force, dry, interactive, no_commit
)
return ret
def _reproduce_stages(
G, stages, node, force, dry, interactive, ignore_build_cache
G, stages, node, force, dry, interactive, ignore_build_cache, no_commit
):
import networkx as nx
result = []
for n in nx.dfs_postorder_nodes(G, node):
try:
ret = _reproduce_stage(stages, n, force, dry, interactive)
ret = _reproduce_stage(
stages, n, force, dry, interactive, no_commit
)
if len(ret) != 0 and ignore_build_cache:
# NOTE: we are walking our pipeline from the top to the
+2 -1
View File
@@ -15,6 +15,7 @@ def run(
overwrite=False,
ignore_build_cache=False,
remove_outs=False,
no_commit=False,
):
from dvc.stage import Stage
@@ -53,7 +54,7 @@ def run(
self.files_to_git_add = []
with self.state:
if not no_exec:
stage.run()
stage.run(no_commit=no_commit)
stage.dump()
+60 -12
View File
@@ -67,6 +67,10 @@ class StageBadCwdError(DvcException):
super(StageBadCwdError, self).__init__(msg.format(cwd))
class StageCommitError(DvcException):
pass
class MissingDep(DvcException):
def __init__(self, deps):
assert len(deps) > 0
@@ -251,7 +255,9 @@ class Stage(object):
self.remove_outs(ignore_remove=True)
os.unlink(self.path)
def reproduce(self, force=False, dry=False, interactive=False):
def reproduce(
self, force=False, dry=False, interactive=False, no_commit=False
):
if not self.changed() and not force:
return None
@@ -268,7 +274,7 @@ class Stage(object):
logger.info("Reproducing '{stage}'".format(stage=self.relpath))
self.run(dry=dry)
self.run(dry=dry, no_commit=no_commit)
logger.debug("'{stage}' was reproduced".format(stage=self.relpath))
@@ -526,6 +532,39 @@ class Stage(object):
self.md5 = self._compute_md5()
@staticmethod
def _changed_entries(entries):
ret = []
for entry in entries:
if entry.checksum and entry.changed_checksum():
ret.append(entry.rel_path)
return ret
def check_can_commit(self, force):
changed_deps = self._changed_entries(self.deps)
changed_outs = self._changed_entries(self.outs)
if changed_deps or changed_outs or self.changed_md5():
msg = (
"dependencies {}".format(changed_deps) if changed_deps else ""
)
msg += " and " if (changed_deps and changed_outs) else ""
msg += "outputs {}".format(changed_outs) if changed_outs else ""
msg += "md5" if not (changed_deps or changed_outs) else ""
msg += " of '{}' changed. Are you sure you commit it?".format(
self.relpath
)
if not force and not prompt.confirm(msg):
raise StageCommitError(
"unable to commit changed '{}'. Use `-f|--force` to "
"force.`".format(self.relpath)
)
self.save()
def commit(self):
for out in self.outs:
out.commit()
def _check_missing_deps(self):
missing = [dep for dep in self.deps if not dep.exists]
@@ -585,7 +624,7 @@ class Stage(object):
if p.returncode != 0:
raise StageCmdFailedError(self)
def run(self, dry=False, resume=False):
def run(self, dry=False, resume=False, no_commit=False):
if self.locked:
logger.info(
"Verifying outputs in locked stage '{stage}'".format(
@@ -625,6 +664,8 @@ class Stage(object):
if not dry:
self.save()
if not no_commit:
self.commit()
def check_missing_outputs(self):
paths = [
@@ -641,26 +682,33 @@ class Stage(object):
out.checkout(force=force)
@staticmethod
def _status(entries, name):
def _status(entries):
ret = {}
for entry in entries:
ret.update(entry.status())
if ret:
return {name: ret}
return {}
return ret
def status(self):
ret = {}
ret = []
if not self.locked:
ret.update(self._status(self.deps, "deps"))
deps_status = self._status(self.deps)
if deps_status:
ret.append({"changed deps": deps_status})
ret.update(self._status(self.outs, "outs"))
outs_status = self._status(self.outs)
if outs_status:
ret.append({"changed outs": outs_status})
if ret or self.changed_md5() or self.is_callback:
if self.changed_md5():
ret.append("changed checksum")
if self.is_callback:
ret.append("always changed")
if ret:
return {self.relpath: ret}
return {}
+13
View File
@@ -264,3 +264,16 @@ class TestShouldUpdateStateEntryForDirectoryAfterAdd(TestDvc):
)
self.assertEqual(ret, 0)
self.assertEqual(file_md5_counter.mock.call_count, 3)
class TestAddCommit(TestDvc):
def test(self):
ret = main(["add", self.FOO, "--no-commit"])
self.assertEqual(ret, 0)
self.assertTrue(os.path.isfile(self.FOO))
self.assertEqual(len(os.listdir(self.dvc.cache.local.cache_dir)), 0)
ret = main(["commit", self.FOO + ".dvc"])
self.assertEqual(ret, 0)
self.assertTrue(os.path.isfile(self.FOO))
self.assertEqual(len(os.listdir(self.dvc.cache.local.cache_dir)), 1)
+91
View File
@@ -0,0 +1,91 @@
import yaml
import shutil
from tests.basic_env import TestDvc
from dvc.stage import StageCommitError
class TestCommitRecursive(TestDvc):
def test(self):
stages = self.dvc.add(self.DATA_DIR, recursive=True, no_commit=True)
self.assertEqual(len(stages), 2)
self.assertNotEqual(self.dvc.status(), {})
self.dvc.commit(self.DATA_DIR, recursive=True)
self.assertEqual(self.dvc.status(), {})
class TestCommitForce(TestDvc):
def test(self):
stages = self.dvc.add(self.FOO, no_commit=True)
self.assertEqual(len(stages), 1)
stage = stages[0]
with self.dvc.state:
self.assertTrue(stage.outs[0].changed_cache())
with open(self.FOO, "a") as fobj:
fobj.write(self.FOO_CONTENTS)
with self.dvc.state:
self.assertTrue(stage.outs[0].changed_cache())
with self.assertRaises(StageCommitError):
self.dvc.commit(stage.path)
with self.dvc.state:
self.assertTrue(stage.outs[0].changed_cache())
self.dvc.commit(stage.path, force=True)
self.assertEqual(self.dvc.status(stage.path), {})
class TestCommitWithDeps(TestDvc):
def test(self):
stages = self.dvc.add(self.FOO, no_commit=True)
self.assertEqual(len(stages), 1)
foo_stage = stages[0]
self.assertTrue(foo_stage is not None)
self.assertEqual(len(foo_stage.outs), 1)
fname = "file"
stage = self.dvc.run(
cmd="python {} {} {}".format(self.CODE, self.FOO, fname),
outs=[fname],
deps=[self.FOO, self.CODE],
no_commit=True,
)
self.assertTrue(stage is not None)
self.assertEqual(len(stage.outs), 1)
with self.dvc.state:
self.assertTrue(foo_stage.outs[0].changed_cache())
self.assertTrue(stage.outs[0].changed_cache())
self.dvc.commit(stage.path, with_deps=True)
with self.dvc.state:
self.assertFalse(foo_stage.outs[0].changed_cache())
self.assertFalse(stage.outs[0].changed_cache())
class TestCommitChangedMd5(TestDvc):
def test(self):
stages = self.dvc.add(self.FOO, no_commit=True)
self.assertEqual(len(stages), 1)
stage = stages[0]
st = None
with open(stage.path, "r") as fobj:
st = yaml.safe_load(fobj)
st["md5"] = "1111111111"
with open(stage.path, "w") as fobj:
yaml.safe_dump(st, fobj, default_flow_style=False)
with self.assertRaises(StageCommitError):
self.dvc.commit(stage.path)
self.dvc.commit(stage.path, force=True)
+8
View File
@@ -1269,3 +1269,11 @@ class TestReproAllPipelines(TestDvc):
ret = main(["repro", "--all-pipelines"])
self.assertEqual(ret, 0)
self.assertEqual(mock_reproduce.call_count, 4)
class TestReproNoCommit(TestRepro):
def test(self):
shutil.rmtree(self.dvc.cache.local.cache_dir)
ret = main(["repro", self.file1_stage, "--no-commit"])
self.assertEqual(ret, 0)
self.assertEqual(len(os.listdir(self.dvc.cache.local.cache_dir)), 0)
+16
View File
@@ -628,3 +628,19 @@ class TestRunDeterministicChangedCmd(TestRunDeterministicBase):
self.cmd += " arg"
with self.assertRaises(StageFileAlreadyExistsError):
self._run()
class TestRunCommit(TestDvc):
def test(self):
fname = "test"
ret = main(
["run", "-o", self.FOO, "--no-commit", "echo", "test", ">", fname]
)
self.assertEqual(ret, 0)
self.assertTrue(os.path.isfile(fname))
self.assertEqual(len(os.listdir(self.dvc.cache.local.cache_dir)), 0)
ret = main(["commit", self.FOO + ".dvc"])
self.assertEqual(ret, 0)
self.assertTrue(os.path.isfile(fname))
self.assertEqual(len(os.listdir(self.dvc.cache.local.cache_dir)), 1)
View File