"""'Git' type Path Source."""
from __future__ import absolute_import
import logging
import os
import shutil
import subprocess
import sys
import tempfile
from typing import Dict, List, Optional, Union # noqa pylint: disable=W
from .source import Source
LOGGER = logging.getLogger(__name__)
[docs]class Git(Source):
"""Git Path Source.
The Git path source can be tasked with cloning a remote repository
and pointing to a specific module folder (or the root).
"""
# Added for documentation purposes
def __init__(self, uri="", location="", options=None, **kwargs):
# type(Dict[str, Union[str, Dict[str, str]]]) -> Source
"""Git Path Source.
Keyword Args:
uri (str): The uniform resource identifier that targets the remote git
repository
location (string): The relative location to the root of the
repository where the module resides. Leaving this as an empty
string, ``/``, or ``./`` will have runway look in the root folder.
options (Union(None, Dict[str, str])): A reference can be passed along via the
options so that a specific version of the repository is cloned.
**commit**, **tag**, **branch** are all valid keys with
respective output
"""
self.uri = uri
self.location = location
self.options = options
if not self.options:
self.options = {}
super(Git, self).__init__(**kwargs)
[docs] def fetch(self):
# type: () -> str
"""Retrieve the git repository from it's remote location."""
from git import Repo # pylint: disable=import-outside-toplevel
ref = self.__determine_git_ref() # type: str
dir_name = "_".join([self.sanitize_git_path(self.uri), ref]) # type: str
cached_dir_path = os.path.join(self.cache_dir, dir_name) # type: str
cached_path = "" # type: str
if not os.path.isdir(cached_dir_path):
tmp_dir = tempfile.mkdtemp()
try:
tmp_repo_path = os.path.join(tmp_dir, dir_name) # type: str
with Repo.clone_from(self.uri, tmp_repo_path) as repo:
repo.head.reference = ref # type: str
repo.head.reset(index=True, working_tree=True)
shutil.move(tmp_repo_path, self.cache_dir)
cached_path = os.path.join(self.cache_dir, dir_name) # type: str
finally:
shutil.rmtree(tmp_dir)
else:
cached_path = cached_dir_path # type: str
return os.path.join(cached_path, self.location)
def __git_ls_remote(self, ref):
# type: (str) -> str
"""List remote repositories based on uri and ref received.
Keyword Args:
ref (str): The git reference value
"""
cmd = ["git", "ls-remote", self.uri, ref]
LOGGER.debug("getting commit ID from repo: %s", " ".join(cmd))
lsremote_output = subprocess.check_output(cmd)
# pylint: disable=unsupported-membership-test
if b"\t" in lsremote_output:
commit_id = lsremote_output.split(b"\t")[0] # type List[str]
LOGGER.debug("matching commit id found: %s", commit_id)
return commit_id
raise ValueError('Ref "%s" not found for repo %s.' % (ref, self.uri))
def __determine_git_ls_remote_ref(self):
# type: () -> str
"""Determine remote ref, defaulting to HEAD unless a branch is found."""
ref = "HEAD"
if self.options.get("branch"):
ref = "refs/heads/%s" % self.options.get("branch") # type: str
return ref
def __determine_git_ref(self):
# type: () -> str
"""Determine the git reference code."""
ref_config_keys = 0 # type: int
for i in ["commit", "tag", "branch"]:
if self.options.get(i):
ref_config_keys += 1
if ref_config_keys > 1:
raise ImportError(
"Fetching remote git sources failed: conflicting revisions "
"(e.g. 'commit', 'tag', 'branch') specified for a package source"
)
if self.options.get("commit"):
ref = self.options.get("commit") # type: str
elif self.options.get("tag"):
ref = self.options.get("tag") # type: str
else:
ref = self.__git_ls_remote(
self.__determine_git_ls_remote_ref()
) # ty pe: str
if sys.version_info[0] > 2 and isinstance(ref, bytes):
return ref.decode()
return ref
[docs] @classmethod
def sanitize_git_path(cls, path):
# type(str) -> str
"""Sanitize the git path for folder/file assignment.
Keyword Args:
path (str): The path string to be sanitized
"""
dir_name = path # type: str
split = path.split("//") # type: List[str]
domain = split[len(split) - 1] # type: str
if domain.endswith(".git"):
dir_name = domain[:-4]
return cls.sanitize_directory_path(dir_name)