resolve commit

Author: filipchristiansen

src/gitingest/__init__.py

@@ -1,8 +1,5 @@
 """Gitingest: A package for ingesting data from Git repositories."""
 
-from gitingest.clone import clone_repo
 from gitingest.entrypoint import ingest, ingest_async
-from gitingest.ingestion import ingest_query
-from gitingest.query_parser import parse_query
 
-__all__ = ["clone_repo", "ingest", "ingest_async", "ingest_query", "parse_query"]
+__all__ = ["ingest", "ingest_async"]

src/gitingest/clone.py

@@ -8,13 +8,15 @@
 from gitingest.config import DEFAULT_TIMEOUT
 from gitingest.utils.git_utils import (
     check_repo_exists,
+    checkout_partial_clone,
     create_git_auth_header,
     create_git_command,
     ensure_git_installed,
     is_github_host,
+    resolve_commit,
     run_command,
 )
-from gitingest.utils.os_utils import ensure_directory
+from gitingest.utils.os_utils import ensure_directory_exists_or_create
 from gitingest.utils.timeout_wrapper import async_timeout
 
 if TYPE_CHECKING:
@@ -45,71 +47,42 @@ async def clone_repo(config: CloneConfig, *, token: str | None = None) -> None:
     # Extract and validate query parameters
     url: str = config.url
     local_path: str = config.local_path
-    commit: str | None = config.commit
-    branch: str | None = config.branch
-    tag: str | None = config.tag
     partial_clone: bool = config.subpath != "/"
 
-    # Create parent directory if it doesn't exist
-    await ensure_directory(Path(local_path).parent)
+    await ensure_git_installed()
+    await ensure_directory_exists_or_create(Path(local_path).parent)
 
-    # Check if the repository exists
     if not await check_repo_exists(url, token=token):
         msg = "Repository not found. Make sure it is public or that you have provided a valid token."
         raise ValueError(msg)
 
+    commit = await resolve_commit(config, token=token)
+
     clone_cmd = ["git"]
     if token and is_github_host(url):
         clone_cmd += ["-c", create_git_auth_header(token, url=url)]
 
-    clone_cmd += ["clone", "--single-branch"]
-
-    if config.include_submodules:
-        clone_cmd += ["--recurse-submodules"]
-
+    clone_cmd += ["clone", "--single-branch", "--no-checkout", "--depth=1"]
     if partial_clone:
         clone_cmd += ["--filter=blob:none", "--sparse"]
 
-    # Shallow clone unless a specific commit is requested
-    if not commit:
-        clone_cmd += ["--depth=1"]
-
-        # Prefer tag over branch when both are provided
-        if tag:
-            clone_cmd += ["--branch", tag]
-        elif branch and branch.lower() not in ("main", "master"):
-            clone_cmd += ["--branch", branch]
-
     clone_cmd += [url, local_path]
 
     # Clone the repository
-    await ensure_git_installed()
     await run_command(*clone_cmd)
 
     # Checkout the subpath if it is a partial clone
     if partial_clone:
-        await _checkout_partial_clone(config, token)
+        await checkout_partial_clone(config, token=token)
 
-    # Checkout the commit if it is provided
-    if commit:
-        checkout_cmd = create_git_command(["git"], local_path, url, token)
-        await run_command(*checkout_cmd, "checkout", commit)
+    git = create_git_command(["git"], local_path, url, token)
 
+    # Ensure the commit is locally available
+    await run_command(*git, "fetch", "--depth=1", "origin", commit)
 
-async def _checkout_partial_clone(config: CloneConfig, token: str | None) -> None:
-    """Configure sparse-checkout for a partially cloned repository.
+    # Write the work-tree at that commit
+    await run_command(*git, "checkout", commit)
 
-    Parameters
-    ----------
-    config : CloneConfig
-        The configuration for cloning the repository, including subpath and blob flag.
-    token : str | None
-        GitHub personal access token (PAT) for accessing private repositories.
-
-    """
-    subpath = config.subpath.lstrip("/")
-    if config.blob:
-        # Remove the file name from the subpath when ingesting from a file url (e.g. blob/branch/path/file.txt)
-        subpath = str(Path(subpath).parent.as_posix())
-    checkout_cmd = create_git_command(["git"], config.local_path, config.url, token)
-    await run_command(*checkout_cmd, "sparse-checkout", "set", subpath)
+    # Update submodules
+    if config.include_submodules:
+        await run_command(*git, "submodule", "update", "--init", "--recursive", "--depth=1")

src/gitingest/entrypoint.py

@@ -8,14 +8,21 @@
 import warnings
 from contextlib import asynccontextmanager
 from pathlib import Path
-from typing import AsyncGenerator
+from typing import TYPE_CHECKING, AsyncGenerator
+from urllib.parse import urlparse
 
 from gitingest.clone import clone_repo
 from gitingest.config import MAX_FILE_SIZE
 from gitingest.ingestion import ingest_query
-from gitingest.query_parser import IngestionQuery, parse_query
+from gitingest.query_parser import parse_local_dir_path, parse_remote_repo
 from gitingest.utils.auth import resolve_token
+from gitingest.utils.compat_func import removesuffix
 from gitingest.utils.ignore_patterns import load_ignore_patterns
+from gitingest.utils.pattern_utils import process_patterns
+from gitingest.utils.query_parser_utils import KNOWN_GIT_HOSTS
+
+if TYPE_CHECKING:
+    from gitingest.schemas import IngestionQuery
 
 
 async def ingest_async(
@@ -74,23 +81,28 @@ async def ingest_async(
     """
     token = resolve_token(token)
 
-    query: IngestionQuery = await parse_query(
-        source=source,
-        max_file_size=max_file_size,
-        from_web=False,
+    source = removesuffix(source.strip(), ".git")
+
+    # Determine the parsing method based on the source type
+    if urlparse(source).scheme in ("https", "http") or any(h in source for h in KNOWN_GIT_HOSTS):
+        # We either have a full URL or a domain-less slug
+        query = await parse_remote_repo(source, token=token)
+        query.include_submodules = include_submodules
+        _override_branch_and_tag(query, branch=branch, tag=tag)
+
+    else:
+        # Local path scenario
+        query = parse_local_dir_path(source)
+
+    query.max_file_size = max_file_size
+    query.ignore_patterns, query.include_patterns = process_patterns(
+        exclude_patterns=exclude_patterns,
         include_patterns=include_patterns,
-        ignore_patterns=exclude_patterns,
-        token=token,
     )
 
     if not include_gitignored:
         _apply_gitignores(query)
 
-    if query.url:
-        _override_branch_and_tag(query, branch=branch, tag=tag)
-
-    query.include_submodules = include_submodules
-
     async with _clone_repo_if_remote(query, token=token):
         summary, tree, content = ingest_query(query)
         await _write_output(tree, content=content, target=output)

src/gitingest/ingestion.py

@@ -11,7 +11,7 @@
 from gitingest.utils.ingestion_utils import _should_exclude, _should_include
 
 if TYPE_CHECKING:
-    from gitingest.query_parser import IngestionQuery
+    from gitingest.schemas import IngestionQuery
 
 
 def ingest_query(query: IngestionQuery) -> tuple[str, str, str]:

src/gitingest/output_formatter.py

@@ -10,7 +10,7 @@
 from gitingest.utils.compat_func import readlink
 
 if TYPE_CHECKING:
-    from gitingest.query_parser import IngestionQuery
+    from gitingest.schemas import IngestionQuery
 
 _TOKEN_THRESHOLDS: list[tuple[int, str]] = [
     (1_000_000, "M"),
@@ -84,6 +84,8 @@ def _create_summary_prefix(query: IngestionQuery, *, single_file: bool = False)
 
     if query.commit:
         parts.append(f"Commit: {query.commit}")
+    elif query.tag:
+        parts.append(f"Tag: {query.tag}")
     elif query.branch and query.branch not in ("main", "master"):
         parts.append(f"Branch: {query.branch}")