repro: support glob/foreach-group to run at once through CLI (#4976)

* repro: support regex/foreach-group to run at once

Fixes #4912
Fixes #4886
Fixes #4958

* Use `tree.isdir` rather than `os.path.isdir`

* Use glob rather than regex

* Update dvc/command/repro.py

* s/regex/glob

* disable glob on `collect_granular`

There's no need for a glob here

* add tests for `collect` and `collect_granular`
This commit is contained in:
Saugat Pachhai
2020-11-27 19:13:38 +05:45
committed by GitHub
parent 011bd18e9f
commit f4f0554884
11 changed files with 353 additions and 101 deletions
+7
View File
@@ -63,6 +63,7 @@ class CmdRepro(CmdBase):
"recursive": self.args.recursive,
"force_downstream": self.args.force_downstream,
"pull": self.args.pull,
"glob": self.args.glob,
}
@@ -175,6 +176,12 @@ def add_arguments(repro_parser):
"from the run-cache."
),
)
repro_parser.add_argument(
"--glob",
action="store_true",
default=False,
help="Allows targets containing shell-style wildcards.",
)
def add_parser(subparsers, parent_parser):
+76 -56
View File
@@ -1,14 +1,14 @@
import logging
import os
from contextlib import contextmanager
from contextlib import contextmanager, suppress
from functools import wraps
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, List
from funcy import cached_property, cat
from git import InvalidGitRepositoryError
from dvc.config import Config
from dvc.dvcfile import PIPELINE_FILE, Dvcfile, is_valid_filename
from dvc.dvcfile import PIPELINE_FILE, is_valid_filename
from dvc.exceptions import FileMissingError
from dvc.exceptions import IsADirectoryError as DvcIsADirectoryError
from dvc.exceptions import (
@@ -17,6 +17,7 @@ from dvc.exceptions import (
OutputNotFoundError,
)
from dvc.path_info import PathInfo
from dvc.repo.stage import StageLoad
from dvc.scm import Base
from dvc.scm.base import SCMError
from dvc.tree.repo import RepoTree
@@ -28,6 +29,9 @@ from .graph import build_graph, build_outs_graph, get_pipeline, get_pipelines
from .trie import build_outs_trie
if TYPE_CHECKING:
from networkx import DiGraph
from dvc.stage import Stage
from dvc.tree.base import BaseTree
@@ -165,6 +169,7 @@ class Repo:
self.cache = Cache(self)
self.cloud = DataCloud(self)
self.stage = StageLoad(self)
if scm or not self.dvc_dir:
self.lock = LockNoop()
@@ -270,25 +275,6 @@ class Repo:
self.scm.ignore_list(flist)
def get_stage(self, path=None, name=None):
if not path:
path = PIPELINE_FILE
logger.debug("Assuming '%s' to be a stage inside '%s'", name, path)
dvcfile = Dvcfile(self, path)
return dvcfile.stages[name]
def get_stages(self, path=None, name=None):
if not path:
path = PIPELINE_FILE
logger.debug("Assuming '%s' to be a stage inside '%s'", name, path)
if name:
return [self.get_stage(path, name)]
dvcfile = Dvcfile(self, path)
return list(dvcfile.stages.values())
def check_modified_graph(self, new_stages):
"""Generate graph including the new stage to check for errors"""
# Building graph might be costly for the ones with many DVC-files,
@@ -306,46 +292,89 @@ class Repo:
if not getattr(self, "_skip_graph_checks", False):
build_graph(self.stages + new_stages)
def _collect_inside(self, path, graph):
@staticmethod
def _collect_inside(path: str, graph: "DiGraph"):
import networkx as nx
stages = nx.dfs_postorder_nodes(graph)
return [stage for stage in stages if path_isin(stage.path, path)]
def collect(
self, target=None, with_deps=False, recursive=False, graph=None
self,
target: str = None,
with_deps: bool = False,
recursive: bool = False,
graph: "DiGraph" = None,
accept_group: bool = False,
glob: bool = False,
):
if not target:
return list(graph) if graph else self.stages
if recursive and os.path.isdir(target):
if recursive and self.tree.isdir(target):
return self._collect_inside(
os.path.abspath(target), graph or self.graph
)
path, name = parse_target(target)
stages = self.get_stages(path, name)
stages = self.stage.from_target(
target, accept_group=accept_group, glob=glob
)
if not with_deps:
return stages
return self._collect_stages_with_deps(stages, graph=graph)
def _collect_stages_with_deps(
self, stages: List["Stage"], graph: "DiGraph" = None
):
res = set()
for stage in stages:
res.update(self._collect_pipeline(stage, graph=graph))
return res
def _collect_pipeline(self, stage, graph=None):
def _collect_pipeline(self, stage: "Stage", graph: "DiGraph" = None):
import networkx as nx
pipeline = get_pipeline(get_pipelines(graph or self.graph), stage)
return nx.dfs_postorder_nodes(pipeline, stage)
def _collect_from_default_dvcfile(self, target):
dvcfile = Dvcfile(self, PIPELINE_FILE)
if dvcfile.exists():
return dvcfile.stages.get(target)
def _collect_specific_target(
self, target: str, with_deps: bool, recursive: bool, accept_group: bool
):
# Optimization: do not collect the graph for a specific target
file, name = parse_target(target)
stages = []
if not file:
# parsing is ambiguous when it does not have a colon
# or if it's not a dvcfile, as it can be a stage name
# in `dvc.yaml` or, an output in a stage.
logger.debug(
"Checking if stage '%s' is in '%s'", target, PIPELINE_FILE
)
if not (
recursive and self.tree.isdir(target)
) and self.tree.exists(PIPELINE_FILE):
with suppress(StageNotFound):
stages = self.stage.load_all(
PIPELINE_FILE, target, accept_group=accept_group
)
if with_deps:
stages = self._collect_stages_with_deps(stages)
elif not with_deps and is_valid_filename(file):
stages = self.stage.load_all(
file, name, accept_group=accept_group,
)
return stages, file, name
def collect_granular(
self, target=None, with_deps=False, recursive=False, graph=None
self,
target: str = None,
with_deps: bool = False,
recursive: bool = False,
graph: "DiGraph" = None,
accept_group: bool = False,
):
"""
Priority is in the order of following in case of ambiguity:
@@ -357,28 +386,11 @@ class Repo:
if not target:
return [(stage, None) for stage in self.stages]
file, name = parse_target(target)
stages = []
# Optimization: do not collect the graph for a specific target
if not file:
# parsing is ambiguous when it does not have a colon
# or if it's not a dvcfile, as it can be a stage name
# in `dvc.yaml` or, an output in a stage.
logger.debug(
"Checking if stage '%s' is in '%s'", target, PIPELINE_FILE
)
if not (recursive and os.path.isdir(target)):
stage = self._collect_from_default_dvcfile(target)
if stage:
stages = (
self._collect_pipeline(stage) if with_deps else [stage]
)
elif not with_deps and is_valid_filename(file):
stages = self.get_stages(file, name)
stages, file, _ = self._collect_specific_target(
target, with_deps, recursive, accept_group
)
if not stages:
if not (recursive and os.path.isdir(target)):
if not (recursive and self.tree.isdir(target)):
try:
(out,) = self.find_outs_by_path(target, strict=False)
filter_info = PathInfo(os.path.abspath(target))
@@ -387,7 +399,13 @@ class Repo:
pass
try:
stages = self.collect(target, with_deps, recursive, graph)
stages = self.collect(
target,
with_deps,
recursive,
graph,
accept_group=accept_group,
)
except StageFileDoesNotExistError as exc:
# collect() might try to use `target` as a stage name
# and throw error that dvc.yaml does not exist, whereas it
@@ -498,7 +516,9 @@ class Repo:
for root, dirs, files in self.tree.walk(self.root_dir):
for file_name in filter(is_valid_filename, files):
new_stages = self.get_stages(os.path.join(root, file_name))
new_stages = self.stage.load_file(
os.path.join(root, file_name)
)
stages.extend(new_stages)
outs.update(
out.fspath
+7 -5
View File
@@ -1,12 +1,14 @@
import typing
from . import locked
if typing.TYPE_CHECKING:
from . import Repo
@locked
def _set(repo, target, frozen):
from dvc.utils import parse_target
path, name = parse_target(target)
stage = repo.get_stage(path, name)
def _set(repo: "Repo", target, frozen):
stage = repo.stage.get_target(target)
stage.frozen = frozen
stage.dvcfile.dump(stage, update_lock=False)
+6 -4
View File
@@ -1,15 +1,17 @@
import logging
import typing
from ..utils import parse_target
from . import locked
if typing.TYPE_CHECKING:
from dvc.repo import Repo
logger = logging.getLogger(__name__)
@locked
def remove(self, target, outs=False):
path, name = parse_target(target)
stages = self.get_stages(path, name)
def remove(self: "Repo", target: str, outs: bool = False):
stages = self.stage.from_target(target)
for stage in stages:
stage.remove(remove_outs=outs, force=outs)
+15 -6
View File
@@ -1,4 +1,5 @@
import logging
import typing
from functools import partial
from dvc.exceptions import InvalidArgumentError, ReproductionError
@@ -8,6 +9,9 @@ from dvc.stage.run import CheckpointKilledError
from . import locked
from .graph import get_pipeline, get_pipelines
if typing.TYPE_CHECKING:
from . import Repo
logger = logging.getLogger(__name__)
@@ -75,15 +79,15 @@ def _get_active_graph(G):
@locked
@scm_context
def reproduce(
self,
self: "Repo",
target=None,
recursive=False,
pipeline=False,
all_pipelines=False,
**kwargs,
):
from dvc.utils import parse_target
glob = kwargs.pop("glob", False)
accept_group = not glob
assert target is None or isinstance(target, str)
if not target and not all_pipelines:
raise InvalidArgumentError(
@@ -97,12 +101,11 @@ def reproduce(
active_graph = _get_active_graph(self.graph)
active_pipelines = get_pipelines(active_graph)
path, name = parse_target(target)
if pipeline or all_pipelines:
if all_pipelines:
pipelines = active_pipelines
else:
stage = self.get_stage(path, name)
stage = self.stage.get_target(target)
pipelines = [get_pipeline(active_pipelines, stage)]
targets = []
@@ -111,7 +114,13 @@ def reproduce(
if pipeline.in_degree(stage) == 0:
targets.append(stage)
else:
targets = self.collect(target, recursive=recursive, graph=active_graph)
targets = self.collect(
target,
recursive=recursive,
graph=active_graph,
accept_group=accept_group,
glob=glob,
)
return _reproduce_stages(active_graph, targets, **kwargs)
+130
View File
@@ -0,0 +1,130 @@
import fnmatch
import logging
import typing
from typing import Iterable, List
from dvc.dvcfile import PIPELINE_FILE, Dvcfile
from dvc.utils import parse_target
logger = logging.getLogger(__name__)
if typing.TYPE_CHECKING:
from dvc.repo import Repo
from dvc.stage import Stage
from dvc.stage.loader import StageLoader
class StageLoad:
def __init__(self, repo: "Repo") -> None:
self.repo = repo
def from_target(
self, target: str, accept_group: bool = False, glob: bool = False,
) -> List["Stage"]:
"""
Returns a list of stage from the provided target.
(see load method below for further details)
"""
path, name = parse_target(target, isa_glob=glob)
return self.load_all(
path=path, name=name, accept_group=accept_group, glob=glob,
)
def get_target(self, target: str) -> "Stage":
"""
Returns a stage from the provided target.
(see load_one method for further details)
"""
path, name = parse_target(target)
return self.load_one(path=path, name=name)
@staticmethod
def _get_filepath(path: str = None, name: str = None) -> str:
if path:
return path
path = PIPELINE_FILE
logger.debug("Assuming '%s' to be a stage inside '%s'", name, path)
return path
@staticmethod
def _get_group_keys(stages: "StageLoader", group: str) -> Iterable[str]:
from dvc.parsing import JOIN
for key in stages:
assert isinstance(key, str)
if key.startswith(f"{group}{JOIN}"):
yield key
def _get_keys(
self,
stages: "StageLoader",
name: str = None,
accept_group: bool = False,
glob: bool = False,
) -> Iterable[str]:
assert not (accept_group and glob)
if not name:
return stages.keys()
if accept_group and stages.is_foreach_generated(name):
return self._get_group_keys(stages, name)
elif glob:
return fnmatch.filter(stages.keys(), name)
return [name]
def load_all(
self,
path: str = None,
name: str = None,
accept_group: bool = False,
glob: bool = False,
) -> List["Stage"]:
"""Load a list of stages from a file.
Args:
path: if not provided, default `dvc.yaml` is assumed.
name: required for `dvc.yaml` files, ignored for `.dvc` files.
accept_group: if true, all of the the stages generated from `name`
foreach are returned.
glob: if true, `name` is considered as a glob, which is
used to filter list of stages from the given `path`.
"""
from dvc.stage.loader import SingleStageLoader, StageLoader
path = self._get_filepath(path, name)
dvcfile = Dvcfile(self.repo, path)
# `dvcfile.stages` is not cached
stages = dvcfile.stages # type: ignore
if isinstance(stages, SingleStageLoader):
stage = stages[name]
return [stage]
assert isinstance(stages, StageLoader)
keys = self._get_keys(stages, name, accept_group, glob)
return [stages[key] for key in keys]
def load_one(self, path: str = None, name: str = None) -> "Stage":
"""Load a single stage from a file.
Args:
path: if not provided, default `dvc.yaml` is assumed.
name: required for `dvc.yaml` files, ignored for `.dvc` files.
"""
path = self._get_filepath(path, name)
dvcfile = Dvcfile(self.repo, path)
stages = dvcfile.stages # type: ignore
return stages[name]
def load_file(self, path: str = None) -> List["Stage"]:
"""Load all of the stages from a file."""
return self.load_all(path)
def load_glob(self, path: str, expr: str = None):
"""Load stages from `path`, filtered with `expr` provided."""
return self.load_all(path, expr, glob=True)
+6 -1
View File
@@ -7,7 +7,7 @@ from funcy import cached_property, get_in, lcat, log_durations, project
from dvc import dependency, output
from dvc.hash_info import HashInfo
from dvc.parsing import JOIN, DataResolver
from dvc.parsing import FOREACH_KWD, JOIN, DataResolver
from dvc.path_info import PathInfo
from . import PipelineStage, Stage, loads_from
@@ -133,6 +133,11 @@ class StageLoader(Mapping):
def __contains__(self, name):
return name in self.resolved_data
def is_foreach_generated(self, name: str):
return (
name in self.stages_data and FOREACH_KWD in self.stages_data[name]
)
class SingleStageLoader(Mapping):
def __init__(self, dvcfile, stage_data, stage_text=None):
+12 -5
View File
@@ -9,6 +9,7 @@ import re
import stat
import sys
import time
from typing import Optional, Tuple
import colorama
import nanotime
@@ -409,7 +410,9 @@ def error_link(name):
return format_link(f"https://error.dvc.org/{name}")
def parse_target(target: str, default: str = None):
def parse_target(
target: str, default: str = None, isa_glob: bool = False
) -> Tuple[Optional[str], Optional[str]]:
from dvc.dvcfile import PIPELINE_FILE, PIPELINE_LOCK, is_valid_filename
from dvc.exceptions import DvcException
from dvc.parsing import JOIN
@@ -417,9 +420,14 @@ def parse_target(target: str, default: str = None):
if not target:
return None, None
default = default or PIPELINE_FILE
if isa_glob:
path, _, glob = target.rpartition(":")
return path or default, glob or None
# look for first "@", so as not to assume too much about stage name
# eg: it might contain ":" in a generated stages from dict which might
# affect further parsings with the regex.
# affect further parsing with the regex.
group, _, key = target.partition(JOIN)
match = TARGET_REGEX.match(group)
@@ -446,10 +454,9 @@ def parse_target(target: str, default: str = None):
return ret if is_valid_filename(target) else ret[::-1]
if not path:
path = default or PIPELINE_FILE
logger.debug("Assuming file to be '%s'", path)
logger.debug("Assuming file to be '%s'", default)
return path, name
return path or default, name
def is_exec(mode):
+2 -2
View File
@@ -355,7 +355,7 @@ def test_dvcfile_dump_preserves_comments(tmp_dir, dvc):
- foo"""
)
tmp_dir.gen("dvc.yaml", text)
stage = dvc.get_stage(name="generate-foo")
stage = dvc.stage.load_one(name="generate-foo")
stage.outs[0].use_cache = False
dvcfile = stage.dvcfile
@@ -377,7 +377,7 @@ def test_dvcfile_try_dumping_parametrized_stage(tmp_dir, dvc, data, name):
dump_yaml("dvc.yaml", {"stages": data, "vars": [{"foo": "foobar"}]})
dvc.config["feature"]["parametrization"] = True
stage = dvc.get_stage(name=name)
stage = dvc.stage.load_one(name=name)
dvcfile = stage.dvcfile
with pytest.raises(ParametrizedDumpError) as exc:
+91 -22
View File
@@ -16,6 +16,7 @@ from dvc.stage.exceptions import (
from dvc.system import System
from dvc.utils import relpath
from dvc.utils.fs import remove
from dvc.utils.serialize import dump_yaml
def test_destroy(tmp_dir, dvc, run_copy):
@@ -160,7 +161,73 @@ def stages(tmp_dir, run_copy):
"lorem-generate": stage2,
"copy-foo-bar": run_copy("foo", "bar", single_stage=True),
"copy-bar-foobar": run_copy("bar", "foobar", name="copy-bar-foobar"),
"copy-lorem-ipsum": run_copy("lorem", "ipsum", name="lorem-ipsum"),
"copy-lorem-ipsum": run_copy(
"lorem", "ipsum", name="copy-lorem-ipsum"
),
}
def test_collect_not_a_group_stage_with_group_flag(tmp_dir, dvc, stages):
assert set(dvc.collect("copy-bar-foobar", accept_group=True)) == {
stages["copy-bar-foobar"]
}
assert set(
dvc.collect("copy-bar-foobar", accept_group=True, with_deps=True)
) == {
stages["copy-bar-foobar"],
stages["copy-foo-bar"],
stages["foo-generate"],
}
assert set(dvc.collect_granular("copy-bar-foobar", accept_group=True)) == {
(stages["copy-bar-foobar"], None),
}
assert set(
dvc.collect_granular(
"copy-bar-foobar", accept_group=True, with_deps=True
)
) == {
(stages["copy-bar-foobar"], None),
(stages["copy-foo-bar"], None),
(stages["foo-generate"], None),
}
def test_collect_generated(tmp_dir, dvc):
dvc.config["feature"]["parametrization"] = True
d = {
"vars": [{"vars": [1, 2, 3, 4, 5]}],
"stages": {
"build": {"foreach": "${vars}", "do": {"cmd": "echo ${item}"}}
},
}
dump_yaml("dvc.yaml", d)
all_stages = set(dvc.stages)
assert len(all_stages) == 5
assert set(dvc.collect()) == all_stages
assert set(dvc.collect("build", accept_group=True)) == all_stages
assert (
set(dvc.collect("build", accept_group=True, with_deps=True))
== all_stages
)
assert set(dvc.collect("build*", glob=True)) == all_stages
assert set(dvc.collect("build*", glob=True, with_deps=True)) == all_stages
stages_info = {(stage, None) for stage in all_stages}
assert set(dvc.collect_granular("build", accept_group=True)) == stages_info
assert (
set(dvc.collect_granular("build", accept_group=True, with_deps=True))
== stages_info
)
def test_collect_glob(tmp_dir, dvc, stages):
assert set(dvc.collect("copy*", glob=True)) == {
stages[key] for key in ["copy-bar-foobar", "copy-lorem-ipsum"]
}
assert set(dvc.collect("copy-lorem*", glob=True, with_deps=True)) == {
stages[key] for key in ["copy-lorem-ipsum", "lorem-generate"]
}
@@ -307,33 +374,33 @@ def test_collect_granular_not_existing_stage_name(tmp_dir, dvc, run_copy):
def test_get_stages(tmp_dir, dvc, run_copy):
with pytest.raises(StageFileDoesNotExistError):
dvc.get_stages()
dvc.stage.load_all()
tmp_dir.gen("foo", "foo")
stage1 = run_copy("foo", "bar", name="copy-foo-bar")
stage2 = run_copy("bar", "foobar", name="copy-bar-foobar")
assert set(dvc.get_stages()) == {stage1, stage2}
assert set(dvc.get_stages(path=PIPELINE_FILE)) == {stage1, stage2}
assert set(dvc.get_stages(name="copy-bar-foobar")) == {stage2}
assert set(dvc.get_stages(path=PIPELINE_FILE, name="copy-bar-foobar")) == {
stage2
}
assert set(dvc.stage.load_all()) == {stage1, stage2}
assert set(dvc.stage.load_all(path=PIPELINE_FILE)) == {stage1, stage2}
assert set(dvc.stage.load_all(name="copy-bar-foobar")) == {stage2}
assert set(
dvc.stage.load_all(path=PIPELINE_FILE, name="copy-bar-foobar")
) == {stage2}
with pytest.raises(StageFileDoesNotExistError):
dvc.get_stages(path=relpath(tmp_dir / ".." / PIPELINE_FILE))
dvc.stage.load_all(path=relpath(tmp_dir / ".." / PIPELINE_FILE))
with pytest.raises(StageNotFound):
dvc.get_stages(path=PIPELINE_FILE, name="copy")
dvc.stage.load_all(path=PIPELINE_FILE, name="copy")
def test_get_stages_old_dvcfile(tmp_dir, dvc):
(stage1,) = tmp_dir.dvc_gen("foo", "foo")
assert set(dvc.get_stages("foo.dvc")) == {stage1}
assert set(dvc.get_stages("foo.dvc", name="foo-generate")) == {stage1}
assert set(dvc.stage.load_all("foo.dvc")) == {stage1}
assert set(dvc.stage.load_all("foo.dvc", name="foo-generate")) == {stage1}
with pytest.raises(StageFileDoesNotExistError):
dvc.get_stages(path=relpath(tmp_dir / ".." / "foo.dvc"))
dvc.stage.load_all(path=relpath(tmp_dir / ".." / "foo.dvc"))
def test_get_stage(tmp_dir, dvc, run_copy):
@@ -341,24 +408,26 @@ def test_get_stage(tmp_dir, dvc, run_copy):
stage1 = run_copy("foo", "bar", name="copy-foo-bar")
with pytest.raises(StageNameUnspecified):
dvc.get_stage()
dvc.stage.load_one()
with pytest.raises(StageNameUnspecified):
dvc.get_stage(path=PIPELINE_FILE)
dvc.stage.load_one(path=PIPELINE_FILE)
assert dvc.get_stage(path=PIPELINE_FILE, name="copy-foo-bar") == stage1
assert dvc.get_stage(name="copy-foo-bar") == stage1
assert (
dvc.stage.load_one(path=PIPELINE_FILE, name="copy-foo-bar") == stage1
)
assert dvc.stage.load_one(name="copy-foo-bar") == stage1
with pytest.raises(StageFileDoesNotExistError):
dvc.get_stage(path="something.yaml", name="name")
dvc.stage.load_one(path="something.yaml", name="name")
with pytest.raises(StageNotFound):
dvc.get_stage(name="random_name")
dvc.stage.load_one(name="random_name")
def test_get_stage_single_stage_dvcfile(tmp_dir, dvc):
(stage1,) = tmp_dir.dvc_gen("foo", "foo")
assert dvc.get_stage("foo.dvc") == stage1
assert dvc.get_stage("foo.dvc", name="jpt") == stage1
assert dvc.stage.load_one("foo.dvc") == stage1
assert dvc.stage.load_one("foo.dvc", name="jpt") == stage1
with pytest.raises(StageFileDoesNotExistError):
dvc.get_stage(path="bar.dvc", name="name")
dvc.stage.load_one(path="bar.dvc", name="name")
+1
View File
@@ -15,6 +15,7 @@ default_arguments = {
"recursive": False,
"force_downstream": False,
"pull": False,
"glob": False,
}