minor refactor on check-ignore (#10854)

This commit is contained in:
skshetry
2025-08-27 19:14:29 +05:45
committed by GitHub
parent 79454557bb
commit 76cfe92960
5 changed files with 146 additions and 81 deletions
+9 -8
View File
@@ -10,11 +10,14 @@ class CmdCheckIgnore(CmdBase):
self.ignore_filter = self.repo.dvcignore
def _show_results(self, result):
if not result.match and not self.args.non_matching:
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
patterns = result.patterns or ["::"]
if not self.args.all:
patterns = patterns[-1:]
@@ -22,13 +25,11 @@ class CmdCheckIgnore(CmdBase):
ui.write(pattern, result.file, sep="\t")
else:
ui.write(result.file)
return bool(result.patterns)
def _check_one_file(self, target):
result = self.ignore_filter.check_ignore(target)
self._show_results(result)
if result.match:
return 0
return 1
return bool(self._show_results(result))
def _interactive_mode(self):
ret = 1
@@ -39,14 +40,14 @@ class CmdCheckIgnore(CmdBase):
break
if not target:
break
if not self._check_one_file(target):
if self._check_one_file(target):
ret = 0
return ret
def _normal_mode(self):
ret = 1
for target in self.args.targets:
if not self._check_one_file(target):
if self._check_one_file(target):
ret = 0
return ret
+123 -66
View File
@@ -1,8 +1,8 @@
import os
import re
from collections import namedtuple
from collections.abc import Iterable, Iterator
from itertools import chain, groupby, takewhile
from typing import TYPE_CHECKING, Optional
from typing import TYPE_CHECKING, Any, Literal, NamedTuple, Optional, Union, overload
from pathspec.patterns import GitWildMatchPattern
from pathspec.util import normalize_file
@@ -13,7 +13,9 @@ from dvc.log import logger
from dvc.pathspec_math import PatternInfo, merge_patterns
if TYPE_CHECKING:
from dvc.fs import AnyFSPath, FileSystem
from typing_extensions import Self
from dvc.fs import FileSystem
logger = logger.getChild(__name__)
@@ -21,36 +23,42 @@ logger = logger.getChild(__name__)
class DvcIgnore:
DVCIGNORE_FILE = ".dvcignore"
def __call__(self, root, dirs, files):
def __call__(
self, root: str, dirs: list[str], files: list[str]
) -> tuple[list[str], list[str]]:
raise NotImplementedError
class DvcIgnorePatterns(DvcIgnore):
def __init__(self, pattern_list, dirname, sep):
def __init__(
self, pattern_list: Iterable[Union[PatternInfo, str]], dirname: str, sep: str
) -> None:
from pathspec.patterns.gitwildmatch import _DIR_MARK
if pattern_list and isinstance(pattern_list[0], str):
pattern_list = [PatternInfo(pattern, "") for pattern in pattern_list]
pattern_infos = [
pattern if isinstance(pattern, PatternInfo) else PatternInfo(pattern, "")
for pattern in pattern_list
]
self.sep = sep
self.pattern_list = pattern_list
self.pattern_list: list[PatternInfo] = pattern_infos
self.dirname = dirname
self.regex_pattern_list = []
for count, pattern in enumerate(pattern_list):
pattern, group = GitWildMatchPattern.pattern_to_regex(pattern.patterns)
if pattern:
pattern = pattern.replace(f"<{_DIR_MARK}>", f"<{_DIR_MARK}{count}>")
self.regex_pattern_list.append((pattern, group))
self.regex_pattern_list: list[tuple[str, bool]] = []
for count, pattern in enumerate(pattern_infos):
regex, ignore = GitWildMatchPattern.pattern_to_regex(pattern.patterns)
if regex is not None and ignore is not None:
regex = regex.replace(f"<{_DIR_MARK}>", f"<{_DIR_MARK}{count}>")
self.regex_pattern_list.append((regex, ignore))
self.ignore_spec = [
(ignore, re.compile("|".join(item[0] for item in group)))
(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
]
@classmethod
def from_file(cls, path, fs, name):
def from_file(cls, path: str, fs: "FileSystem", name: str) -> "Self":
assert fs.isabs(path)
dirname = fs.normpath(fs.dirname(path))
with fs.open(path, encoding="utf-8") as fobj:
@@ -62,13 +70,15 @@ class DvcIgnorePatterns(DvcIgnore):
return cls(path_spec_lines, dirname, fs.sep)
def __call__(self, root: list[str], dirs: list[str], files: list[str]):
def __call__(
self, root: str, dirs: list[str], files: list[str]
) -> tuple[list[str], list[str]]:
files = [f for f in files if not self.matches(root, f)]
dirs = [d for d in dirs if not self.matches(root, d, True)]
return dirs, files
def _get_normalize_path(self, dirname, basename):
def _get_normalize_path(self, dirname: str, basename: str) -> Optional[str]:
# NOTE: `relpath` is too slow, so we have to assume that both
# `dirname` and `self.dirname` are relative or absolute together.
@@ -81,22 +91,46 @@ class DvcIgnorePatterns(DvcIgnore):
# NOTE: `os.path.join` is ~x5.5 slower
path = f"{rel}{self.sep}{basename}"
else:
return False
return None
if os.name == "nt":
return normalize_file(path)
return path
def matches(self, dirname, basename, is_dir=False, details: bool = False):
@overload
def matches(
self,
dirname: str,
basename: str,
is_dir: bool = False,
details: Literal[False] = ...,
) -> bool: ...
@overload
def matches(
self,
dirname: str,
basename: str,
is_dir: bool = False,
details: Literal[True] = ...,
) -> tuple[bool, list[str]]: ...
@overload
def matches(
self, dirname: str, basename: str, is_dir: bool = False, details: bool = False
) -> Union[bool, tuple[bool, list[str]]]: ...
def matches(
self, dirname: str, basename: str, is_dir: bool = False, details: bool = False
) -> Union[bool, tuple[bool, list[str]]]:
path = self._get_normalize_path(dirname, basename)
if not path:
return False
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, is_dir):
def ignore(self, path: str, is_dir: bool) -> bool:
def matches(pattern, path, is_dir) -> bool:
matches_ = bool(pattern.match(path))
@@ -113,30 +147,31 @@ class DvcIgnorePatterns(DvcIgnore):
break
return result
def _ignore_details(self, path, is_dir: bool):
result = []
for (regex, _), pattern_info in list(
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)
):
# skip system pattern
if not pattern_info.file_info:
continue
regex = re.compile(regex)
pattern = re.compile(regex)
matches = bool(regex.match(path))
matches = bool(pattern.match(path))
if is_dir:
matches |= bool(regex.match(f"{path}/"))
matches |= bool(pattern.match(f"{path}/"))
if matches:
result.append(pattern_info.file_info)
matched_patterns.append(pattern_info.file_info)
result = ignore
return result, matched_patterns
return result
def __hash__(self):
def __hash__(self) -> int:
return hash(self.dirname + ":" + str(self.pattern_list))
def __eq__(self, other):
def __eq__(self, other: object) -> bool:
if not isinstance(other, DvcIgnorePatterns):
return NotImplemented
return (self.dirname == other.dirname) & (
@@ -144,21 +179,18 @@ class DvcIgnorePatterns(DvcIgnore):
== [pattern.patterns for pattern in other.pattern_list]
)
def __bool__(self):
def __bool__(self) -> bool:
return bool(self.pattern_list)
CheckIgnoreResult = namedtuple( # noqa: PYI024
"CheckIgnoreResult", ["file", "match", "patterns"]
)
def _no_match(path):
return CheckIgnoreResult(path, False, ["::"])
class CheckIgnoreResult(NamedTuple):
file: str
match: bool
patterns: list[str]
class DvcIgnoreFilter:
def __init__(self, fs, root_dir):
def __init__(self, fs: "FileSystem", root_dir: str) -> None:
from dvc.repo import Repo
default_ignore_patterns = [
@@ -193,7 +225,7 @@ class DvcIgnoreFilter:
ignore_subrepos=True,
)
def _get_key(self, path):
def _get_key(self, path: str) -> tuple[str, ...]:
parts = self.fs.relparts(path, self.root_dir)
if parts == (os.curdir,):
return ()
@@ -241,7 +273,7 @@ class DvcIgnoreFilter:
for dname in dnames:
self._update_sub_repo(self.fs.join(dirname, dname), ignore_trie)
def _update_sub_repo(self, path, ignore_trie: Trie):
def _update_sub_repo(self, path: str, ignore_trie: Trie) -> None:
from dvc.repo import Repo
if path == self.root_dir:
@@ -268,7 +300,9 @@ class DvcIgnoreFilter:
else:
ignore_trie[key] = new_pattern
def __call__(self, root, dirs, files, ignore_subrepos=True):
def __call__(
self, root: str, dirs: list[str], files: list[str], ignore_subrepos: bool = True
) -> tuple[list[str], list[str]]:
abs_root = self.fs.abspath(root)
ignore_pattern = self._get_trie_pattern(
abs_root, dnames=dirs, ignore_subrepos=ignore_subrepos
@@ -277,7 +311,24 @@ class DvcIgnoreFilter:
dirs, files = ignore_pattern(abs_root, dirs, files)
return dirs, files
def ls(self, fs, path, detail=True, **kwargs):
@overload
def ls(
self, fs: "FileSystem", path: str, detail: Literal[True], **kwargs: Any
) -> list[dict[str, Any]]: ...
@overload
def ls(
self, fs: "FileSystem", path: str, detail: Literal[False], **kwargs
) -> list[str]: ...
@overload
def ls(
self, fs: "FileSystem", path: str, detail: bool = True, **kwargs
) -> Union[list[str], list[dict[str, Any]]]: ...
def ls(
self, fs: "FileSystem", path: str, detail: bool = True, **kwargs: Any
) -> Union[list[str], list[dict[str, Any]]]:
fs_dict = {}
dirs = []
nondirs = []
@@ -297,22 +348,29 @@ class DvcIgnoreFilter:
return [fs_dict[name] for name in chain(dirs, nondirs)]
def walk(self, fs: "FileSystem", path: "AnyFSPath", **kwargs):
def walk(
self, fs: "FileSystem", path: str, **kwargs: Any
) -> Iterator[
Union[
tuple[str, list[str], list[str]],
tuple[str, dict[str, dict], dict[str, dict]],
]
]:
detail = kwargs.get("detail", False)
ignore_subrepos = kwargs.pop("ignore_subrepos", True)
if fs.protocol == Schemes.LOCAL:
for root, dirs, files in fs.walk(path, **kwargs):
if detail:
all_dnames = set(dirs.keys())
all_fnames = set(files.keys())
assert isinstance(dirs, dict)
assert isinstance(files, dict)
dnames, fnames = self(
root,
all_dnames,
all_fnames,
list(dirs),
list(files),
ignore_subrepos=ignore_subrepos,
)
list(map(dirs.pop, all_dnames - set(dnames)))
list(map(files.pop, all_fnames - set(fnames)))
list(map(dirs.pop, dirs.keys() - set(dnames)))
list(map(files.pop, files.keys() - set(fnames)))
else:
dirs[:], files[:] = self(
root, dirs, files, ignore_subrepos=ignore_subrepos
@@ -321,7 +379,7 @@ class DvcIgnoreFilter:
else:
yield from fs.walk(path, **kwargs)
def find(self, fs: "FileSystem", path: "AnyFSPath", **kwargs):
def find(self, fs: "FileSystem", path: str, **kwargs: Any) -> Iterator[str]:
if fs.protocol == Schemes.LOCAL:
for root, _, files in self.walk(fs, path, **kwargs):
for file in files:
@@ -331,7 +389,7 @@ class DvcIgnoreFilter:
yield from fs.find(path)
def _get_trie_pattern(
self, dirname, dnames: Optional["list"] = None, ignore_subrepos=True
self, dirname: str, dnames: Optional[list[str]] = None, ignore_subrepos=True
) -> Optional["DvcIgnorePatterns"]:
if ignore_subrepos:
ignores_trie = self.ignores_trie_fs
@@ -367,7 +425,7 @@ class DvcIgnoreFilter:
def _is_ignored(
self, path: str, is_dir: bool = False, ignore_subrepos: bool = True
):
) -> bool:
if self._outside_repo(path):
return False
dirname, basename = self.fs.split(self.fs.normpath(path))
@@ -389,24 +447,23 @@ class DvcIgnoreFilter:
path = self.fs.abspath(path)
return self._is_ignored(path, False, ignore_subrepos=ignore_subrepos)
def _outside_repo(self, path):
def _outside_repo(self, path: str) -> bool:
return not self.fs.isin_or_eq(path, self.root_dir)
def check_ignore(self, target):
def check_ignore(self, target: str) -> CheckIgnoreResult:
# 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] = []
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:
matches = pattern.matches(
ignore, matched_patterns = pattern.matches(
dirname, basename, self.fs.isdir(full_target), True
)
if matches:
return CheckIgnoreResult(target, True, matches)
return _no_match(target)
return CheckIgnoreResult(target, ignore, matched_patterns)
def is_ignored(
self, fs: "FileSystem", path: str, ignore_subrepos: bool = True
@@ -424,7 +481,7 @@ class DvcIgnoreFilter:
)
def init(path):
def init(path: Union[str, os.PathLike[str]]) -> str:
dvcignore = os.path.join(path, DvcIgnore.DVCIGNORE_FILE)
if os.path.exists(dvcignore):
return dvcignore
@@ -439,7 +496,7 @@ def init(path):
return dvcignore
def destroy(path):
def destroy(path: Union[str, os.PathLike[str]]) -> None:
from dvc.utils.fs import remove
dvcignore = os.path.join(path, DvcIgnore.DVCIGNORE_FILE)
+5 -2
View File
@@ -2,13 +2,16 @@
# Including changing base dir of path specification patterns and merging
# of two path specification patterns with different base
# All the operations follow the documents of `gitignore`
from collections import namedtuple
from typing import NamedTuple
from pathspec.util import normalize_file
from dvc.utils import relpath
PatternInfo = namedtuple("PatternInfo", ["patterns", "file_info"]) # noqa: PYI024
class PatternInfo(NamedTuple):
patterns: str
file_info: str
def _not_ignore(rule):
+5 -3
View File
@@ -67,14 +67,14 @@ def collect_files(
is_local_fs = isinstance(fs, LocalFileSystem)
def is_ignored(path):
def is_ignored(path: str) -> bool:
# apply only for the local fs
return is_local_fs and scm.is_ignored(path)
def is_dvcfile_and_not_ignored(root, file):
def is_dvcfile_and_not_ignored(root: str, file: str) -> bool:
return is_valid_filename(file) and not is_ignored(f"{root}{sep}{file}")
def is_out_or_ignored(root, directory):
def is_out_or_ignored(root: str, directory: str) -> bool:
dir_path = f"{root}{sep}{directory}"
# trailing slash needed to check if a directory is gitignored
return dir_path in outs or is_ignored(f"{dir_path}{sep}")
@@ -84,6 +84,8 @@ def collect_files(
walk_iter = log_walk(walk_iter)
for root, dirs, files in walk_iter:
assert isinstance(dirs, list)
assert isinstance(files, list)
dvcfile_filter = partial(is_dvcfile_and_not_ignored, root)
for file in filter(dvcfile_filter, files):
file_path = fs.join(root, file)
+4 -2
View File
@@ -6,7 +6,7 @@ from funcy import first
from voluptuous import MultipleInvalid, Schema
from dvc.fs import RemoteMissingDepsError
from dvc.ignore import _no_match
from dvc.ignore import CheckIgnoreResult
from dvc.output import CHECKSUM_SCHEMA, Output
from dvc.stage import Stage
from dvc.utils.fs import remove
@@ -88,7 +88,9 @@ def test_get_used_objs(exists, expected_message, mocker, caplog):
mocker.patch.object(stage.repo, "root_dir", os.getcwd())
mocker.patch.object(stage.repo.dvcignore, "is_ignored", return_value=False)
mocker.patch.object(
stage.repo.dvcignore, "check_ignore", return_value=_no_match("path")
stage.repo.dvcignore,
"check_ignore",
return_value=CheckIgnoreResult("path", False, []),
)
stage.repo.fs.version_aware = False
stage.repo.fs.PARAM_CHECKSUM = "md5"