repo_on_fire.git

Wrapper around git.

This module provides a thin wrapper around the platform's default git command. In addition, it brings the necessary tools to setup a wrapper such that we can inject our git wrapper into other processes, like repo.

Git

Utilities to work with git.

Source code in repo_on_fire/git.py

class Git:
    """Utilities to work with git."""

    def __init__(self, configuration: Configuration):
        """Create a new git wrapper object.

        Args:
            configuration: The configuration object to be used by the object.
        """
        self._configuration = configuration

    @staticmethod
    def make_git_wrapper(directory_path: Path):
        """Create a wrapper which is callable as ``git`` but invoke ``git-on-fire``.

        This function can be used to create a platform native wrapper script,
        which can be invoked using ``git`` but instead, it will call
        ``git-on-fire``.

        Args:
            directory_path: The path to the directory where to create the
                wrapper.
        """
        if platform.system() == "Windows":
            template = _WINDOWS_WRAPPER_SCRIPT
            suffix = ".bat"
            make_executable = False
        else:
            template = _UNIX_WRAPPER_SCRIPT
            suffix = ""
            make_executable = True
        native_git_executable = Git.find_git()
        script_content = template.format(
            native_git_executable=native_git_executable,
            native_git_executable_sh=shlex.quote(native_git_executable),
            python_executable=sys.executable,
            python_executable_sh=shlex.quote(sys.executable),
        )
        directory_path.mkdir(parents=True, exist_ok=True)
        file_path = directory_path / ("git" + suffix)
        if file_path.exists():
            # If the file exists, check if we need to update the content:
            if file_path.read_text(encoding="utf-8") != script_content:
                file_path.write_text(script_content, encoding="utf-8")
        else:
            # Otherwise, always write:
            file_path.write_text(script_content, encoding="utf-8")
        if make_executable:
            os.chmod(file_path, 0o755)

    @staticmethod
    def find_git():
        """Find the git command.

        This method can be used to find the git command. This method assumes
        that the current process has been called by a previously created
        git wrapper (using the [make_git_wrapper][repo_on_fire.git.Git.make_git_wrapper] method).
        In this case, special care is taken to avoid that the wrapper recursively finds
        itself, which would lead to an infinite loop (and exhaustion of
        resources of the host).

        Raises:
            ExecutableNotFound: The git command could not be found.
        """
        # Compile a list of search paths. There are two possibilities:
        GIT_ON_FIRE_WRAPPER_DIR = "GIT_ON_FIRE_WRAPPER_DIR"
        if GIT_ON_FIRE_WRAPPER_DIR in os.environ:
            # We are called from a git wrapper. In this case, the list of
            # paths to look for git is compiled by searching PATH for the
            # directory where the wrapper is stored in and - if found - we only
            # include the paths after the entry.
            path_variable = os.environ.get("PATH", os.defpath)
            paths = path_variable.split(os.pathsep)
            wrapper_path = Path(os.environ[GIT_ON_FIRE_WRAPPER_DIR]).resolve()
            search_paths = []
            for i, path in enumerate(paths):
                path_path = Path(path).resolve()
                if wrapper_path == path_path:
                    search_paths = paths[i + 1 :]
                    break
            if len(search_paths) == 0:
                # The wrapper path was not in PATH. This can happen if the
                # wrapper is called without adding its path to PATH. Fall back
                # to using the default path.
                search_paths = paths
            joined_path = os.pathsep.join(search_paths)
            git = shutil.which("git", path=joined_path)
            if git is None:
                raise ExecutableNotFound(f"Did not find git in '{joined_path}'")
            return git
        else:
            # We didn't end here due to a call to the git wrapper. Hence,
            # simply return git from wherever we can find it:
            git = shutil.which("git")
            if git is None:
                raise ExecutableNotFound("Unable to find git executable")
            return git

    @staticmethod
    def find_repo_root(cwd: Optional[Union[str, Path]] = None):
        """Try to find the rep root directory.

        This method looks for the ``.repo`` directory in ``cwd`` and recursively
        upwards.

        Args:
            cwd: The directory where to start searching. The method starts
                 looking for the ``.repo`` directory there and follows upwards
                 in the file system hierarchy. If None, search starts
                 in ``os.curdir``.

        Raises:
            RepoRootNotFound: If no ``.repo`` directory can be found.
        """
        if cwd is None:
            cwd = os.getcwd()
        path = Path(cwd)
        while path != path.parent:
            repo_dir = path / ".repo"
            if repo_dir.exists():
                return path
            path = path.parent
        raise RepoRootNotFound(f"Could not find .repo in '{cwd}")

    def clone(self, args: List[str], global_args: Optional[List[str]] = None) -> int:
        """Clone a repository using the local git cache.

        The first clone for a specific clone request is run in the cache. Later
        clones update that cache entry and copy the cached working tree to the
        originally requested destination.

        Args:
            args: Arguments passed after ``git clone``.
            global_args: Arguments passed before ``clone`` to the ``git``
                command itself.

        Returns:
            The native git-compatible exit code.
        """
        if global_args is None:
            global_args = []

        effective_cwd, filtered_global_args = self._extract_global_cwd(global_args)
        if self._is_clone_help_requested(args):
            return self._run_git([*filtered_global_args, "clone", *args], cwd=effective_cwd)

        try:
            repository, destination, destination_index = self._parse_clone_args(args)
        except ValueError:
            return self._run_git([*filtered_global_args, "clone", *args], cwd=effective_cwd)

        cache_entry = self.get_clone_cache_entry(args, destination_index)
        cache_args = self._replace_clone_destination(args, destination_index, cache_entry)
        destination_path = self._get_clone_destination_path(
            repository, destination, cwd=effective_cwd
        )

        with lock_path(cache_entry):
            if not cache_entry.exists():
                exit_code = self._run_git_for_cache(
                    [*filtered_global_args, "clone", *cache_args], cwd=effective_cwd
                )
                if exit_code != 0:
                    shutil.rmtree(cache_entry, ignore_errors=True)
                    return exit_code
            else:
                exit_code = self._update_clone_cache(cache_entry, filtered_global_args)
                if exit_code != 0:
                    return exit_code

            try:
                self._copy_clone_cache(cache_entry, destination_path)
            except OSError as error:
                print(f"git-on-fire: could not copy cached clone: {error}", file=sys.stderr)
                return 128

        return 0

    def fetch(self, args: List[str], global_args: Optional[List[str]] = None) -> int:
        """Fetch through the local git cache when a repository can be identified."""
        if global_args is None:
            global_args = []

        effective_cwd, filtered_global_args = self._extract_global_cwd(global_args)
        native_fetch_args = [*filtered_global_args, "fetch", *args]
        fetch_repository = self._parse_fetch_repository(args)
        if fetch_repository is None:
            return self._run_git(native_fetch_args, cwd=effective_cwd)

        repository, repository_index = fetch_repository
        if self._is_local_repository_path(repository):
            return self._run_git(native_fetch_args, cwd=effective_cwd)
        if not self._is_fetch_target_empty(effective_cwd):
            return self._run_git(native_fetch_args, cwd=effective_cwd)

        repository_url = self._resolve_fetch_repository_url(
            repository, filtered_global_args, effective_cwd
        )
        if repository_url is None:
            return self._run_git(native_fetch_args, cwd=effective_cwd)

        cache_entry = self.get_fetch_cache_entry(repository_url)

        with lock_path(cache_entry):
            if not self._is_git_repository(cache_entry):
                exit_code = self._create_fetch_cache_if_missing(repository_url, cache_entry)
            else:
                exit_code = self._run_git_for_cache(
                    [
                        *filtered_global_args,
                        "-C",
                        str(cache_entry),
                        "fetch",
                        "--all",
                        "--tags",
                        "--prune",
                    ]
                )

            if exit_code != 0:
                return exit_code

        cache_args = list(args)
        cache_args[repository_index] = str(cache_entry)
        return self._run_git([*filtered_global_args, "fetch", *cache_args], cwd=effective_cwd)

    def ls_remote(self, repository_url: str, ref: str) -> str:
        """Query a remote repository for a ref."""
        return self._run_git_output(["ls-remote", repository_url, ref])

    def get_clone_cache_entry(self, args: List[str], destination_index: Optional[int]) -> Path:
        """Get the cache entry path for a ``git clone`` request."""
        cache_key_args = [
            arg
            for index, arg in enumerate(args)
            if destination_index is None or index != destination_index
        ]
        cache_id = sha256("\0".join(cache_key_args).encode("utf-8")).hexdigest()
        return self._configuration.cache_path / "git" / cache_id[0:2] / cache_id

    def get_fetch_cache_entry(self, repository_url: str) -> Path:
        """Get the cache entry path for a fetch repository."""
        cache_id = sha256(f"fetch\0{repository_url}".encode()).hexdigest()
        return self._configuration.cache_path / "git" / cache_id[0:2] / cache_id

    @contextmanager
    def with_git_wrapper(self, cwd: Optional[Union[str, Path]] = None):
        """Run a command with a patched git command.

        This method implements a context manager which can be used to run
        commands with a modified ``PATH`` environment variable such that
        instead of the usual ``git``, a wrapper is called which provides
        addon functionality. The wrapper is created using
        [make_git_wrapper][repo_on_fire.git.Git.make_git_wrapper].

        Args:
            cwd: The path where to search for the repo workspace. If set to
                 None, the current working directory will be used instead.
        """
        try:
            repo_root = Git.find_repo_root(cwd)
            try:
                # Create a git wrapper in the repo folder
                wrapper_path = repo_root / ".repo" / "repo-on-fire" / "bin"
                Git.make_git_wrapper(wrapper_path)
                env = os.environ
                path = env.get("PATH", "")
                modified_path = os.pathsep.join([str(wrapper_path), path])
                os.environ["PATH"] = modified_path
                yield dict(os.environ)
            finally:
                os.environ["PATH"] = path
        except RepoRootNotFound:
            # If we are outside a workspace, create a wrapper in a temporary
            # folder:
            with TemporaryDirectory() as tmp_dir:
                try:
                    # Create a git wrapper in the temp dir
                    wrapper_path = Path(tmp_dir)
                    Git.make_git_wrapper(wrapper_path)
                    env = os.environ
                    path = env.get("PATH", "")
                    modified_path = os.pathsep.join([str(wrapper_path), path])
                    os.environ["PATH"] = modified_path
                    yield dict(os.environ)
                finally:
                    os.environ["PATH"] = path

    @staticmethod
    def _parse_clone_args(args: List[str]) -> Tuple[str, Optional[str], Optional[int]]:
        option_value_names = {
            "-b",
            "--branch",
            "-o",
            "--origin",
            "-u",
            "--upload-pack",
            "--template",
            "--reference",
            "--reference-if-able",
            "--separate-git-dir",
            "--depth",
            "--shallow-since",
            "--shallow-exclude",
            "-j",
            "--jobs",
            "--server-option",
            "-c",
            "--config",
        }
        positionals: List[Tuple[int, str]] = []
        i = 0
        end_of_options = False
        while i < len(args):
            arg = args[i]
            if not end_of_options and arg == "--":
                end_of_options = True
                i += 1
                continue
            if not end_of_options and arg.startswith("--"):
                option_name = arg.split("=", 1)[0]
                i += 2 if option_name in option_value_names and "=" not in arg else 1
                continue
            if not end_of_options and arg.startswith("-") and arg != "-":
                i += 2 if arg in option_value_names else 1
                continue
            positionals.append((i, arg))
            i += 1

        if len(positionals) == 0:
            raise ValueError("git clone requires a repository argument")
        if len(positionals) > _MAX_CLONE_POSITIONALS:
            raise ValueError("git clone received too many repository/destination arguments")

        repository = positionals[0][1]
        destination = positionals[1][1] if len(positionals) == _MAX_CLONE_POSITIONALS else None
        destination_index = (
            positionals[1][0] if len(positionals) == _MAX_CLONE_POSITIONALS else None
        )
        return repository, destination, destination_index

    @staticmethod
    def _is_clone_help_requested(args: List[str]) -> bool:
        return "-h" in args or "--help" in args

    @staticmethod
    def _replace_clone_destination(
        args: List[str], destination_index: Optional[int], cache_entry: Path
    ) -> List[str]:
        cache_args = list(args)
        if destination_index is None:
            cache_args.append(str(cache_entry))
        else:
            cache_args[destination_index] = str(cache_entry)
        return cache_args

    @staticmethod
    def _parse_fetch_repository(args: List[str]) -> Optional[Tuple[str, int]]:
        option_value_names = {
            "--upload-pack",
            "--depth",
            "--deepen",
            "--shallow-since",
            "--shallow-exclude",
            "--jobs",
            "-j",
            "--server-option",
            "--negotiation-tip",
            "--recurse-submodules",
            "--refmap",
            "-o",
            "--set-upstream",
            "--submodule-prefix",
            "--filter",
        }
        skip_cache_options = {"--all", "--multiple", "--stdin"}
        end_of_options = False
        i = 0
        while i < len(args):
            arg = args[i]
            if not end_of_options and arg == "--":
                end_of_options = True
                i += 1
                continue
            if not end_of_options and arg in skip_cache_options:
                return None
            if not end_of_options and arg.startswith("--"):
                option_name = arg.split("=", 1)[0]
                i += 2 if option_name in option_value_names and "=" not in arg else 1
                continue
            if not end_of_options and arg.startswith("-") and arg != "-":
                i += 2 if arg in option_value_names else 1
                continue
            return arg, i

        return None

    def _resolve_fetch_repository_url(
        self, repository: str, global_args: List[str], cwd: Path
    ) -> Optional[str]:
        if self._is_explicit_repository(repository):
            return repository

        try:
            url = self._run_git_output([*global_args, "remote", "get-url", repository], cwd=cwd)
        except subprocess.CalledProcessError:
            return None
        return url.strip()

    @staticmethod
    def _is_explicit_repository(repository: str) -> bool:
        return "://" in repository or "@" in repository

    @staticmethod
    def _is_local_repository_path(repository: str) -> bool:
        repository_path = Path(repository)
        return repository_path.exists() or repository_path.is_absolute()

    @staticmethod
    def _get_clone_destination_path(repository: str, destination: Optional[str], cwd: Path) -> Path:
        if destination is not None:
            destination_path = Path(destination)
        else:
            repository_name = repository.rstrip("/").rsplit("/", 1)[-1]
            if repository_name.endswith(".git"):
                repository_name = repository_name[:-4]
            destination_path = Path(repository_name)

        if destination_path.is_absolute():
            return destination_path
        return cwd / destination_path

    @staticmethod
    def _extract_global_cwd(global_args: List[str]) -> Tuple[Path, List[str]]:
        cwd = Path.cwd()
        filtered_args: List[str] = []
        i = 0
        while i < len(global_args):
            arg = global_args[i]
            if arg == "-C" and i + 1 < len(global_args):
                cwd = Git._resolve_git_cwd(cwd, global_args[i + 1])
                i += 2
                continue
            if arg.startswith("-C") and len(arg) > _SHORT_OPTION_WITH_VALUE_LENGTH:
                cwd = Git._resolve_git_cwd(cwd, arg[2:])
                i += 1
                continue
            filtered_args.append(arg)
            i += 1
        return cwd, filtered_args

    @staticmethod
    def _resolve_git_cwd(cwd: Path, path: str) -> Path:
        new_cwd = Path(path)
        if new_cwd.is_absolute():
            return new_cwd
        return cwd / new_cwd

    @staticmethod
    def _copy_clone_cache(cache_entry: Path, destination_path: Path):
        if destination_path.exists() and any(destination_path.iterdir()):
            raise FileExistsError(
                f"destination path '{destination_path}' already exists and is not empty"
            )
        destination_path.parent.mkdir(parents=True, exist_ok=True)
        shutil.copytree(cache_entry, destination_path, symlinks=True, dirs_exist_ok=True)

    def _update_clone_cache(self, cache_entry: Path, global_args: List[str]) -> int:
        fetch_exit_code = self._run_git_for_cache(
            [*global_args, "-C", str(cache_entry), "fetch", "--all", "--tags", "--prune"]
        )
        if fetch_exit_code != 0:
            return fetch_exit_code
        return self._run_git_for_cache([*global_args, "-C", str(cache_entry), "pull", "--ff-only"])

    def _create_fetch_cache_if_missing(self, repository_url: str, cache_entry: Path) -> int:
        if self._is_git_repository(cache_entry):
            return 0

        shutil.rmtree(cache_entry, ignore_errors=True)
        exit_code = self._run_git_for_cache(["clone", "--mirror", repository_url, str(cache_entry)])
        if exit_code != 0:
            shutil.rmtree(cache_entry, ignore_errors=True)
        return exit_code

    def _run_git(self, args: List[str], cwd: Optional[Path] = None) -> int:
        command = [Git.find_git(), *args]
        return subprocess.call(command, cwd=str(cwd) if cwd is not None else None)

    def _run_git_for_cache(self, args: List[str], cwd: Optional[Path] = None) -> int:
        command = [Git.find_git(), *args]
        return subprocess.call(
            command, cwd=str(cwd) if cwd is not None else None, env=self._get_cache_git_env()
        )

    def _run_git_output(self, args: List[str], cwd: Optional[Path] = None) -> str:
        command = [Git.find_git(), *args]
        output = subprocess.check_output(command, cwd=str(cwd) if cwd is not None else None)
        return output.decode("utf-8")

    @staticmethod
    def _get_cache_git_env() -> dict:
        env = dict(os.environ)
        for key in _GIT_ENVIRONMENT_VARIABLES:
            env.pop(key, None)
        return env

    @staticmethod
    def _is_git_repository(path: Path) -> bool:
        return (path / "objects").is_dir() and (path / "HEAD").is_file()

    def _is_fetch_target_empty(self, cwd: Path) -> bool:
        object_directory = os.environ.get("GIT_OBJECT_DIRECTORY")
        if object_directory is None:
            try:
                object_directory = self._run_git_output(["rev-parse", "--git-path", "objects"], cwd)
            except subprocess.CalledProcessError:
                return False

        object_path = Path(object_directory.strip())
        return not self._object_store_has_objects(object_path)

    @staticmethod
    def _object_store_has_objects(object_path: Path) -> bool:
        if not object_path.exists():
            return False

        pack_path = object_path / "pack"
        if pack_path.exists() and any(pack_path.glob("*.pack")):
            return True

        for child in object_path.iterdir():
            if child.name in {"info", "pack"} or not child.is_dir():
                continue
            try:
                next(child.iterdir())
                return True
            except StopIteration:
                continue

        return False

__init__(configuration)

Create a new git wrapper object.

Parameters:

Name	Type	Description	Default
configuration	Configuration	The configuration object to be used by the object.	required

Source code in repo_on_fire/git.py

def __init__(self, configuration: Configuration):
    """Create a new git wrapper object.

    Args:
        configuration: The configuration object to be used by the object.
    """
    self._configuration = configuration

clone(args, global_args=None)

Clone a repository using the local git cache.

The first clone for a specific clone request is run in the cache. Later clones update that cache entry and copy the cached working tree to the originally requested destination.

Parameters:

Name	Type	Description	Default
args	List[str]	Arguments passed after git clone.	required
global_args	Optional[List[str]]	Arguments passed before clone to the git command itself.	None

Returns:

Type	Description
int	The native git-compatible exit code.

Source code in repo_on_fire/git.py

def clone(self, args: List[str], global_args: Optional[List[str]] = None) -> int:
    """Clone a repository using the local git cache.

    The first clone for a specific clone request is run in the cache. Later
    clones update that cache entry and copy the cached working tree to the
    originally requested destination.

    Args:
        args: Arguments passed after ``git clone``.
        global_args: Arguments passed before ``clone`` to the ``git``
            command itself.

    Returns:
        The native git-compatible exit code.
    """
    if global_args is None:
        global_args = []

    effective_cwd, filtered_global_args = self._extract_global_cwd(global_args)
    if self._is_clone_help_requested(args):
        return self._run_git([*filtered_global_args, "clone", *args], cwd=effective_cwd)

    try:
        repository, destination, destination_index = self._parse_clone_args(args)
    except ValueError:
        return self._run_git([*filtered_global_args, "clone", *args], cwd=effective_cwd)

    cache_entry = self.get_clone_cache_entry(args, destination_index)
    cache_args = self._replace_clone_destination(args, destination_index, cache_entry)
    destination_path = self._get_clone_destination_path(
        repository, destination, cwd=effective_cwd
    )

    with lock_path(cache_entry):
        if not cache_entry.exists():
            exit_code = self._run_git_for_cache(
                [*filtered_global_args, "clone", *cache_args], cwd=effective_cwd
            )
            if exit_code != 0:
                shutil.rmtree(cache_entry, ignore_errors=True)
                return exit_code
        else:
            exit_code = self._update_clone_cache(cache_entry, filtered_global_args)
            if exit_code != 0:
                return exit_code

        try:
            self._copy_clone_cache(cache_entry, destination_path)
        except OSError as error:
            print(f"git-on-fire: could not copy cached clone: {error}", file=sys.stderr)
            return 128

    return 0

fetch(args, global_args=None)

Fetch through the local git cache when a repository can be identified.

Source code in repo_on_fire/git.py

def fetch(self, args: List[str], global_args: Optional[List[str]] = None) -> int:
    """Fetch through the local git cache when a repository can be identified."""
    if global_args is None:
        global_args = []

    effective_cwd, filtered_global_args = self._extract_global_cwd(global_args)
    native_fetch_args = [*filtered_global_args, "fetch", *args]
    fetch_repository = self._parse_fetch_repository(args)
    if fetch_repository is None:
        return self._run_git(native_fetch_args, cwd=effective_cwd)

    repository, repository_index = fetch_repository
    if self._is_local_repository_path(repository):
        return self._run_git(native_fetch_args, cwd=effective_cwd)
    if not self._is_fetch_target_empty(effective_cwd):
        return self._run_git(native_fetch_args, cwd=effective_cwd)

    repository_url = self._resolve_fetch_repository_url(
        repository, filtered_global_args, effective_cwd
    )
    if repository_url is None:
        return self._run_git(native_fetch_args, cwd=effective_cwd)

    cache_entry = self.get_fetch_cache_entry(repository_url)

    with lock_path(cache_entry):
        if not self._is_git_repository(cache_entry):
            exit_code = self._create_fetch_cache_if_missing(repository_url, cache_entry)
        else:
            exit_code = self._run_git_for_cache(
                [
                    *filtered_global_args,
                    "-C",
                    str(cache_entry),
                    "fetch",
                    "--all",
                    "--tags",
                    "--prune",
                ]
            )

        if exit_code != 0:
            return exit_code

    cache_args = list(args)
    cache_args[repository_index] = str(cache_entry)
    return self._run_git([*filtered_global_args, "fetch", *cache_args], cwd=effective_cwd)

find_git() staticmethod

Find the git command.

This method can be used to find the git command. This method assumes that the current process has been called by a previously created git wrapper (using the make_git_wrapper method). In this case, special care is taken to avoid that the wrapper recursively finds itself, which would lead to an infinite loop (and exhaustion of resources of the host).

Raises:

Type	Description
ExecutableNotFound	The git command could not be found.

Source code in repo_on_fire/git.py

@staticmethod
def find_git():
    """Find the git command.

    This method can be used to find the git command. This method assumes
    that the current process has been called by a previously created
    git wrapper (using the [make_git_wrapper][repo_on_fire.git.Git.make_git_wrapper] method).
    In this case, special care is taken to avoid that the wrapper recursively finds
    itself, which would lead to an infinite loop (and exhaustion of
    resources of the host).

    Raises:
        ExecutableNotFound: The git command could not be found.
    """
    # Compile a list of search paths. There are two possibilities:
    GIT_ON_FIRE_WRAPPER_DIR = "GIT_ON_FIRE_WRAPPER_DIR"
    if GIT_ON_FIRE_WRAPPER_DIR in os.environ:
        # We are called from a git wrapper. In this case, the list of
        # paths to look for git is compiled by searching PATH for the
        # directory where the wrapper is stored in and - if found - we only
        # include the paths after the entry.
        path_variable = os.environ.get("PATH", os.defpath)
        paths = path_variable.split(os.pathsep)
        wrapper_path = Path(os.environ[GIT_ON_FIRE_WRAPPER_DIR]).resolve()
        search_paths = []
        for i, path in enumerate(paths):
            path_path = Path(path).resolve()
            if wrapper_path == path_path:
                search_paths = paths[i + 1 :]
                break
        if len(search_paths) == 0:
            # The wrapper path was not in PATH. This can happen if the
            # wrapper is called without adding its path to PATH. Fall back
            # to using the default path.
            search_paths = paths
        joined_path = os.pathsep.join(search_paths)
        git = shutil.which("git", path=joined_path)
        if git is None:
            raise ExecutableNotFound(f"Did not find git in '{joined_path}'")
        return git
    else:
        # We didn't end here due to a call to the git wrapper. Hence,
        # simply return git from wherever we can find it:
        git = shutil.which("git")
        if git is None:
            raise ExecutableNotFound("Unable to find git executable")
        return git

find_repo_root(cwd=None) staticmethod

Try to find the rep root directory.

This method looks for the .repo directory in cwd and recursively upwards.

Parameters:

Name	Type	Description	Default
cwd	Optional[Union[str, Path]]	The directory where to start searching. The method starts looking for the .repo directory there and follows upwards in the file system hierarchy. If None, search starts in os.curdir.	None

Raises:

Type	Description
RepoRootNotFound	If no .repo directory can be found.

Source code in repo_on_fire/git.py

@staticmethod
def find_repo_root(cwd: Optional[Union[str, Path]] = None):
    """Try to find the rep root directory.

    This method looks for the ``.repo`` directory in ``cwd`` and recursively
    upwards.

    Args:
        cwd: The directory where to start searching. The method starts
             looking for the ``.repo`` directory there and follows upwards
             in the file system hierarchy. If None, search starts
             in ``os.curdir``.

    Raises:
        RepoRootNotFound: If no ``.repo`` directory can be found.
    """
    if cwd is None:
        cwd = os.getcwd()
    path = Path(cwd)
    while path != path.parent:
        repo_dir = path / ".repo"
        if repo_dir.exists():
            return path
        path = path.parent
    raise RepoRootNotFound(f"Could not find .repo in '{cwd}")

get_clone_cache_entry(args, destination_index)

Get the cache entry path for a git clone request.

Source code in repo_on_fire/git.py

def get_clone_cache_entry(self, args: List[str], destination_index: Optional[int]) -> Path:
    """Get the cache entry path for a ``git clone`` request."""
    cache_key_args = [
        arg
        for index, arg in enumerate(args)
        if destination_index is None or index != destination_index
    ]
    cache_id = sha256("\0".join(cache_key_args).encode("utf-8")).hexdigest()
    return self._configuration.cache_path / "git" / cache_id[0:2] / cache_id

get_fetch_cache_entry(repository_url)

Get the cache entry path for a fetch repository.

Source code in repo_on_fire/git.py

def get_fetch_cache_entry(self, repository_url: str) -> Path:
    """Get the cache entry path for a fetch repository."""
    cache_id = sha256(f"fetch\0{repository_url}".encode()).hexdigest()
    return self._configuration.cache_path / "git" / cache_id[0:2] / cache_id

ls_remote(repository_url, ref)

Query a remote repository for a ref.

Source code in repo_on_fire/git.py

def ls_remote(self, repository_url: str, ref: str) -> str:
    """Query a remote repository for a ref."""
    return self._run_git_output(["ls-remote", repository_url, ref])

make_git_wrapper(directory_path) staticmethod

Create a wrapper which is callable as git but invoke git-on-fire.

This function can be used to create a platform native wrapper script, which can be invoked using git but instead, it will call git-on-fire.

Parameters:

Name	Type	Description	Default
directory_path	Path	The path to the directory where to create the wrapper.	required

Source code in repo_on_fire/git.py

@staticmethod
def make_git_wrapper(directory_path: Path):
    """Create a wrapper which is callable as ``git`` but invoke ``git-on-fire``.

    This function can be used to create a platform native wrapper script,
    which can be invoked using ``git`` but instead, it will call
    ``git-on-fire``.

    Args:
        directory_path: The path to the directory where to create the
            wrapper.
    """
    if platform.system() == "Windows":
        template = _WINDOWS_WRAPPER_SCRIPT
        suffix = ".bat"
        make_executable = False
    else:
        template = _UNIX_WRAPPER_SCRIPT
        suffix = ""
        make_executable = True
    native_git_executable = Git.find_git()
    script_content = template.format(
        native_git_executable=native_git_executable,
        native_git_executable_sh=shlex.quote(native_git_executable),
        python_executable=sys.executable,
        python_executable_sh=shlex.quote(sys.executable),
    )
    directory_path.mkdir(parents=True, exist_ok=True)
    file_path = directory_path / ("git" + suffix)
    if file_path.exists():
        # If the file exists, check if we need to update the content:
        if file_path.read_text(encoding="utf-8") != script_content:
            file_path.write_text(script_content, encoding="utf-8")
    else:
        # Otherwise, always write:
        file_path.write_text(script_content, encoding="utf-8")
    if make_executable:
        os.chmod(file_path, 0o755)

with_git_wrapper(cwd=None)

Run a command with a patched git command.

This method implements a context manager which can be used to run commands with a modified PATH environment variable such that instead of the usual git, a wrapper is called which provides addon functionality. The wrapper is created using make_git_wrapper.

Parameters:

Name	Type	Description	Default
cwd	Optional[Union[str, Path]]	The path where to search for the repo workspace. If set to None, the current working directory will be used instead.	None

Source code in repo_on_fire/git.py

@contextmanager
def with_git_wrapper(self, cwd: Optional[Union[str, Path]] = None):
    """Run a command with a patched git command.

    This method implements a context manager which can be used to run
    commands with a modified ``PATH`` environment variable such that
    instead of the usual ``git``, a wrapper is called which provides
    addon functionality. The wrapper is created using
    [make_git_wrapper][repo_on_fire.git.Git.make_git_wrapper].

    Args:
        cwd: The path where to search for the repo workspace. If set to
             None, the current working directory will be used instead.
    """
    try:
        repo_root = Git.find_repo_root(cwd)
        try:
            # Create a git wrapper in the repo folder
            wrapper_path = repo_root / ".repo" / "repo-on-fire" / "bin"
            Git.make_git_wrapper(wrapper_path)
            env = os.environ
            path = env.get("PATH", "")
            modified_path = os.pathsep.join([str(wrapper_path), path])
            os.environ["PATH"] = modified_path
            yield dict(os.environ)
        finally:
            os.environ["PATH"] = path
    except RepoRootNotFound:
        # If we are outside a workspace, create a wrapper in a temporary
        # folder:
        with TemporaryDirectory() as tmp_dir:
            try:
                # Create a git wrapper in the temp dir
                wrapper_path = Path(tmp_dir)
                Git.make_git_wrapper(wrapper_path)
                env = os.environ
                path = env.get("PATH", "")
                modified_path = os.pathsep.join([str(wrapper_path), path])
                os.environ["PATH"] = modified_path
                yield dict(os.environ)
            finally:
                os.environ["PATH"] = path

cli()

Git on Fire Command Line.

Source code in repo_on_fire/git.py

@click.group("Git on Fire Commands")
def cli():
    """Git on Fire Command Line."""

main()

Git on Fire Entry Point.

Source code in repo_on_fire/git.py

def main():
    """Git on Fire Entry Point."""
    command_index = _find_git_command_index(sys.argv[1:])
    if command_index is not None:
        command_index += 1
        if sys.argv[command_index] == "clone":
            git = Git(Configuration())
            sys.exit(git.clone(sys.argv[command_index + 1 :], sys.argv[1:command_index]))
        if sys.argv[command_index] == "fetch":
            git = Git(Configuration())
            sys.exit(git.fetch(sys.argv[command_index + 1 :], sys.argv[1:command_index]))

    # Check if the first non-option argument is a known command
    for _, arg in enumerate(sys.argv[1:], 1):
        if not arg.startswith("-") and arg in cli.commands:
            break
    else:
        # Insert 'unknown' as the command to handle unknown subcommands.
        # The actual list of arguments now looks like this:
        # ["unknown", "push", "-u", "origin", "HEAD"]
        sys.argv.insert(1, "unknown")
    cli()