diff --git a/src/kaggle/api/kaggle_api_extended.py b/src/kaggle/api/kaggle_api_extended.py index e52aace7..4918c3b5 100644 --- a/src/kaggle/api/kaggle_api_extended.py +++ b/src/kaggle/api/kaggle_api_extended.py @@ -416,30 +416,105 @@ def __str__(self): return self.value +DEFAULT_IGNORE_PATTERNS = [ + ".git/", + "*/.git/", + ".cache/", + ".huggingface/", +] + + +def should_ignore(rel_path: str, is_dir: bool, patterns: List[str]) -> bool: + """Helper to check if a path should be ignored based on patterns.""" + import fnmatch + + rel_path = rel_path.replace(os.path.sep, "/") + match_path = rel_path + "/" if is_dir and not rel_path.endswith("/") else rel_path + + for pattern in patterns: + pattern = pattern.replace(os.path.sep, "/") + if pattern.endswith("/"): + if not is_dir: + continue + if fnmatch.fnmatch(match_path, pattern): + return True + return False + + class DirectoryArchive(object): - """ - Context manager for handling directory archives. + """Context manager for handling directory archives with filtering. - This class provides a context manager for working with directory archives in various formats. - It manages the lifecycle of the archive, including opening and closing resources as needed. + This class provides a context manager for working with directory archives in + various formats. It manages the lifecycle of the archive, including opening + and closing resources as needed. It supports filtering files based on + ignore patterns. """ - def __init__(self, fullpath, fmt): + def __init__(self, fullpath, fmt, ignore_patterns=None): self._fullpath = fullpath self._format = fmt + self._ignore_patterns = ignore_patterns or [] self.name = None self.path = None def __enter__(self): self._temp_dir = tempfile.mkdtemp() _, dir_name = os.path.split(self._fullpath) - self.path = shutil.make_archive(os.path.join(self._temp_dir, dir_name), self._format, self._fullpath) + archive_base_path = os.path.join(self._temp_dir, dir_name) + + if self._ignore_patterns: + self.path = self._create_archive_with_filters(archive_base_path) + else: + self.path = shutil.make_archive(archive_base_path, self._format, self._fullpath) + _, self.name = os.path.split(self.path) return self def __exit__(self, *args): shutil.rmtree(self._temp_dir) + def _create_archive_with_filters(self, base_name): + if self._format == "zip": + archive_path = base_name + ".zip" + with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_DEFLATED) as zipf: + for root, dirs, files in os.walk(self._fullpath): + dirs[:] = [ + d + for d in dirs + if not should_ignore( + os.path.relpath(os.path.join(root, d), self._fullpath), + True, + self._ignore_patterns, + ) + ] + for file in files: + file_path = os.path.join(root, file) + rel_path = os.path.relpath(file_path, self._fullpath) + if not should_ignore(rel_path, False, self._ignore_patterns): + zipf.write(file_path, rel_path) + return archive_path + elif self._format == "tar": + archive_path = base_name + ".tar" + with tarfile.open(archive_path, "w") as tarf: + for root, dirs, files in os.walk(self._fullpath): + dirs[:] = [ + d + for d in dirs + if not should_ignore( + os.path.relpath(os.path.join(root, d), self._fullpath), + True, + self._ignore_patterns, + ) + ] + for file in files: + file_path = os.path.join(root, file) + rel_path = os.path.relpath(file_path, self._fullpath) + if not should_ignore(rel_path, False, self._ignore_patterns): + tarf.add(file_path, rel_path) + return archive_path + else: + raise ValueError(f"Unsupported archive format: {self._format}") + class ResumableUploadContext(object): """ diff --git a/tests/unit/test_ignore_patterns.py b/tests/unit/test_ignore_patterns.py new file mode 100644 index 00000000..41f1ed68 --- /dev/null +++ b/tests/unit/test_ignore_patterns.py @@ -0,0 +1,152 @@ +import os +import shutil +import tempfile +import unittest +import zipfile +import tarfile + +from kaggle.api.kaggle_api_extended import should_ignore, DirectoryArchive, DEFAULT_IGNORE_PATTERNS + + +class TestIgnorePatterns(unittest.TestCase): + + def setUp(self): + self.temp_dir = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.temp_dir) + + def test_should_ignore_default_patterns(self): + # .git/ at root should be ignored + self.assertTrue(should_ignore(".git", True, DEFAULT_IGNORE_PATTERNS)) + self.assertTrue(should_ignore(".git/", True, DEFAULT_IGNORE_PATTERNS)) + + # .git/ in sub dir should be ignored (due to */.git/) + self.assertTrue(should_ignore("sub/.git", True, DEFAULT_IGNORE_PATTERNS)) + self.assertTrue(should_ignore("sub/.git/", True, DEFAULT_IGNORE_PATTERNS)) + self.assertTrue(should_ignore("sub/dir/.git", True, DEFAULT_IGNORE_PATTERNS)) + + # .git file (not dir) should NOT be ignored + self.assertFalse(should_ignore(".git", False, DEFAULT_IGNORE_PATTERNS)) + self.assertFalse(should_ignore("sub/.git", False, DEFAULT_IGNORE_PATTERNS)) + + # .cache/ at root should be ignored + self.assertTrue(should_ignore(".cache", True, DEFAULT_IGNORE_PATTERNS)) + + # .cache/ in sub dir should NOT be ignored (no */.cache/ in defaults) + self.assertFalse(should_ignore("sub/.cache", True, DEFAULT_IGNORE_PATTERNS)) + + # .huggingface/ at root should be ignored + self.assertTrue(should_ignore(".huggingface", True, DEFAULT_IGNORE_PATTERNS)) + + # .huggingface/ in sub dir should NOT be ignored + self.assertFalse(should_ignore("sub/.huggingface", True, DEFAULT_IGNORE_PATTERNS)) + + def test_should_ignore_custom_patterns(self): + patterns = ["*.tmp", "ignore_dir/", "*/nested_ignore_dir/", "exact_file.txt"] + + # Wildcard file pattern + self.assertTrue(should_ignore("foo.tmp", False, patterns)) + self.assertTrue(should_ignore("sub/foo.tmp", False, patterns)) + self.assertFalse(should_ignore("foo.tmp.bak", False, patterns)) + + # Directory pattern (root only) + self.assertTrue(should_ignore("ignore_dir", True, patterns)) + self.assertFalse(should_ignore("sub/ignore_dir", True, patterns)) + # Directory pattern (root only) should not match file + self.assertFalse(should_ignore("ignore_dir", False, patterns)) + + # Directory pattern (nested) + self.assertTrue(should_ignore("sub/nested_ignore_dir", True, patterns)) + self.assertTrue(should_ignore("sub/dir/nested_ignore_dir", True, patterns)) + self.assertFalse(should_ignore("nested_ignore_dir", True, patterns)) + + # Exact file pattern + self.assertTrue(should_ignore("exact_file.txt", False, patterns)) + # fnmatch "exact_file.txt" will not match "sub/exact_file.txt" if we don't have wildcard + # Wait, fnmatch("sub/exact_file.txt", "exact_file.txt") is False. + self.assertFalse(should_ignore("sub/exact_file.txt", False, patterns)) + + def test_directory_archive_zip_filtering(self): + # Create dummy directory structure + src_dir = os.path.join(self.temp_dir, "src") + os.makedirs(src_dir) + os.makedirs(os.path.join(src_dir, ".git")) + os.makedirs(os.path.join(src_dir, "sub")) + os.makedirs(os.path.join(src_dir, "sub", ".git")) + os.makedirs(os.path.join(src_dir, "ignore_dir")) + + # Create files + self._create_file(os.path.join(src_dir, "keep.txt")) + self._create_file(os.path.join(src_dir, ".git", "config")) + self._create_file(os.path.join(src_dir, "sub", "keep2.txt")) + self._create_file(os.path.join(src_dir, "sub", ".git", "config")) + self._create_file(os.path.join(src_dir, "sub", "temp.tmp")) + self._create_file(os.path.join(src_dir, "ignore_dir", "file.txt")) + + patterns = DEFAULT_IGNORE_PATTERNS + ["*.tmp", "ignore_dir/"] + + # Archive + with DirectoryArchive(src_dir, "zip", ignore_patterns=patterns) as ar: + archive_path = ar.path + self.assertTrue(os.path.exists(archive_path)) + + # Extract and verify + extract_dir = os.path.join(self.temp_dir, "extract") + os.makedirs(extract_dir) + with zipfile.ZipFile(archive_path, "r") as zip_ref: + zip_ref.extractall(extract_dir) + + # Check files + self.assertTrue(os.path.exists(os.path.join(extract_dir, "keep.txt"))) + self.assertTrue(os.path.exists(os.path.join(extract_dir, "sub", "keep2.txt"))) + + # Checked ignored + self.assertFalse(os.path.exists(os.path.join(extract_dir, ".git"))) + self.assertFalse(os.path.exists(os.path.join(extract_dir, "sub", ".git"))) + self.assertFalse(os.path.exists(os.path.join(extract_dir, "sub", "temp.tmp"))) + self.assertFalse(os.path.exists(os.path.join(extract_dir, "ignore_dir"))) + + def test_directory_archive_tar_filtering(self): + # Create dummy directory structure + src_dir = os.path.join(self.temp_dir, "src") + os.makedirs(src_dir) + os.makedirs(os.path.join(src_dir, ".git")) + os.makedirs(os.path.join(src_dir, "sub")) + os.makedirs(os.path.join(src_dir, "sub", ".git")) + os.makedirs(os.path.join(src_dir, "ignore_dir")) + + # Create files + self._create_file(os.path.join(src_dir, "keep.txt")) + self._create_file(os.path.join(src_dir, ".git", "config")) + self._create_file(os.path.join(src_dir, "sub", "keep2.txt")) + self._create_file(os.path.join(src_dir, "sub", ".git", "config")) + self._create_file(os.path.join(src_dir, "sub", "temp.tmp")) + self._create_file(os.path.join(src_dir, "ignore_dir", "file.txt")) + + patterns = DEFAULT_IGNORE_PATTERNS + ["*.tmp", "ignore_dir/"] + + # Archive + with DirectoryArchive(src_dir, "tar", ignore_patterns=patterns) as ar: + archive_path = ar.path + self.assertTrue(os.path.exists(archive_path)) + + # Extract and verify + extract_dir = os.path.join(self.temp_dir, "extract") + os.makedirs(extract_dir) + with tarfile.open(archive_path, "r") as tar_ref: + tar_ref.extractall(extract_dir) + + # Check files + self.assertTrue(os.path.exists(os.path.join(extract_dir, "keep.txt"))) + self.assertTrue(os.path.exists(os.path.join(extract_dir, "sub", "keep2.txt"))) + + # Checked ignored + self.assertFalse(os.path.exists(os.path.join(extract_dir, ".git"))) + self.assertFalse(os.path.exists(os.path.join(extract_dir, "sub", ".git"))) + self.assertFalse(os.path.exists(os.path.join(extract_dir, "sub", "temp.tmp"))) + self.assertFalse(os.path.exists(os.path.join(extract_dir, "ignore_dir"))) + + def _create_file(self, path): + with open(path, "w") as f: + f.write("dummy content")