diff --git a/dvc/output/__init__.py b/dvc/output/__init__.py index a7630e5b6..43d2d2a28 100644 --- a/dvc/output/__init__.py +++ b/dvc/output/__init__.py @@ -110,16 +110,9 @@ def loadd_from(stage, d_list): 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, - persist=persist, - ) + return [ + _get( + stage, s, info={}, cache=use_cache, metric=metric, persist=persist, ) - return ret + for s in s_list + ] diff --git a/dvc/schema.py b/dvc/schema.py index 9695a9bfe..429e006af 100644 --- a/dvc/schema.py +++ b/dvc/schema.py @@ -1,8 +1,8 @@ from voluptuous import Any, Optional, Required, Schema from dvc import dependency, output -from dvc.output import CHECKSUMS_SCHEMA -from dvc.stage.params import OutputParams, StageParams +from dvc.output import CHECKSUMS_SCHEMA, BaseOutput +from dvc.stage.params import StageParams STAGES = "stages" SINGLE_STAGE_SCHEMA = { @@ -25,16 +25,29 @@ LOCK_FILE_STAGE_SCHEMA = { } LOCKFILE_SCHEMA = {str: LOCK_FILE_STAGE_SCHEMA} +OUT_PSTAGE_DETAILED_SCHEMA = { + str: { + BaseOutput.PARAM_CACHE: bool, + BaseOutput.PARAM_PERSIST: bool, + BaseOutput.PARAM_METRIC: bool, + } +} +PARAM_PSTAGE_NON_DEFAULT_SCHEMA = {str: [str]} + SINGLE_PIPELINE_STAGE_SCHEMA = { str: { StageParams.PARAM_CMD: str, Optional(StageParams.PARAM_WDIR): str, Optional(StageParams.PARAM_DEPS): [str], - Optional(StageParams.PARAM_PARAMS): [Any(str, {str: [str]})], + Optional(StageParams.PARAM_PARAMS): [ + Any(str, PARAM_PSTAGE_NON_DEFAULT_SCHEMA) + ], Optional(StageParams.PARAM_LOCKED): bool, Optional(StageParams.PARAM_META): object, Optional(StageParams.PARAM_ALWAYS_CHANGED): bool, - **{Optional(p.value): [str] for p in OutputParams}, + Optional(StageParams.PARAM_OUTS): [ + Any(str, OUT_PSTAGE_DETAILED_SCHEMA) + ], } } MULTI_STAGE_SCHEMA = {STAGES: SINGLE_PIPELINE_STAGE_SCHEMA} diff --git a/dvc/serialize.py b/dvc/serialize.py index e5aa55087..00487f92f 100644 --- a/dvc/serialize.py +++ b/dvc/serialize.py @@ -6,6 +6,8 @@ from typing import TYPE_CHECKING, List from funcy import lsplit, rpartial from dvc.dependency import ParamsDependency +from dvc.output import BaseOutput +from dvc.stage.params import StageParams from dvc.stage.utils import resolve_wdir from dvc.utils.collections import apply_diff from dvc.utils.stage import parse_stage_for_update @@ -13,26 +15,35 @@ from dvc.utils.stage import parse_stage_for_update if TYPE_CHECKING: from dvc.stage import PipelineStage, Stage -PARAM_PATH = ParamsDependency.PARAM_PATH PARAM_PARAMS = ParamsDependency.PARAM_PARAMS +PARAM_PATH = ParamsDependency.PARAM_PATH + +PARAM_DEPS = StageParams.PARAM_DEPS +PARAM_OUTS = StageParams.PARAM_OUTS + +PARAM_CACHE = BaseOutput.PARAM_CACHE +PARAM_METRIC = BaseOutput.PARAM_METRIC +PARAM_PERSIST = BaseOutput.PARAM_PERSIST + DEFAULT_PARAMS_FILE = ParamsDependency.DEFAULT_PARAMS_FILE sort_by_path = partial(sorted, key=attrgetter("def_path")) -def _get_outs(stage: "PipelineStage"): - outs_bucket = {} - for o in sort_by_path(stage.outs): - bucket_key = ["metrics"] if o.metric else ["outs"] +def _get_out(out): + res = OrderedDict() + if not out.use_cache: + res[PARAM_CACHE] = False + if out.metric: + res[PARAM_METRIC] = True + elif out.persist: + res[PARAM_PERSIST] = True + return out.def_path if not res else {out.def_path: res} - if not o.metric and o.persist: - bucket_key += ["persist"] - if not o.use_cache: - bucket_key += ["no_cache"] - key = "_".join(bucket_key) - outs_bucket[key] = outs_bucket.get(key, []) + [o.def_path] - return [(key, outs_bucket[key]) for key in sorted(outs_bucket.keys())] + +def _get_outs(outs): + return [_get_out(out) for out in sort_by_path(outs)] def get_params_deps(stage: "PipelineStage"): @@ -81,7 +92,7 @@ def to_pipeline_file(stage: "PipelineStage"): (stage.PARAM_WDIR, resolve_wdir(stage.wdir, stage.path)), (stage.PARAM_DEPS, sorted([d.def_path for d in deps])), (stage.PARAM_PARAMS, serialized_params), - *_get_outs(stage), + (PARAM_OUTS, _get_outs(stage.outs)), (stage.PARAM_LOCKED, stage.locked), (stage.PARAM_ALWAYS_CHANGED, stage.always_changed), ] @@ -98,18 +109,21 @@ def to_single_stage_lockfile(stage: "Stage") -> dict: deps, outs = [ [ OrderedDict( - [("path", item.def_path), (item.checksum_type, item.checksum)] + [ + (PARAM_PATH, item.def_path), + (item.checksum_type, item.checksum), + ] ) for item in sort_by_path(items) ] for items in [deps, stage.outs] ] if deps: - res["deps"] = deps + res[PARAM_DEPS] = deps if params: - _, res["params"] = _serialize_params(params) + _, res[PARAM_PARAMS] = _serialize_params(params) if outs: - res["outs"] = outs + res[PARAM_OUTS] = outs return res diff --git a/dvc/stage/loader.py b/dvc/stage/loader.py index ffae8348e..4f30a9d44 100644 --- a/dvc/stage/loader.py +++ b/dvc/stage/loader.py @@ -11,7 +11,6 @@ from dvc import dependency, output from ..dependency import ParamsDependency from .exceptions import StageNameUnspecified, StageNotFound -from .utils import fill_stage_dependencies, fill_stage_outputs logger = logging.getLogger(__name__) @@ -112,11 +111,35 @@ class StageLoader(Mapping): path = first(key) res[path].extend(key[path]) - stage.deps += dependency.loadd_from( - stage, - [{"path": key, "params": params} for key, params in res.items()], + stage.deps.extend( + dependency.loadd_from( + stage, + [ + {"path": key, "params": params} + for key, params in res.items() + ], + ) ) + @classmethod + def _load_outs(cls, stage, data): + d = [] + for key in data: + extra_kwargs = {} + if isinstance(key, str): + path = key + elif isinstance(key, dict): + path = first(key) + extra_kwargs = key[path] + else: + continue + d.append({"path": path, **extra_kwargs}) + stage.outs.extend(output.loadd_from(stage, d)) + + @classmethod + def _load_deps(cls, stage, data): + stage.deps.extend(dependency.loads_from(stage, data)) + @classmethod def load_stage(cls, dvcfile, name, stage_data, lock_data): from . import PipelineStage, Stage, loads_from @@ -126,9 +149,12 @@ class StageLoader(Mapping): ) stage = loads_from(PipelineStage, dvcfile.repo, path, wdir, stage_data) stage.name = name - fill_stage_dependencies(stage, deps=stage_data.get("deps", [])) - fill_stage_outputs(stage, **stage_data) + stage.deps, stage.outs = [], [] + + cls._load_outs(stage, stage_data.get("outs", [])) + cls._load_deps(stage, stage_data.get("deps", [])) cls._load_params(stage, stage_data.get("params", [])) + if lock_data: stage.cmd_changed = lock_data.get( Stage.PARAM_CMD diff --git a/tests/dir_helpers.py b/tests/dir_helpers.py index b550a5366..136c5c383 100644 --- a/tests/dir_helpers.py +++ b/tests/dir_helpers.py @@ -46,6 +46,7 @@ from global repo template to creating everything inplace, which: import os import pathlib from contextlib import contextmanager +from textwrap import dedent import pytest from funcy import lmap, retry @@ -59,6 +60,7 @@ __all__ = [ "scm", "dvc", "run_copy", + "run_head", "erepo_dir", "git_dir", "setup_remote", @@ -270,6 +272,39 @@ def run_copy(tmp_dir, dvc): return run_copy +@pytest.fixture +def run_head(tmp_dir, dvc): + """Output first line of each file to different file with '-1' appended. + Useful for tracking multiple outputs/dependencies which are not a copy + of each others. + """ + tmp_dir.gen( + { + "head.py": dedent( + """ + import sys + for file in sys.argv[1:]: + with open(file) as f, open(file +"-1","w+") as w: + w.write(f.readline()) + """ + ) + } + ) + script = os.path.abspath(tmp_dir / "head.py") + + def run(*args, **run_kwargs): + return dvc.run( + **{ + "cmd": "python {} {}".format(script, " ".join(args)), + "outs": [dep + "-1" for dep in args], + "deps": list(args), + **run_kwargs, + } + ) + + return run + + @pytest.fixture def erepo_dir(make_tmp_dir): return make_tmp_dir("erepo", scm=True, dvc=True) diff --git a/tests/func/test_lockfile.py b/tests/func/test_lockfile.py index d22d5ca5c..caac97e5f 100644 --- a/tests/func/test_lockfile.py +++ b/tests/func/test_lockfile.py @@ -1,8 +1,6 @@ from collections import OrderedDict from operator import itemgetter -from textwrap import dedent -import pytest import yaml from dvc.dvcfile import PIPELINE_LOCK @@ -20,32 +18,6 @@ FS_STRUCTURE = { } -@pytest.fixture -def run_head(tmp_dir, dvc): - """Output first line of each file to different file with '-1' appended.""" - tmp_dir.gen( - "head.py", - dedent( - """ - import sys - for file in sys.argv[1:]: - with open(file) as f, open(file +"-1","w+") as w: - w.write(f.readline()) - """ - ), - ) - - def run(*args, **run_kwargs): - return dvc.run( - cmd="python head.py {}".format(" ".join(args)), - outs=[dep + "-1" for dep in args], - deps=args, - **run_kwargs - ) - - return run - - def read_lock_file(file=PIPELINE_LOCK): with open(file) as f: data = parse_stage_for_update(f.read(), file) diff --git a/tests/func/test_run_multistage.py b/tests/func/test_run_multistage.py index bb8e7d489..13a357727 100644 --- a/tests/func/test_run_multistage.py +++ b/tests/func/test_run_multistage.py @@ -115,28 +115,18 @@ def test_graph(tmp_dir, dvc): dvc.run(deps=["baz"], outs=["foo"], cmd="echo baz > foo", name="3") -def test_run_dump_on_multistage(tmp_dir, dvc): +def test_run_dump_on_multistage(tmp_dir, dvc, run_head): from dvc.dvcfile import Dvcfile, PIPELINE_FILE - tmp_dir.gen({"dir": {"foo": "foo", "bar": "bar"}}) - dvc.run( - cmd="cp foo foo1", - deps=["foo"], - name="copy-foo-foo1", - outs=["foo1"], - wdir="dir", - ) - data, _ = Dvcfile(dvc, PIPELINE_FILE)._load() - assert data == { - "stages": { - "copy-foo-foo1": { - "cmd": "cp foo foo1", - "wdir": "dir", - "deps": ["foo"], - "outs": ["foo1"], + tmp_dir.gen( + { + "dir": { + "foo": "foo\nfoo", + "bar": "bar\nbar", + "foobar": "foobar\foobar", } } - } + ) dvc.run( cmd="cp foo foo2", @@ -146,15 +136,43 @@ def test_run_dump_on_multistage(tmp_dir, dvc): outs_persist=["foo2"], always_changed=True, ) - assert Dvcfile(dvc, PIPELINE_FILE)._load()[0] == { + data = Dvcfile(dvc, PIPELINE_FILE)._load()[0] + assert data == { "stages": { "copy-foo-foo2": { "cmd": "cp foo foo2", "deps": ["foo"], - "outs_persist": ["foo2"], + "outs": [{"foo2": {"persist": True}}], "always_changed": True, "wdir": "dir", }, + } + } + + run_head( + "foo", + "bar", + "foobar", + name="head-files", + outs=["bar-1"], + outs_persist=["foo-1"], + metrics_no_cache=["foobar-1"], + wdir="dir", + ) + assert Dvcfile(dvc, PIPELINE_FILE)._load()[0] == { + "stages": { + "head-files": { + "cmd": "python {} foo bar foobar".format( + (tmp_dir / "head.py").resolve() + ), + "wdir": "dir", + "deps": ["bar", "foo", "foobar"], + "outs": [ + "bar-1", + {"foo-1": {"persist": True}}, + {"foobar-1": {"metric": True, "cache": False}}, + ], + }, **data["stages"], } }