refactor: consolidate pattern parsing and ignore file handling in CLI and query parser

Author: AbhiRam162105

src/gitingest/cli.py

@@ -6,57 +6,11 @@
 
 import click
 
-from gitingest.config import MAX_FILE_SIZE, OUTPUT_FILE_PATH
+from gitingest.config import MAX_FILE_SIZE
+from gitingest.query_parser import _parse_patterns, parse_ignore_file
 from gitingest.repository_ingest import ingest
 
 
-def parse_ignore_file(ignore_file_path: Path) -> set[str]:
-    """
-    Parse the .gitingestignore file and return a set of patterns to ignore.
-
-    Parameters
-    ----------
-    ignore_file_path : Path
-        Path to the .gitingestignore file
-
-    Returns
-    -------
-    set[str]
-        Set of patterns to ignore
-    """
-    if not ignore_file_path.exists():
-        return set()
-
-    with open(ignore_file_path, encoding="utf-8") as f:
-        # Read lines, strip whitespace, and filter out empty lines and comments
-        patterns = {line.strip() for line in f if line.strip() and not line.startswith("#")}
-
-    return patterns
-
-
-def parse_patterns(patterns: tuple[str, ...]) -> set[str]:
-    """
-    Parse patterns from command line arguments.
-    Handles both space-separated patterns in a single string
-    and multiple -e/-i arguments.
-
-    Parameters
-    ----------
-    patterns : tuple[str, ...]
-        Tuple of patterns from command line
-
-    Returns
-    -------
-    set[str]
-        Set of parsed patterns
-    """
-    result = set()
-    for pattern_str in patterns:
-        # Split on spaces and add each pattern
-        result.update(p.strip() for p in pattern_str.split() if p.strip())
-    return result
-
-
 @click.command()
 @click.argument("source", type=str, default=".")
 @click.option("--output", "-o", default=None, help="Output file path (default: <repo_name>.txt in current directory)")
@@ -139,8 +93,8 @@ async def async_main(
             output = f"{repo_name}.txt"
 
         # Parse command line patterns
-        exclude_patterns = parse_patterns(exclude_pattern)
-        include_patterns = parse_patterns(include_pattern)
+        exclude_patterns = _parse_patterns(exclude_pattern)
+        include_patterns = _parse_patterns(include_pattern)
 
         # Read and add patterns from ignore file
         ignore_file_path = Path(source) / ignore_file
@@ -170,4 +124,4 @@ async def async_main(
 
 
 if __name__ == "__main__":
-    main()
+    main()

src/gitingest/ignore_patterns.py

@@ -153,4 +153,6 @@
     "*.tfstate*",
     ## Dependencies in various languages
     "vendor/",
+    ## gitingestignore file
+    ".gitingestignore",
 }

src/gitingest/query_parser.py

@@ -12,7 +12,7 @@
 from gitingest.config import MAX_FILE_SIZE, TMP_BASE_PATH
 from gitingest.exceptions import InvalidPatternError
 from gitingest.ignore_patterns import DEFAULT_IGNORE_PATTERNS
-from gitingest.repository_clone import _check_repo_exists, fetch_remote_branch_list
+from gitingest.repository_clone import CloneConfig, _check_repo_exists, clone_repo, fetch_remote_branch_list
 
 HEX_DIGITS: set[str] = set(string.hexdigits)
 
@@ -48,6 +48,30 @@ class ParsedQuery:  # pylint: disable=too-many-instance-attributes
     pattern_type: str | None = None
 
 
+def parse_ignore_file(ignore_file_path: Path) -> set[str]:
+    """
+    Parse the .gitingestignore file and return a set of patterns to ignore.
+
+    Parameters
+    ----------
+    ignore_file_path : Path
+        Path to the .gitingestignore file
+
+    Returns
+    -------
+    set[str]
+        Set of patterns to ignore
+    """
+    if not ignore_file_path.exists():
+        return set()
+
+    with open(ignore_file_path, encoding="utf-8") as f:
+        # Read lines, strip whitespace, and filter out empty lines and comments
+        patterns = {line.strip() for line in f if line.strip() and not line.startswith("#")}
+
+    return patterns
+
+
 async def parse_query(
     source: str,
     max_file_size: int,
@@ -89,6 +113,24 @@ async def parse_query(
         # Local path scenario
         parsed_query = _parse_path(source)
 
+    # Clone the repository if it's a URL
+    if parsed_query.url:
+        clone_config = CloneConfig(
+            url=parsed_query.url,
+            local_path=str(parsed_query.local_path),
+            commit=parsed_query.commit,
+            branch=parsed_query.branch,
+        )
+        await clone_repo(clone_config)
+
+        # Look for .gitingestignore file in the cloned repository
+        ignore_file_path = Path(parsed_query.local_path) / ".gitingestignore"
+        additional_ignore_patterns = parse_ignore_file(ignore_file_path)
+        if ignore_patterns:
+            ignore_patterns.update(additional_ignore_patterns)
+        else:
+            ignore_patterns = additional_ignore_patterns
+
     # Combine default ignore patterns + custom patterns
     ignore_patterns_set = DEFAULT_IGNORE_PATTERNS.copy()
     if ignore_patterns:
@@ -283,17 +325,18 @@ def _normalize_pattern(pattern: str) -> str:
     return pattern
 
 
-def _parse_patterns(pattern: set[str] | str) -> set[str]:
+def _parse_patterns(patterns: tuple[str, ...] | set[str] | str) -> set[str]:
     """
     Parse and validate file/directory patterns for inclusion or exclusion.
 
-    Takes either a single pattern string or set of pattern strings and processes them into a normalized list.
-    Patterns are split on commas and spaces, validated for allowed characters, and normalized.
+    Takes either a single pattern string, a tuple of pattern strings, or a set of pattern strings
+    and processes them into a normalized list. Patterns are split on commas and spaces, validated
+    for allowed characters, and normalized.
 
     Parameters
     ----------
-    pattern : set[str] | str
-        Pattern(s) to parse - either a single string or set of strings
+    patterns : tuple[str, ...] | set[str] | str
+        Pattern(s) to parse - either a single string, a tuple of strings, or a set of strings
 
     Returns
     -------
@@ -307,7 +350,11 @@ def _parse_patterns(pattern: set[str] | str) -> set[str]:
         dash (-), underscore (_), dot (.), forward slash (/), plus (+), and
         asterisk (*) are allowed.
     """
-    patterns = pattern if isinstance(pattern, set) else {pattern}
+    # Convert patterns to a set if it's not already a set
+    if isinstance(patterns, tuple):
+        patterns = set(patterns)
+    elif isinstance(patterns, str):
+        patterns = {patterns}
 
     parsed_patterns: set[str] = set()
     for p in patterns:

src/gitingest/repository_clone.py

@@ -1,7 +1,11 @@
-""" This module contains functions for cloning a Git repository to a local path. """
+"""
+Module for cloning repositories in the gitingest package.
+"""
 
 import asyncio
+import shutil
 from dataclasses import dataclass
+from pathlib import Path
 
 from gitingest.utils import async_timeout
 
@@ -78,6 +82,11 @@ async def clone_repo(config: CloneConfig) -> tuple[bytes, bytes]:
     if not await _check_repo_exists(url):
         raise ValueError("Repository not found, make sure it is public")
 
+    # Remove the directory if it exists and is not empty
+    local_path_obj = Path(local_path)
+    if local_path_obj.exists() and any(local_path_obj.iterdir()):
+        shutil.rmtree(local_path_obj)
+
     if commit:
         # Scenario 1: Clone and checkout a specific commit
         # Clone the repository without depth to ensure full history for checkout
@@ -89,7 +98,6 @@ async def clone_repo(config: CloneConfig) -> tuple[bytes, bytes]:
         return await _run_git_command(*checkout_cmd)
 
     if branch and branch.lower() not in ("main", "master"):
-
         # Scenario 2: Clone a specific branch with shallow depth
         clone_cmd = ["git", "clone", "--depth=1", "--single-branch", "--branch", branch, url, local_path]
         return await _run_git_command(*clone_cmd)

src/gitingest/repository_ingest.py

@@ -1,6 +1,7 @@
-""" Main entry point for ingesting a source and processing its contents. """
+"""
+Module for ingesting repositories in the gitingest package.
+"""
 
-import asyncio
 import inspect
 import shutil
 
@@ -71,7 +72,7 @@ async def ingest(
             clone_result = clone_repo(clone_config)
 
             if inspect.iscoroutine(clone_result):
-                asyncio.run(clone_result)
+                await clone_result
             else:
                 raise TypeError("clone_repo did not return a coroutine as expected.")
 
@@ -85,5 +86,4 @@ async def ingest(
     finally:
         # Clean up the temporary directory if it was created
         if parsed_query.url:
-            # Clean up the temporary directory
             shutil.rmtree(TMP_BASE_PATH, ignore_errors=True)