mirror of
https://github.com/Stone-Red-Code/dvc.git
synced 2026-11-01 23:08:43 +01:00
ignore: simplify and reuse same implementation across check-ignore and is_ignored API (#10856)
This commit is contained in:
skshetry
@@ -1,23 +1,28 @@
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from dvc.cli import completion, formatter
|
||||
from dvc.cli.command import CmdBase
|
||||
from dvc.cli.utils import append_doc_link
|
||||
from dvc.ui import ui
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from dvc.ignore import CheckIgnoreResult
|
||||
|
||||
|
||||
class CmdCheckIgnore(CmdBase):
|
||||
def __init__(self, args):
|
||||
super().__init__(args)
|
||||
self.ignore_filter = self.repo.dvcignore
|
||||
|
||||
def _show_results(self, result):
|
||||
def _show_results(self, result: "CheckIgnoreResult"):
|
||||
if not result.match and not self.args.details:
|
||||
return
|
||||
|
||||
if not result.patterns and not self.args.non_matching:
|
||||
return
|
||||
|
||||
if self.args.details:
|
||||
patterns = result.patterns or ["::"]
|
||||
pattern_infos = result.pattern_infos
|
||||
patterns = [str(pi) for pi in pattern_infos]
|
||||
if not patterns and self.args.non_matching:
|
||||
patterns = ["::"]
|
||||
if not self.args.all:
|
||||
patterns = patterns[-1:]
|
||||
|
||||
@@ -25,7 +30,7 @@ class CmdCheckIgnore(CmdBase):
|
||||
ui.write(pattern, result.file, sep="\t")
|
||||
else:
|
||||
ui.write(result.file)
|
||||
return bool(result.patterns)
|
||||
return bool(result.pattern_infos)
|
||||
|
||||
def _check_one_file(self, target):
|
||||
result = self.ignore_filter.check_ignore(target)
|
||||
@@ -77,6 +82,8 @@ class CmdCheckIgnore(CmdBase):
|
||||
|
||||
|
||||
def add_parser(subparsers, parent_parser):
|
||||
import argparse
|
||||
|
||||
ADD_HELP = "Check whether files or directories are excluded due to `.dvcignore`."
|
||||
|
||||
parser = subparsers.add_parser(
|
||||
@@ -94,14 +101,7 @@ def add_parser(subparsers, parent_parser):
|
||||
help="Show the exclude patterns along with each target path.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"-a",
|
||||
"--all",
|
||||
action="store_true",
|
||||
default=False,
|
||||
help=(
|
||||
"Include the target paths which don't match any pattern "
|
||||
"in the `--details` list."
|
||||
),
|
||||
"-a", "--all", action="store_true", default=False, help=argparse.SUPPRESS
|
||||
)
|
||||
parser.add_argument(
|
||||
"-n",
|
||||
|
||||
+110
-54
@@ -1,3 +1,4 @@
|
||||
import functools
|
||||
import os
|
||||
import re
|
||||
from collections.abc import Iterable, Iterator
|
||||
@@ -41,21 +42,58 @@ class DvcIgnorePatterns(DvcIgnore):
|
||||
]
|
||||
|
||||
self.sep = sep
|
||||
self.pattern_list: list[PatternInfo] = pattern_infos
|
||||
self.pattern_list: list[PatternInfo] = []
|
||||
self.dirname = dirname
|
||||
self.find_matching_pattern = functools.cache(self._find_matching_pattern)
|
||||
|
||||
self.regex_pattern_list: list[tuple[str, bool]] = []
|
||||
for count, pattern in enumerate(pattern_infos):
|
||||
regex, ignore = GitWildMatchPattern.pattern_to_regex(pattern.patterns)
|
||||
regex_pattern_list: list[tuple[str, bool, bool, PatternInfo]] = []
|
||||
for count, pattern_info in enumerate(pattern_infos):
|
||||
regex, ignore = GitWildMatchPattern.pattern_to_regex(pattern_info.patterns)
|
||||
if regex is not None and ignore is not None:
|
||||
self.pattern_list.append(pattern_info)
|
||||
regex = regex.replace(f"<{_DIR_MARK}>", f"<{_DIR_MARK}{count}>")
|
||||
self.regex_pattern_list.append((regex, ignore))
|
||||
regex_pattern_list.append(
|
||||
(regex, ignore, pattern_info.patterns.endswith("/"), pattern_info)
|
||||
)
|
||||
|
||||
self.ignore_spec = [
|
||||
(ignore, re.compile("|".join(regex for regex, _ in group)))
|
||||
for ignore, group in groupby(self.regex_pattern_list, lambda x: x[1])
|
||||
if ignore is not None
|
||||
def keyfunc(item: tuple[str, bool, bool, PatternInfo]) -> tuple[bool, bool]:
|
||||
_, ignore, dir_only_pattern, _ = item
|
||||
return ignore, dir_only_pattern
|
||||
|
||||
self.ignore_spec: list[
|
||||
tuple[
|
||||
re.Pattern[str],
|
||||
bool,
|
||||
bool,
|
||||
dict[Optional[str], tuple[str, PatternInfo]],
|
||||
]
|
||||
]
|
||||
self.ignore_spec = []
|
||||
for (ignore, dir_only_pattern), group in groupby(
|
||||
regex_pattern_list, key=keyfunc
|
||||
):
|
||||
if ignore:
|
||||
# For performance, we combine all exclude patterns.
|
||||
# But we still need to figure out which pattern matched which rule,
|
||||
# (eg: to show in `dvc check-ignore`).
|
||||
# So, we use named groups and keep a map of group name to pattern.
|
||||
pattern_map: dict[Optional[str], tuple[str, PatternInfo]] = {
|
||||
f"rule_{i}": (regex, pi)
|
||||
for i, (regex, _, _, pi) in enumerate(group)
|
||||
}
|
||||
combined_regex = "|".join(
|
||||
f"(?P<{name}>{regex})" for name, (regex, _) in pattern_map.items()
|
||||
)
|
||||
self.ignore_spec.append(
|
||||
(re.compile(combined_regex), ignore, dir_only_pattern, pattern_map)
|
||||
)
|
||||
else:
|
||||
# unignored patterns are not combined with `|`.
|
||||
for regex, _, _, pi in group:
|
||||
pattern_map = {None: (regex, pi)}
|
||||
self.ignore_spec.append(
|
||||
(re.compile(regex), ignore, dir_only_pattern, pattern_map)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, path: str, fs: "FileSystem", name: str) -> "Self":
|
||||
@@ -113,60 +151,78 @@ class DvcIgnorePatterns(DvcIgnore):
|
||||
basename: str,
|
||||
is_dir: bool = False,
|
||||
details: Literal[True] = ...,
|
||||
) -> tuple[bool, list[str]]: ...
|
||||
) -> tuple[bool, list[PatternInfo]]: ...
|
||||
|
||||
@overload
|
||||
def matches(
|
||||
self, dirname: str, basename: str, is_dir: bool = False, details: bool = False
|
||||
) -> Union[bool, tuple[bool, list[str]]]: ...
|
||||
self,
|
||||
dirname: str,
|
||||
basename: str,
|
||||
is_dir: bool = False,
|
||||
details: bool = False,
|
||||
) -> Union[bool, tuple[bool, list[PatternInfo]]]: ...
|
||||
|
||||
def matches(
|
||||
self, dirname: str, basename: str, is_dir: bool = False, details: bool = False
|
||||
) -> Union[bool, tuple[bool, list[str]]]:
|
||||
self,
|
||||
dirname: str,
|
||||
basename: str,
|
||||
is_dir: bool = False,
|
||||
details: bool = False,
|
||||
) -> Union[bool, tuple[bool, list[PatternInfo]]]:
|
||||
path = self._get_normalize_path(dirname, basename)
|
||||
if not path:
|
||||
return (False, []) if details else False
|
||||
if details:
|
||||
return self._ignore_details(path, is_dir)
|
||||
return self.ignore(path, is_dir)
|
||||
|
||||
def ignore(self, path: str, is_dir: bool) -> bool:
|
||||
def matches(pattern, path, is_dir) -> bool:
|
||||
matches_ = bool(pattern.match(path))
|
||||
|
||||
if is_dir:
|
||||
matches_ |= bool(pattern.match(f"{path}/"))
|
||||
|
||||
return matches_
|
||||
|
||||
result = False
|
||||
_match: list[PatternInfo] = []
|
||||
if path:
|
||||
result, _match = self._ignore(path, is_dir)
|
||||
return (result, _match) if details else result
|
||||
|
||||
for ignore, pattern in self.ignore_spec[::-1]:
|
||||
if matches(pattern, path, is_dir):
|
||||
result = ignore
|
||||
break
|
||||
return result
|
||||
def _find_matching_pattern(
|
||||
self, path: str, is_dir: bool
|
||||
) -> tuple[bool, list[PatternInfo]]:
|
||||
paths = [path]
|
||||
if is_dir and not path.endswith("/"):
|
||||
paths.append(f"{path}/")
|
||||
|
||||
def _ignore_details(self, path: str, is_dir: bool) -> tuple[bool, list[str]]:
|
||||
result = False
|
||||
matched_patterns = []
|
||||
for (regex, ignore), pattern_info in list(
|
||||
zip(self.regex_pattern_list, self.pattern_list)
|
||||
for pattern, ignore, dir_only_pattern, pattern_map in reversed(
|
||||
self.ignore_spec
|
||||
):
|
||||
# skip system pattern
|
||||
if not pattern_info.file_info:
|
||||
if dir_only_pattern and not is_dir:
|
||||
continue
|
||||
for p in paths:
|
||||
match = pattern.match(p)
|
||||
if not match:
|
||||
continue
|
||||
if ignore:
|
||||
group_name, _match = next(
|
||||
(
|
||||
(name, _match)
|
||||
for name, _match in match.groupdict().items()
|
||||
if name.startswith("rule_") and _match is not None
|
||||
)
|
||||
)
|
||||
else:
|
||||
# unignored patterns are not combined with `|`,
|
||||
# so there are no groups.
|
||||
group_name = None
|
||||
_regex, pattern_info = pattern_map[group_name]
|
||||
return ignore, [pattern_info]
|
||||
return False, []
|
||||
|
||||
pattern = re.compile(regex)
|
||||
|
||||
matches = bool(pattern.match(path))
|
||||
if is_dir:
|
||||
matches |= bool(pattern.match(f"{path}/"))
|
||||
|
||||
if matches:
|
||||
matched_patterns.append(pattern_info.file_info)
|
||||
result = ignore
|
||||
return result, matched_patterns
|
||||
def _ignore(self, path: str, is_dir: bool) -> tuple[bool, list[PatternInfo]]:
|
||||
parts = path.split("/")
|
||||
result = False
|
||||
matches: list[PatternInfo] = []
|
||||
for i in range(1, len(parts) + 1):
|
||||
rel_path = "/".join(parts[:i])
|
||||
result, _matches = self.find_matching_pattern(
|
||||
rel_path, is_dir or i < len(parts)
|
||||
)
|
||||
if i < len(parts) and not result:
|
||||
continue
|
||||
matches.extend(_matches)
|
||||
if result:
|
||||
break
|
||||
return result, matches
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash(self.dirname + ":" + str(self.pattern_list))
|
||||
@@ -186,7 +242,7 @@ class DvcIgnorePatterns(DvcIgnore):
|
||||
class CheckIgnoreResult(NamedTuple):
|
||||
file: str
|
||||
match: bool
|
||||
patterns: list[str]
|
||||
pattern_infos: list[PatternInfo]
|
||||
|
||||
|
||||
class DvcIgnoreFilter:
|
||||
@@ -454,14 +510,14 @@ class DvcIgnoreFilter:
|
||||
# NOTE: can only be used in `dvc check-ignore`, see
|
||||
# https://github.com/iterative/dvc/issues/5046
|
||||
full_target = self.fs.abspath(target)
|
||||
matched_patterns: list[str] = []
|
||||
matched_patterns: list[PatternInfo] = []
|
||||
ignore = False
|
||||
if not self._outside_repo(full_target):
|
||||
dirname, basename = self.fs.split(self.fs.normpath(full_target))
|
||||
pattern = self._get_trie_pattern(dirname)
|
||||
if pattern:
|
||||
ignore, matched_patterns = pattern.matches(
|
||||
dirname, basename, self.fs.isdir(full_target), True
|
||||
dirname, basename, self.fs.isdir(full_target), details=True
|
||||
)
|
||||
return CheckIgnoreResult(target, ignore, matched_patterns)
|
||||
|
||||
|
||||
+6
-5
@@ -45,7 +45,7 @@ if TYPE_CHECKING:
|
||||
from dvc_data.hashfile.obj import HashFile
|
||||
from dvc_data.index import DataIndexKey
|
||||
|
||||
from .ignore import DvcIgnoreFilter
|
||||
from .ignore import CheckIgnoreResult, DvcIgnoreFilter
|
||||
|
||||
logger = logger.getChild(__name__)
|
||||
|
||||
@@ -267,9 +267,9 @@ class OutputIsStageFileError(DvcException):
|
||||
|
||||
|
||||
class OutputIsIgnoredError(DvcException):
|
||||
def __init__(self, match):
|
||||
lines = "\n".join(match.patterns)
|
||||
super().__init__(f"Path '{match.file}' is ignored by\n{lines}")
|
||||
def __init__(self, result: "CheckIgnoreResult"):
|
||||
pi = result.pattern_infos[-1]
|
||||
super().__init__(f"Path {result.file!r} is ignored by {pi!s}")
|
||||
|
||||
|
||||
class CheckoutCallback(TqdmCallback):
|
||||
@@ -1203,7 +1203,8 @@ class Output:
|
||||
if stage:
|
||||
abs_path = os.path.join(stage.wdir, path)
|
||||
if self._is_path_dvcignore(abs_path):
|
||||
check = stage.repo.dvcignore.check_ignore(abs_path)
|
||||
check: CheckIgnoreResult = stage.repo.dvcignore.check_ignore(abs_path)
|
||||
if check.match:
|
||||
raise self.IsIgnoredError(check)
|
||||
|
||||
def _check_can_merge(self, out):
|
||||
|
||||
@@ -13,6 +13,9 @@ class PatternInfo(NamedTuple):
|
||||
patterns: str
|
||||
file_info: str
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.file_info or f":{self.patterns}"
|
||||
|
||||
|
||||
def _not_ignore(rule):
|
||||
return (True, rule[1:]) if rule.startswith("!") else (False, rule)
|
||||
|
||||
@@ -78,7 +78,7 @@ def test_check_ignore_dir(tmp_dir, dvc, path, ret):
|
||||
|
||||
|
||||
def test_check_ignore_default_dir(tmp_dir, dvc):
|
||||
assert main(["check-ignore", "-q", ".dvc"]) == 1
|
||||
assert main(["check-ignore", "-q", ".dvc"]) == 0
|
||||
|
||||
|
||||
def test_check_ignore_out_side_repo(tmp_dir, dvc):
|
||||
@@ -125,7 +125,6 @@ def test_check_ignore_details_all(tmp_dir, dvc, capsys):
|
||||
|
||||
assert main(["check-ignore", "-d", "-a", "foo"]) == 0
|
||||
out, _ = capsys.readouterr()
|
||||
assert f"{DvcIgnore.DVCIGNORE_FILE}:1:f*\tfoo\n" in out
|
||||
assert f"{DvcIgnore.DVCIGNORE_FILE}:2:!foo\tfoo\n" in out
|
||||
|
||||
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import os
|
||||
from os.path import join
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -202,3 +203,154 @@ def test_should_ignore_dir(omit_dir, sub_dir):
|
||||
|
||||
assert set(new_dirs) == {"dir1", "dir2"}
|
||||
assert set(new_files) == {"file1", "file2", omit_dir}
|
||||
|
||||
|
||||
def test_ignore_complex(tmp_dir, dvc):
|
||||
from dvc.fs import localfs
|
||||
|
||||
spec = """\
|
||||
# Ignore everything
|
||||
1/**
|
||||
# Except directories (leaves all files ignored)
|
||||
!1/**/
|
||||
# Don't ignore files in 3
|
||||
!seq/**/3/**
|
||||
|
||||
data/
|
||||
!data/keep.csv
|
||||
|
||||
data2/**
|
||||
!data2/**/
|
||||
!data2/**/*.csv
|
||||
|
||||
ignore.txt
|
||||
!no-ignore.txt
|
||||
"""
|
||||
(tmp_dir / ".dvcignore").write_text(spec)
|
||||
(tmp_dir / "1" / "2" / "3").mkdir(parents=True, exist_ok=True)
|
||||
(tmp_dir / "1" / "2" / "shouldIgnore.txt").touch()
|
||||
(tmp_dir / "1" / "2" / "3" / "shouldKeep.txt").touch()
|
||||
(tmp_dir / "data" / "subdir").mkdir(parents=True, exist_ok=True)
|
||||
(tmp_dir / "data2" / "subdir").mkdir(parents=True, exist_ok=True)
|
||||
(tmp_dir / "data" / "keep.csv").touch()
|
||||
(tmp_dir / "data" / "other.csv").touch()
|
||||
(tmp_dir / "data" / "subdir" / "file.txt").touch()
|
||||
(tmp_dir / "data2" / "keep.csv").touch()
|
||||
(tmp_dir / "data2" / "other.txt").touch()
|
||||
(tmp_dir / "data2" / "subdir" / "keep.csv").touch()
|
||||
(tmp_dir / "data2" / "subdir" / "other.txt").touch()
|
||||
(tmp_dir / "ignore.txt").touch()
|
||||
(tmp_dir / "no-ignore.txt").touch()
|
||||
|
||||
ignore_file = DvcIgnorePatterns.from_file(
|
||||
os.fspath(tmp_dir / ".dvcignore"), localfs, ".dvcignore"
|
||||
)
|
||||
dvc.__dict__.pop("dvcignore", None)
|
||||
|
||||
def matches(path):
|
||||
result, _matches = ignore_file.matches(
|
||||
os.fspath(tmp_dir), path, (tmp_dir / path).is_dir(), details=True
|
||||
)
|
||||
return result, [str(m) for m in _matches]
|
||||
|
||||
for path, *expected in [
|
||||
("1", False, [".dvcignore:4:!1/**/"]),
|
||||
(join("1", ""), False, [".dvcignore:4:!1/**/"]),
|
||||
(join("1", "2"), False, [".dvcignore:4:!1/**/"]),
|
||||
(join("1", "2", ""), False, [".dvcignore:4:!1/**/"]),
|
||||
(join("1", "2", "shouldIgnore.txt"), True, [".dvcignore:2:1/**"]),
|
||||
(join("1", "2", "3"), False, [".dvcignore:4:!1/**/"]),
|
||||
(join("1", "2", "3", ""), False, [".dvcignore:4:!1/**/"]),
|
||||
(join("1", "2", "3", "shouldKeep.txt"), True, [".dvcignore:2:1/**"]),
|
||||
("data", True, [".dvcignore:8:data/"]),
|
||||
(join("data", ""), True, [".dvcignore:8:data/"]),
|
||||
(join("data", "keep.csv"), True, [".dvcignore:8:data/"]),
|
||||
(join("data", "other.csv"), True, [".dvcignore:8:data/"]),
|
||||
(join("data", "subdir", "file.txt"), True, [".dvcignore:8:data/"]),
|
||||
("data2", False, [".dvcignore:12:!data2/**/"]),
|
||||
(join("data2", ""), False, [".dvcignore:12:!data2/**/"]),
|
||||
(join("data2", "keep.csv"), False, [".dvcignore:13:!data2/**/*.csv"]),
|
||||
(join("data2", "other.txt"), True, [".dvcignore:11:data2/**"]),
|
||||
(join("data2", "subdir"), False, [".dvcignore:12:!data2/**/"]),
|
||||
(join("data2", "subdir", ""), False, [".dvcignore:12:!data2/**/"]),
|
||||
(join("data2", "subdir", "keep.csv"), False, [".dvcignore:13:!data2/**/*.csv"]),
|
||||
(join("data2", "subdir", "other.txt"), True, [".dvcignore:11:data2/**"]),
|
||||
("ignore.txt", True, [".dvcignore:15:ignore.txt"]),
|
||||
("no-ignore.txt", False, [".dvcignore:16:!no-ignore.txt"]),
|
||||
]:
|
||||
assert matches(path) == tuple(expected), f"for {path}"
|
||||
|
||||
def sorted_walk(path):
|
||||
for root, dirs, files in dvc.dvcignore.walk(localfs, path):
|
||||
dirs.sort()
|
||||
files.sort()
|
||||
yield root, dirs, files
|
||||
|
||||
assert list(sorted_walk(os.curdir)) == [
|
||||
(
|
||||
os.curdir,
|
||||
["1", "data2"],
|
||||
[".dvcignore", "no-ignore.txt"],
|
||||
),
|
||||
("1", ["2"], []),
|
||||
(join("1", "2"), ["3"], []),
|
||||
(join("1", "2", "3"), [], []),
|
||||
("data2", ["subdir"], ["keep.csv"]),
|
||||
(join("data2", "subdir"), [], ["keep.csv"]),
|
||||
]
|
||||
|
||||
|
||||
def test_ignore_unignore_from_git_example(tmp_dir, dvc, scm):
|
||||
from dvc.fs import localfs
|
||||
|
||||
spec = """\
|
||||
# exclude everything except directory foo/bar
|
||||
/*
|
||||
!/foo
|
||||
/foo/*
|
||||
!/foo/bar
|
||||
"""
|
||||
(tmp_dir / ".dvcignore").write_text(spec)
|
||||
for d in [
|
||||
tmp_dir,
|
||||
tmp_dir / "foo",
|
||||
tmp_dir / "foo" / "bar",
|
||||
tmp_dir / "foo" / "baz",
|
||||
tmp_dir / "foobar",
|
||||
]:
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
(d / "myfile").touch()
|
||||
|
||||
ignore_file = DvcIgnorePatterns.from_file(
|
||||
os.fspath(tmp_dir / ".dvcignore"), localfs, ".dvcignore"
|
||||
)
|
||||
dvc.__dict__.pop("dvcignore", None)
|
||||
|
||||
def matches(path):
|
||||
result, _matches = ignore_file.matches(
|
||||
os.fspath(tmp_dir), path, (tmp_dir / path).is_dir(), details=True
|
||||
)
|
||||
return result, [str(m) for m in _matches]
|
||||
|
||||
for path, *expected in [
|
||||
("foo", False, [".dvcignore:3:!/foo"]),
|
||||
(join("foo", ""), False, [".dvcignore:3:!/foo"]),
|
||||
(join("foo", "myfile"), True, [".dvcignore:4:/foo/*"]),
|
||||
(join("foo", "bar"), False, [".dvcignore:5:!/foo/bar"]),
|
||||
(join("foo", "bar", ""), False, [".dvcignore:5:!/foo/bar"]),
|
||||
# matching pattern differs from git for foo/bar/myfile
|
||||
(join("foo", "bar", "myfile"), False, [".dvcignore:5:!/foo/bar"]),
|
||||
(join("foo", "baz"), True, [".dvcignore:4:/foo/*"]),
|
||||
(join("foo", "baz", ""), True, [".dvcignore:4:/foo/*"]),
|
||||
(join("foo", "baz", "myfile"), True, [".dvcignore:4:/foo/*"]),
|
||||
(join("foo", "foobar"), True, [".dvcignore:4:/foo/*"]),
|
||||
(join("foo", "foobar", ""), True, [".dvcignore:4:/foo/*"]),
|
||||
(join("foo", "foobar", "myfile"), True, [".dvcignore:4:/foo/*"]),
|
||||
]:
|
||||
assert matches(path) == tuple(expected), f"for {path}"
|
||||
|
||||
assert sorted(dvc.dvcignore.walk(localfs, os.curdir), key=lambda r: r[0]) == [
|
||||
(os.curdir, ["foo"], []),
|
||||
("foo", ["bar"], []),
|
||||
(join("foo", "bar"), [], ["myfile"]),
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user