From 76cfe929608017d9d2f760eca26259b26ea10368 Mon Sep 17 00:00:00 2001 From: skshetry <18718008+skshetry@users.noreply.github.com> Date: Wed, 27 Aug 2025 19:14:29 +0545 Subject: [PATCH] minor refactor on check-ignore (#10854) --- dvc/commands/check_ignore.py | 17 +-- dvc/ignore.py | 189 ++++++++++++++++++++----------- dvc/pathspec_math.py | 7 +- dvc/repo/index.py | 8 +- tests/unit/output/test_output.py | 6 +- 5 files changed, 146 insertions(+), 81 deletions(-) diff --git a/dvc/commands/check_ignore.py b/dvc/commands/check_ignore.py index d8f2597f9..cf49d6550 100644 --- a/dvc/commands/check_ignore.py +++ b/dvc/commands/check_ignore.py @@ -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 diff --git a/dvc/ignore.py b/dvc/ignore.py index a622fb1fd..f3a75ef1e 100644 --- a/dvc/ignore.py +++ b/dvc/ignore.py @@ -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) diff --git a/dvc/pathspec_math.py b/dvc/pathspec_math.py index 88fbf9cda..5d73c6125 100644 --- a/dvc/pathspec_math.py +++ b/dvc/pathspec_math.py @@ -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): diff --git a/dvc/repo/index.py b/dvc/repo/index.py index 2f042727e..f28db66b6 100644 --- a/dvc/repo/index.py +++ b/dvc/repo/index.py @@ -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) diff --git a/tests/unit/output/test_output.py b/tests/unit/output/test_output.py index bf86b88b2..172fc787c 100644 --- a/tests/unit/output/test_output.py +++ b/tests/unit/output/test_output.py @@ -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"