Stone_Red/dvc
1
0
Fork 0
mirror of https://github.com/Stone-Red-Code/dvc.git synced 2026-11-02 07:18:25 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add versioning scheme in the dvc.lock (#5128)

Browse Source 
* Add versioning scheme in the dvc.lock

* remove whitespace

* use actual size in tests assumptions
This commit is contained in:
Saugat Pachhai
2021-01-06 15:44:19 +05:45
committed by GitHub
parent 4f89db8e0d
commit 0cc5227f47
10 changed files with 242 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dvc/dvcfile.py
+64 -14
View File
@@ -6,6 +6,7 @@ import os
from voluptuous import MultipleInvalid
from dvc.exceptions import DvcException
from dvc.parsing.versions import LOCKFILE_VERSION, META_KWD
from dvc.stage import serialize
from dvc.stage.exceptions import (
StageFileBadNameError,
@@ -113,14 +114,13 @@ class FileMixin:
with self.repo.tree.open(self.path) as fd:
stage_text = fd.read()
d = parse_yaml(stage_text, self.path)
self.validate(d, self.relpath)
return d, stage_text
return self.validate(d, self.relpath), stage_text
@classmethod
def validate(cls, d, fname=None):
assert isinstance(cls.SCHEMA, collections.abc.Callable)
try:
cls.SCHEMA(d) # pylint: disable=not-callable
return cls.SCHEMA(d) # pylint: disable=not-callable
except MultipleInvalid as exc:
raise StageFileFormatError(f"'{fname}' format error: {exc}")
@@ -273,8 +273,40 @@ class PipelineFile(FileMixin):
raise NotImplementedError
def get_lockfile_schema(d):
from dvc.schema import (
COMPILED_LOCKFILE_V1_SCHEMA,
COMPILED_LOCKFILE_V2_SCHEMA,
)
schema = {
LOCKFILE_VERSION.V1: COMPILED_LOCKFILE_V1_SCHEMA,
LOCKFILE_VERSION.V2: COMPILED_LOCKFILE_V2_SCHEMA,
}
version = LOCKFILE_VERSION.from_dict(d)
return schema[version]
def migrate_lock_v1_to_v2(d, meta):
stages = {k: v for k, v in d.items()}
for key in stages:
d.pop(key)
# forcing order, meta should always be at the top
d.update(meta)
d["stages"] = stages
class Lockfile(FileMixin):
from dvc.schema import COMPILED_LOCKFILE_SCHEMA as SCHEMA
@classmethod
def validate(cls, d, fname=None):
schema = get_lockfile_schema(d)
try:
return schema(d)
except MultipleInvalid as exc:
raise StageFileFormatError(f"'{fname}' format error: {exc}")
def load(self):
if not self.exists():
@@ -282,26 +314,42 @@ class Lockfile(FileMixin):
data = load_yaml(self.path, tree=self.repo.tree)
try:
self.validate(data, fname=self.relpath)
except StageFileFormatError:
data = self.validate(data, fname=self.relpath)
except StageFileFormatError as exc:
raise LockfileCorruptedError(
f"Lockfile '{self.relpath}' is corrupted."
)
) from exc
return data
@property
def meta(self):
version = LOCKFILE_VERSION.V2.value # pylint:disable=no-member
return {META_KWD: {"version": version}}
def dump(self, stage, **kwargs):
stage_data = serialize.to_lockfile(stage)
with modify_yaml(self.path, tree=self.repo.tree) as data:
if not data:
logger.info("Generating lock file '%s'", self.relpath)
version = LOCKFILE_VERSION.from_dict(data)
if version == LOCKFILE_VERSION.V1:
logger.info(
"Migrating lock file '%s' from v1 to v2", self.relpath
)
migrate_lock_v1_to_v2(data, self.meta)
else:
if not data:
data.update(self.meta)
# order is important, meta should always be at the top
logger.info("Generating lock file '%s'", self.relpath)
modified = data.get(stage.name, {}) != stage_data.get(
data["stages"] = data.get("stages", {})
modified = data["stages"].get(stage.name, {}) != stage_data.get(
stage.name, {}
)
if modified:
logger.info("Updating lock file '%s'", self.relpath)
data.update(stage_data)
data["stages"].update(stage_data)
if modified:
self.repo.scm.track_file(self.relpath)
@@ -314,13 +362,15 @@ class Lockfile(FileMixin):
d = parse_yaml_for_update(f.read(), self.path)
self.validate(d, self.path)
if stage.name not in d:
version = LOCKFILE_VERSION.from_dict(d)
data = d if version == LOCKFILE_VERSION.V1 else d.get("stages", {})
if stage.name not in data:
return
logger.debug("Removing '%s' from '%s'", stage.name, self.path)
del d[stage.name]
del data[stage.name]
if d:
if data:
dump_yaml(self.path, d)
else:
self.remove()
dvc/parsing/versions.py
+38
View File
@@ -0,0 +1,38 @@
import enum
from collections.abc import Mapping
from voluptuous import validators
VERSION_KWD = "version"
META_KWD = "meta"
def lockfile_version_schema(value):
expected = [LOCKFILE_VERSION.V2.value] # pylint: disable=no-member
msg = "invalid version {}, expected one of {}".format(value, expected)
return validators.Any(*expected, msg=msg)(value)
class VersionEnum(str, enum.Enum):
@classmethod
def all_versions(cls):
return [v.value for v in cls]
class LOCKFILE_VERSION(VersionEnum):
V1 = "1.0"
V2 = "2.0"
@classmethod
def from_dict(cls, data):
# 1) if it's empty or or is not a dict, use the latest one (V2).
# 2) use the `meta.version`
# 3) if it's not in any of the supported version, use the latest schema
# 4) if there's no version identifier, it's a V1
if not data or not isinstance(data, Mapping):
return cls(cls.V2)
version = data.get(META_KWD, {}).get(VERSION_KWD)
if version:
return cls(version if version in cls.all_versions() else cls.V2)
return cls(cls.V1)
dvc/schema.py
+12 -3
View File
@@ -4,6 +4,7 @@ from dvc import dependency, output
from dvc.hash_info import HashInfo
from dvc.output import CHECKSUMS_SCHEMA, BaseOutput
from dvc.parsing import DO_KWD, FOREACH_KWD, VARS_KWD
from dvc.parsing.versions import META_KWD, VERSION_KWD, lockfile_version_schema
from dvc.stage.params import StageParams
STAGES = "stages"
@@ -13,7 +14,7 @@ SINGLE_STAGE_SCHEMA = {
StageParams.PARAM_WDIR: Any(str, None),
StageParams.PARAM_DEPS: Any([dependency.SCHEMA], None),
StageParams.PARAM_OUTS: Any([output.SCHEMA], None),
StageParams.PARAM_LOCKED: bool, # backard compatibility
StageParams.PARAM_LOCKED: bool, # backward compatibility
StageParams.PARAM_FROZEN: bool,
StageParams.PARAM_META: object,
StageParams.PARAM_ALWAYS_CHANGED: bool,
@@ -33,7 +34,14 @@ LOCK_FILE_STAGE_SCHEMA = {
StageParams.PARAM_PARAMS: {str: {str: object}},
StageParams.PARAM_OUTS: [DATA_SCHEMA],
}
LOCKFILE_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA}
LOCKFILE_STAGES_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA}
LOCKFILE_V1_SCHEMA = LOCKFILE_STAGES_SCHEMA
LOCKFILE_V2_SCHEMA = {
STAGES: LOCKFILE_STAGES_SCHEMA,
META_KWD: {Required(VERSION_KWD): lockfile_version_schema},
}
OUT_PSTAGE_DETAILED_SCHEMA = {
str: {
@@ -100,4 +108,5 @@ MULTI_STAGE_SCHEMA = {
COMPILED_SINGLE_STAGE_SCHEMA = Schema(SINGLE_STAGE_SCHEMA)
COMPILED_MULTI_STAGE_SCHEMA = Schema(MULTI_STAGE_SCHEMA)
COMPILED_LOCK_FILE_STAGE_SCHEMA = Schema(LOCK_FILE_STAGE_SCHEMA)
COMPILED_LOCKFILE_SCHEMA = Schema(LOCKFILE_SCHEMA)
COMPILED_LOCKFILE_V1_SCHEMA = Schema(LOCKFILE_V1_SCHEMA)
COMPILED_LOCKFILE_V2_SCHEMA = Schema(LOCKFILE_V2_SCHEMA)
dvc/stage/loader.py
+8 -1
View File
@@ -8,6 +8,7 @@ from funcy import cached_property, get_in, lcat, once, project
from dvc import dependency, output
from dvc.hash_info import HashInfo
from dvc.parsing import FOREACH_KWD, JOIN, DataResolver, EntryNotFound
from dvc.parsing.versions import LOCKFILE_VERSION
from dvc.path_info import PathInfo
from . import PipelineStage, Stage, loads_from
@@ -24,7 +25,13 @@ class StageLoader(Mapping):
self.data = data or {}
self.stages_data = self.data.get("stages", {})
self.repo = self.dvcfile.repo
self._lockfile_data = lockfile_data or {}
lockfile_data = lockfile_data or {}
version = LOCKFILE_VERSION.from_dict(lockfile_data)
if version == LOCKFILE_VERSION.V1:
self._lockfile_data = lockfile_data
else:
self._lockfile_data = lockfile_data.get("stages", {})
@cached_property
def resolver(self):
tests/func/test_dvcfile.py
+7 -3
View File
@@ -201,10 +201,12 @@ def test_remove_stage_lockfile(tmp_dir, dvc, run_copy):
lock_file = dvc_file._lockfile
assert dvc_file.exists()
assert lock_file.exists()
assert {"copy-bar-foobar", "copy-foo-bar"} == set(lock_file.load().keys())
assert {"copy-bar-foobar", "copy-foo-bar"} == set(
lock_file.load()["stages"].keys()
)
lock_file.remove_stage(stage)
assert ["copy-bar-foobar"] == list(lock_file.load().keys())
assert ["copy-bar-foobar"] == list(lock_file.load()["stages"].keys())
# sanity check
stage2.reload()
@@ -374,7 +376,9 @@ def test_dvcfile_dump_preserves_comments(tmp_dir, dvc):
],
)
def test_dvcfile_try_dumping_parametrized_stage(tmp_dir, dvc, data, name):
dump_yaml("dvc.yaml", {"stages": data, "vars": [{"foo": "foobar"}]})
dump_yaml(
"dvc.yaml", {"stages": data, "vars": [{"foo": "foobar"}]},
)
stage = dvc.stage.load_one(name=name)
dvcfile = stage.dvcfile
tests/func/test_lockfile.py
+79 -8
View File
@@ -1,10 +1,20 @@
import logging
import os
from collections import OrderedDict
from operator import itemgetter
from dvc.dvcfile import PIPELINE_LOCK
import pytest
from dvc.dvcfile import PIPELINE_LOCK, Lockfile, LockfileCorruptedError
from dvc.hash_info import HashInfo
from dvc.stage.utils import split_params_deps
from dvc.utils.fs import remove
from dvc.utils.serialize import dumps_yaml, parse_yaml_for_update
from dvc.utils.serialize import (
dump_yaml,
dumps_yaml,
load_yaml,
parse_yaml_for_update,
)
from tests.func.test_run_multistage import supported_params
FS_STRUCTURE = {
@@ -37,7 +47,7 @@ def test_deps_outs_are_sorted_by_path(tmp_dir, dvc, run_head):
run_head(*deps, name="copy-first-line")
initial_content = read_lock_file()
lock = initial_content["copy-first-line"]
lock = initial_content["stages"]["copy-first-line"]
# lock stage key order:
assert list(lock.keys()) == ["cmd", "deps", "outs"]
@@ -54,9 +64,9 @@ def test_deps_outs_are_sorted_by_path(tmp_dir, dvc, run_head):
assert list(map(itemgetter("path"), lock["deps"])) == sorted(deps)
# outs are too
assert list(
map(itemgetter("path"), initial_content["copy-first-line"]["outs"])
) == [d + "-1" for d in sorted(deps)]
assert list(map(itemgetter("path"), lock["outs"])) == [
d + "-1" for d in sorted(deps)
]
def test_order_is_preserved_when_pipeline_order_changes(
@@ -98,7 +108,7 @@ def test_cmd_changes_other_orders_are_preserved(tmp_dir, dvc, run_head):
stage.cmd = " ".join(stage.cmd.split())
stage.dvcfile._dump_pipeline_file(stage)
initial_content["copy-first-line"]["cmd"] = stage.cmd
initial_content["stages"]["copy-first-line"]["cmd"] = stage.cmd
assert dvc.reproduce(stage.addressing) == [stage]
@@ -121,7 +131,7 @@ def test_params_dump(tmp_dir, dvc, run_head):
)
initial_content = read_lock_file()
lock = initial_content["copy-first-line"]
lock = initial_content["stages"]["copy-first-line"]
# lock stage key order:
assert list(lock.keys()) == ["cmd", "deps", "params", "outs"]
@@ -155,3 +165,64 @@ def test_params_dump(tmp_dir, dvc, run_head):
remove(item)
assert dvc.reproduce(stage.addressing) == [stage]
assert_eq_lockfile(initial_content, read_lock_file())
@pytest.fixture
def v1_repo_lock(tmp_dir, dvc):
"""Generates a repo having v1 format lockfile"""
size = 5 if os.name == "nt" else 4
hi = HashInfo(
name="md5", size=size, value="c157a79031e1c40f85931829bc5fc552"
)
v1_lockdata = {
"foo": {"cmd": "echo foo"},
"bar": {
"cmd": "echo bar>bar.txt",
"outs": [{"path": "bar.txt", **hi.to_dict()}],
},
}
dvc.run(cmd="echo foo", name="foo", no_exec=True)
dvc.run(cmd="echo bar>bar.txt", outs=["bar.txt"], name="bar", no_exec=True)
dump_yaml(tmp_dir / "dvc.lock", v1_lockdata)
yield v1_lockdata
def test_can_read_v1_lockfile(tmp_dir, dvc, v1_repo_lock):
assert dvc.status() == {
"bar": [
{"changed outs": {"bar.txt": "not in cache"}},
"always changed",
],
"foo": ["always changed"],
}
def test_migrates_v1_lockfile_to_v2_during_dump(
tmp_dir, dvc, v1_repo_lock, caplog
):
caplog.clear()
with caplog.at_level(logging.INFO, logger="dvc.dvcfile"):
assert dvc.reproduce()
assert "Migrating lock file 'dvc.lock' from v1 to v2" in caplog.messages
d = load_yaml(tmp_dir / "dvc.lock")
assert d == {"stages": v1_repo_lock, "meta": {"version": "2.0"}}
@pytest.mark.parametrize(
"version_info",
[{"version": "1.1"}, {"version": "2.1"}, {"version": "3.0"}],
)
def test_lockfile_invalid_versions(tmp_dir, dvc, version_info):
lockdata = {"meta": version_info, "stages": {"foo": {"cmd": "echo foo"}}}
dump_yaml("dvc.lock", lockdata)
with pytest.raises(LockfileCorruptedError) as exc_info:
Lockfile(dvc, tmp_dir / "dvc.lock").load()
assert str(exc_info.value) == "Lockfile 'dvc.lock' is corrupted."
assert (
str(exc_info.value.__cause__) == "'dvc.lock' format error: "
f"invalid version {version_info['version']}, "
"expected one of ['2.0'] for dictionary value @ "
"data['meta']['version']"
)
tests/func/test_repro_multistage.py
+1 -1
View File
@@ -502,7 +502,7 @@ def test_repro_multiple_params(tmp_dir, dvc):
assert len(stage.outs) == 1
lockfile = stage.dvcfile._lockfile
assert lockfile.load()["read_params"]["params"] == {
assert lockfile.load()["stages"]["read_params"]["params"] == {
"params2.yaml": {
"lists": [42, 42.0, "42"],
"floats": 42.0,
tests/func/test_run_multistage.py
+21 -22
View File
@@ -34,14 +34,10 @@ def test_run_no_exec(tmp_dir, dvc, run_copy):
assert not os.path.exists(PIPELINE_LOCK)
data, _ = stage.dvcfile._load()
assert data == {
"stages": {
"copy-foo-to-bar": {
"cmd": "python copy.py foo bar",
"deps": ["copy.py", "foo"],
"outs": ["bar"],
}
}
assert data["stages"]["copy-foo-to-bar"] == {
"cmd": "python copy.py foo bar",
"deps": ["copy.py", "foo"],
"outs": ["bar"],
}
@@ -172,7 +168,7 @@ def test_run_dump_on_multistage(tmp_dir, dvc, run_head):
"always_changed": True,
"wdir": "dir",
},
}
},
}
run_head(
@@ -197,7 +193,7 @@ def test_run_dump_on_multistage(tmp_dir, dvc, run_head):
"metrics": [{"foobar-1": {"cache": False}}],
},
**data["stages"],
}
},
}
@@ -244,7 +240,7 @@ def test_run_params_default(tmp_dir, dvc):
assert stage.deps[0].params == ["nested.nested1.nested2"]
lockfile = stage.dvcfile._lockfile
assert lockfile.load()["read_params"]["params"] == {
assert lockfile.load()["stages"]["read_params"]["params"] == {
"params.yaml": {"nested.nested1.nested2": "42"}
}
@@ -267,7 +263,7 @@ def test_run_params_custom_file(tmp_dir, dvc):
isinstance(stage.deps[0], ParamsDependency)
assert stage.deps[0].params == ["lists"]
lockfile = stage.dvcfile._lockfile
assert lockfile.load()["read_params"]["params"] == {
assert lockfile.load()["stages"]["read_params"]["params"] == {
"params2.yaml": {"lists": [42, 42.0, "42"]}
}
@@ -424,16 +420,19 @@ def test_run_external_outputs(
assert (tmp_dir / "dvc.yaml").read_text() == dvc_yaml
assert (tmp_dir / "dvc.lock").read_text() == (
"mystage:\n"
" cmd: mycmd\n"
" deps:\n"
" - path: remote://workspace/foo\n"
f" {hash_name}: {foo_hash}\n"
" size: 3\n"
" outs:\n"
" - path: remote://workspace/bar\n"
f" {hash_name}: {bar_hash}\n"
" size: 3\n"
"meta:\n"
" version: '2.0'\n"
"stages:\n"
" mystage:\n"
" cmd: mycmd\n"
" deps:\n"
" - path: remote://workspace/foo\n"
f" {hash_name}: {foo_hash}\n"
" size: 3\n"
" outs:\n"
" - path: remote://workspace/bar\n"
f" {hash_name}: {bar_hash}\n"
" size: 3\n"
)
assert (workspace / "foo").read_text() == "foo"
tests/unit/stage/test_serialize_pipeline_lock.py
+2 -2
View File
@@ -5,7 +5,7 @@ from voluptuous import Schema as _Schema
from dvc.dvcfile import PIPELINE_FILE
from dvc.hash_info import HashInfo
from dvc.schema import LOCK_FILE_STAGE_SCHEMA, LOCKFILE_SCHEMA
from dvc.schema import LOCK_FILE_STAGE_SCHEMA, LOCKFILE_STAGES_SCHEMA
from dvc.stage import PipelineStage, create_stage
from dvc.stage.serialize import DEFAULT_PARAMS_FILE, to_lockfile
from dvc.stage.serialize import (
@@ -227,7 +227,7 @@ def test_to_lockfile(dvc):
stage.deps[0].hash_info = HashInfo("md5", "md-five")
entry = to_lockfile(stage)
assert len(entry) == 1
_Schema(LOCKFILE_SCHEMA)(entry)
_Schema(LOCKFILE_STAGES_SCHEMA)(entry)
assert entry == {
"something": OrderedDict(
[
tests/unit/test_lockfile.py
+10 -6
View File
@@ -9,7 +9,10 @@ def test_stage_dump_no_outs_deps(tmp_dir, dvc):
stage = PipelineStage(name="s1", repo=dvc, path="path", cmd="command")
lockfile = Lockfile(dvc, "path.lock")
lockfile.dump(stage)
assert lockfile.load() == {"s1": {"cmd": "command"}}
assert lockfile.load() == {
"stages": {"s1": {"cmd": "command"}},
"meta": {"version": "2.0"},
}
def test_stage_dump_when_already_exists(tmp_dir, dvc):
@@ -19,8 +22,8 @@ def test_stage_dump_when_already_exists(tmp_dir, dvc):
lockfile = Lockfile(dvc, "path.lock")
lockfile.dump(stage)
assert lockfile.load() == {
**data,
"s2": {"cmd": "command2"},
"stages": {**data, "s2": {"cmd": "command2"}},
"meta": {"version": "2.0"},
}
@@ -37,8 +40,8 @@ def test_stage_dump_with_deps_and_outs(tmp_dir, dvc):
stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command2")
lockfile.dump(stage)
assert lockfile.load() == {
**data,
"s2": {"cmd": "command2"},
"stages": {**data, "s2": {"cmd": "command2"}},
"meta": {"version": "2.0"},
}
@@ -49,7 +52,8 @@ def test_stage_overwrites_if_already_exists(tmp_dir, dvc):
stage = PipelineStage(name="s2", repo=dvc, path="path", cmd="command3")
lockfile.dump(stage)
assert lockfile.load() == {
"s2": {"cmd": "command3"},
"stages": {"s2": {"cmd": "command3"}},
"meta": {"version": "2.0"},
}