resolve commit

Author: filipchristiansen

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, url=url, 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)
 
-    # 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/query_parser.py

@@ -55,6 +55,9 @@ async def parse_query(
         A dataclass object containing the parsed details of the repository or file path.
 
     """
+    if source.endswith(".git"):
+        source = source[:-4]
+
     # Determine the parsing method based on the source type
     if from_web or 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

src/gitingest/utils/git_utils.py

@@ -6,7 +6,8 @@
 import base64
 import re
 import sys
-from typing import Final
+from pathlib import Path
+from typing import TYPE_CHECKING, Final, Iterable, Literal
 from urllib.parse import urlparse
 
 import httpx
@@ -16,6 +17,9 @@
 from gitingest.utils.exceptions import InvalidGitHubTokenError
 from server.server_utils import Colors
 
+if TYPE_CHECKING:
+    from gitingest.schemas import CloneConfig
+
 # GitHub Personal-Access tokens (classic + fine-grained).
 #   - ghp_ / gho_ / ghu_ / ghs_ / ghr_  → 36 alphanumerics
 #   - github_pat_                       → 22 alphanumerics + "_" + 59 alphanumerics
@@ -321,3 +325,146 @@ def validate_github_token(token: str) -> None:
     """
     if not re.fullmatch(_GITHUB_PAT_PATTERN, token):
         raise InvalidGitHubTokenError
+
+
+async def checkout_partial_clone(config: CloneConfig, token: str | None) -> None:
+    """Configure sparse-checkout for a partially cloned repository.
+
+    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)
+
+
+async def resolve_commit(config: CloneConfig, url: str, token: str | None) -> str:
+    """Resolve the commit to use for the clone.
+
+    Parameters
+    ----------
+    config : CloneConfig
+        The configuration for cloning the repository.
+    url : str
+        The URL of the remote repository.
+    token : str | None
+        GitHub personal access token (PAT) for accessing private repositories.
+
+    Returns
+    -------
+    str
+
+    """
+    if config.commit:
+        commit = config.commit
+    elif config.tag:
+        commit = await _resolve_ref_to_sha(url, ref=config.tag, kind="tag", token=token)
+    elif config.branch:
+        commit = await _resolve_ref_to_sha(url, ref=config.branch, kind="branch", token=token)
+    else:
+        commit = await _resolve_ref_to_sha(url, ref="HEAD", kind="branch", token=token)
+    return commit
+
+
+async def _resolve_ref_to_sha(
+    url: str,
+    ref: str,
+    kind: Literal["branch", "tag"],
+    *,
+    token: str | None = None,
+) -> str:
+    """Return the commit SHA that <kind>/<ref> points to in <url>.
+
+    * Branch → first line from ``git ls-remote``.
+    * Tag    → if annotated, prefer the peeled ``^{}`` line (commit).
+
+    Parameters
+    ----------
+    url : str
+        The URL of the remote repository.
+    ref : str
+        The reference to resolve to a commit SHA.
+    kind : Literal["branch", "tag"]
+        The kind of reference to resolve to a commit SHA.
+    token : str | None
+        GitHub personal access token (PAT) for accessing private repositories.
+
+    Returns
+    -------
+    str
+        The commit SHA.
+
+    Raises
+    ------
+    ValueError
+        If the ref does not exist in the remote repository.
+
+    """
+    await ensure_git_installed()
+
+    # Build: git [-c http.<host>/.extraheader=Auth...] ls-remote <url> <pattern>
+    cmd: list[str] = ["git"]
+    if token and is_github_host(url):
+        cmd += ["-c", create_git_auth_header(token, url=url)]
+
+    if ref == "HEAD":
+        pattern = "HEAD"
+    elif kind == "branch":
+        pattern = f"refs/heads/{ref}"
+    else:  # tag
+        pattern = f"refs/tags/{ref}*"
+
+    cmd += ["ls-remote", url, pattern]
+    stdout, _ = await run_command(*cmd)
+
+    lines = stdout.decode().splitlines()
+
+    sha = _pick_commit_sha(lines)
+    if not sha:
+        msg = f"{kind} {ref!r} not found in {url}"
+        raise ValueError(msg)
+
+    return sha
+
+
+def _pick_commit_sha(lines: Iterable[str]) -> str | None:
+    """Return a commit SHA from ``git ls-remote`` output.
+
+    • Annotated tag            →  prefer the peeled line (<sha> refs/tags/x^{})
+    • Branch / lightweight tag →  first non-peeled line
+
+
+    Parameters
+    ----------
+    lines : Iterable[str]
+        The lines of a ``git ls-remote`` output.
+
+    Returns
+    -------
+    str | None
+        The commit SHA, or ``None`` if no commit SHA is found.
+
+    """
+    first_non_peeled: str | None = None
+
+    for ln in lines:
+        if not ln.strip():
+            continue
+
+        sha, ref = ln.split(maxsplit=1)
+
+        if ref.endswith("^{}"):  # peeled commit of annotated tag
+            return sha  # ← best match, done
+
+        if first_non_peeled is None:  # remember the first ordinary line
+            first_non_peeled = sha
+
+    return first_non_peeled  # branch or lightweight tag (or None)

src/gitingest/utils/os_utils.py

@@ -3,7 +3,7 @@
 from pathlib import Path
 
 
-async def ensure_directory(path: Path) -> None:
+async def ensure_directory_exists_or_create(path: Path) -> None:
     """Ensure the directory exists, creating it if necessary.
 
     Parameters

tests/test_clone.py

@@ -33,7 +33,7 @@ async def test_clone_with_commit(repo_exists_true: AsyncMock, run_command_mock:
     When ``clone_repo`` is called,
     Then the repository should be cloned and checked out at that commit.
     """
-    expected_call_count = 2
+    expected_call_count = 3  # clone + fetch + checkout
     clone_config = CloneConfig(
         url=DEMO_URL,
         local_path=LOCAL_REPO_PATH,