edq.util.git

  1"""
  2Handle interfacing with git repos.
  3"""
  4
  5import os
  6import re
  7import typing
  8
  9import git
 10
 11UNKNOWN_VERSION: str = 'UNKNOWN'
 12VERSION_LEN: int = 8
 13DIRTY_SIFFIX: str = '-d'
 14
 15def get_version(path: str = '.', throw: bool = False) -> str:
 16    """
 17    Get a version string from the git repo.
 18    This is just a commit hash with some dressup.
 19    """
 20
 21    if (os.path.isfile(path)):
 22        path = os.path.dirname(path)
 23
 24    try:
 25        repo = git.Repo(path, search_parent_directories = True)
 26        version = repo.head.commit.tree.hexsha[:VERSION_LEN]
 27    except Exception as ex:
 28        if (throw):
 29            raise ValueError(f"Path '{path}' is not a valid Git repo.") from ex
 30
 31        return UNKNOWN_VERSION
 32
 33    if (repo.is_dirty()):
 34        version += DIRTY_SIFFIX
 35
 36    return version
 37
 38def ensure_repo(
 39        url: str,
 40        path: str,
 41        username: typing.Union[str, None] = None,
 42        token: typing.Union[str, None] = None,
 43        update: bool = False,
 44        ref: typing.Union[str, None] = None,
 45        ) -> None:
 46    """
 47    Ensure that a git repo exists locally.
 48    Clone the repo if it does not exist.
 49    Optionally update (pull) the repo.
 50    """
 51
 52    if (os.path.isfile(path)):
 53        raise ValueError(f"Target git path exists and is a file: '{path}'.")
 54
 55    if (not os.path.exists(path)):
 56        clone(url, path, username = username, token = token)
 57
 58    repo = get_repo(path)
 59
 60    if (update):
 61        update_repo(repo)
 62
 63    if (ref is not None):
 64        checkout_repo(repo, ref)
 65
 66def get_repo(path: str) -> git.Repo:
 67    """ Get a reference to a git repo. """
 68
 69    return git.Repo(path)
 70
 71def clone(
 72        url: str,
 73        path: str,
 74        username: typing.Union[str, None] = None,
 75        token: typing.Union[str, None] = None,
 76        ) -> git.Repo:
 77    """ Clone a git repo to the target location and return a reference to it. """
 78
 79    # If we have a username or password, we need to rewrite the URL.
 80    # This is not very robust, but should work.
 81    # After clone, the credentials are saved in the local git config.
 82    if (username is not None):
 83        if (token is None):
 84            raise ValueError("If username is specified, a token must also be specified.")
 85
 86        auth_text = f"{username}:{token}"
 87        url = re.sub(r'(http(s?)://)', rf"\1{auth_text}@", url)
 88
 89    return git.Repo.clone_from(url, path)
 90
 91def checkout_repo(repo: git.Repo, ref: str) -> None:
 92    """ Checkout the given reference on the given repo. """
 93
 94    repo.git.checkout(ref)
 95
 96def update_repo(repo: git.Repo) -> bool:
 97    """
 98    Update (pull) the given repo.
 99    Return true if an update occurred.
100    """
101
102    fetch_results = repo.remotes.origin.pull()
103
104    for fetch_result in fetch_results:
105        if (fetch_result.ref.name != f"origin/{repo.active_branch.name}"):
106            continue
107
108        if (fetch_result.flags == git.remote.FetchInfo.HEAD_UPTODATE):
109            return False
110
111        return True
112
113    return False