"""
Static helper functions for cloning and working with a Git repository
"""
import errno
import os
import re
from contextlib import contextmanager
from urllib.parse import urlparse
from git import Git, Repo
from git.exc import GitCommandError
from pydantic import BaseModel, model_validator
[docs]
class GitArgs(BaseModel):
"""
GitArgs captures information required to identify scripts
located in git repositories.
"""
git_repo: str | None = "https://gitlab.com/ska-telescope/oso/ska-oso-scripting.git"
git_branch: str | None = None
git_commit: str | None = None
[docs]
def __init__(
self,
git_repo: str = "https://gitlab.com/ska-telescope/oso/ska-oso-scripting.git",
git_branch: str | None = None,
git_commit: str | None = None,
):
super(GitArgs, self).__init__(
git_repo=git_repo, git_branch=git_branch, git_commit=git_commit
)
@model_validator(mode="after")
def post_init(self):
# We only want to set the default branch if the commit isn't set, as the user
# might just give a commit hash from a feature branch
if self.git_branch is None and self.git_commit is None:
self.git_branch = "master"
return self
class GitManager:
base_dir = "/tmp/clones/"
@staticmethod
def clone_repo(git_args: GitArgs) -> str:
"""
Clone a remote repository into the local filesystem, with the HEAD pointing to the revision
defined in the input
If a Git commit hash is not supplied, a shallow clone of the branch is done, minimising the
data transferred over the network. If a Git commit hash is supplied, the full repo must be
cloned and then the commit checked out, as Git doesn't allow a specific commit to be cloned
:param git_args: Information about the repository and the required point in its history
:return: Returns the location of the cloned project
"""
git_commit = GitManager.get_commit_hash(git_args)
clone_args = {}
if not git_args.git_commit:
clone_args["depth"] = 1
clone_args["single_branch"] = True
clone_args["branch"] = git_args.git_branch
project_name = GitManager.get_project_name(git_args.git_repo)
clone_dir = GitManager.base_dir + project_name + "/" + git_commit
if os.path.exists(clone_dir):
return clone_dir
with GitManager._translate_missing_repo(git_args.git_repo):
Repo.clone_from(git_args.git_repo, clone_dir, **clone_args)
if not os.path.exists(clone_dir):
raise IOError(
"Something went wrong when cloning the project, directory"
f" {clone_dir} does not exist"
)
if git_args.git_commit:
GitManager._checkout_commit(clone_dir, git_args.git_commit)
return clone_dir
@staticmethod
def get_commit_hash(git_args: GitArgs, short_hash=False) -> str:
"""
Get a commit hash from a remote repository
:param git_args: Arguments to point to git environment to get hash for
:param short_hash: Return first 7 characters of the hash
:return: The SHA for the specified commit.
If a tag and a branch are both supplied, the tag takes precedence.
If neither are supplied, the latest commit on the default branch is used
"""
if git_args.git_commit:
return git_args.git_commit
with GitManager._translate_missing_repo(git_args.git_repo):
if git_args.git_branch != "master":
response = Git().ls_remote("-h", git_args.git_repo, git_args.git_branch)
else:
response = Git().ls_remote(git_args.git_repo, "HEAD")
if short_hash:
return response[:7]
return response[:40]
@staticmethod
def get_project_name(git_repo: str):
"""Get the git project name including full folder tree to avoid project
name clashes (e.g. name for project at http://gitlab.com/ska-telescope/oso/ska-oso-scripting
is ska-telescope-oso-ska-oso-scripting)"""
return urlparse(git_repo).path[1:].replace("/", "-").split(".")[0]
@staticmethod
def _checkout_commit(location: str, hexsha: str) -> None:
"""
Checkout an existing repository to a specific commit
:param location: The filepath location of the repository
:param hexsha: The commit SHA to check out
:return: None, but has the side effect changing the files
inside the repository to the state they were in at the commit
"""
path = os.path.abspath(location)
Repo(path).git.checkout(hexsha)
@staticmethod
@contextmanager
def _translate_missing_repo(git_repo: str):
"""
Wrap a git operation so that a GitCommandError indicating a missing or
unreachable repository is translated into a FileNotFoundError, giving
callers a clear "repository not found" rather than a raw git error. All
other git errors propagate unchanged.
:param git_repo: The repository URL/path used in the wrapped operation,
surfaced as the FileNotFoundError filename.
"""
try:
yield
except GitCommandError as err:
git_error = f"{err.stderr or ''}\n{err.stdout or ''}".lower()
# These phrases indicate the repository itself is missing or
# unreachable. They are deliberately specific: a bare "not found"
# would also match the error raised when cloning an existing
# repository with the wrong branch ("remote branch <name> not found
# in upstream origin"), which would incorrectly report that the
# repository does not exist.
if (
re.search(r"repository\b.*\bnot found", git_error)
or "does not appear to be a git repository" in git_error
or "no such device or address" in git_error
or "could not read from remote repository" in git_error
):
raise FileNotFoundError(
errno.ENOENT, "Git repository not found", git_repo
) from None
raise