commit: Dict[str, 'Commit']

linenos: range

orig_path: Optional[str]

"""Represents a git repository and allows you to query references,

DAEMON_EXPORT_FILE = 'git-daemon-export-ok'

git = cast('Git', None) # Must exist, or __del__ will fail in case we raise on `__init__()`

working_dir: Optional[PathLike] = None

_working_tree_dir: Optional[PathLike] = None

git_dir: PathLike = ""

_common_dir: PathLike = ""

# precompiled regex

re_whitespace = re.compile(r'\s+')

re_hexsha_only = re.compile('^[0-9A-Fa-f]{40}$')

re_hexsha_shortened = re.compile('^[0-9A-Fa-f]{4,40}$')

re_envvars = re.compile(r'(\$(\{\s?)?[a-zA-Z_]\w*(\}\s?)?|%\s?[a-zA-Z_]\w*\s?%)')

re_author_committer_start = re.compile(r'^(author|committer)')

re_tab_full_line = re.compile(r'^\t(.*)$')

# invariants

# represents the configuration level of a configuration file

config_level: ConfigLevels_Tup = ("system", "user", "global", "repository")

# Subclass configuration

# Subclasses may easily bring in their own custom types by placing a constructor or type here

GitCommandWrapperType = Git

def __init__(self, path: Optional[PathLike] = None, odbt: Type[LooseObjectDB] = GitCmdObjectDB,

search_parent_directories: bool = False, expand_vars: bool = True) -> None:

"""Create a new Repo instance

epath = path or os.getenv('GIT_DIR')

if not epath:

epath = os.getcwd()

if Git.is_cygwin():

epath = decygpath(epath)

epath = epath or path or os.getcwd()

if not isinstance(epath, str):

epath = str(epath)

if expand_vars and re.search(self.re_envvars, epath):

warnings.warn("The use of environment variables in paths is deprecated" +

"\nfor security reasons and may be removed in the future!!")

epath = expand_path(epath, expand_vars)

if epath is not None:

if not os.path.exists(epath):

raise NoSuchPathError(epath)

## Walk up the path to find the `.git` dir.

#

curpath = epath

while curpath:

# ABOUT osp.NORMPATH

# It's important to normalize the paths, as submodules will otherwise initialize their

# repo instances with paths that depend on path-portions that will not exist after being

# removed. It's just cleaner.

if is_git_dir(curpath):

self.git_dir = curpath

# from man git-config : core.worktree

# Set the path to the root of the working tree. If GIT_COMMON_DIR environment

# variable is set, core.worktree is ignored and not used for determining the

# root of working tree. This can be overridden by the GIT_WORK_TREE environment

# variable. The value can be an absolute path or relative to the path to the .git

# directory, which is either specified by GIT_DIR, or automatically discovered.

# If GIT_DIR is specified but none of GIT_WORK_TREE and core.worktree is specified,

# the current working directory is regarded as the top level of your working tree.

self._working_tree_dir = os.path.dirname(self.git_dir)

if os.environ.get('GIT_COMMON_DIR') is None:

gitconf = self.config_reader("repository")

if gitconf.has_option('core', 'worktree'):

self._working_tree_dir = gitconf.get('core', 'worktree')

if 'GIT_WORK_TREE' in os.environ:

self._working_tree_dir = os.getenv('GIT_WORK_TREE')

break

dotgit = osp.join(curpath, '.git')

sm_gitpath = find_submodule_git_dir(dotgit)

if sm_gitpath is not None:

self.git_dir = osp.normpath(sm_gitpath)

sm_gitpath = find_submodule_git_dir(dotgit)

if sm_gitpath is None:

sm_gitpath = find_worktree_git_dir(dotgit)

if sm_gitpath is not None:

self.git_dir = expand_path(sm_gitpath, expand_vars)

self._working_tree_dir = curpath

break

if not search_parent_directories:

break

curpath, tail = osp.split(curpath)

if not tail:

break

# END while curpath

if self.git_dir is None:

raise InvalidGitRepositoryError(epath)

self._bare = False

try:

self._bare = self.config_reader("repository").getboolean('core', 'bare')

except Exception:

# lets not assume the option exists, although it should

pass

try:

common_dir = open(osp.join(self.git_dir, 'commondir'), 'rt').readlines()[0].strip()

self._common_dir = osp.join(self.git_dir, common_dir)

except OSError:

self._common_dir = ""

# adjust the wd in case we are actually bare - we didn't know that

# in the first place

if self._bare:

self._working_tree_dir = None

# END working dir handling

self.working_dir: Optional[PathLike] = self._working_tree_dir or self.common_dir

self.git = self.GitCommandWrapperType(self.working_dir)

# special handling, in special times

rootpath = osp.join(self.common_dir, 'objects')

if issubclass(odbt, GitCmdObjectDB):

self.odb = odbt(rootpath, self.git)

else:

self.odb = odbt(rootpath)

def __enter__(self) -> 'Repo':

return self

def __exit__(self, *args: Any) -> None:

self.close()

def __del__(self) -> None:

try:

self.close()

except Exception:

pass

def close(self) -> None:

if self.git:

self.git.clear_cache()

# Tempfiles objects on Windows are holding references to

# open files until they are collected by the garbage

# collector, thus preventing deletion.

# TODO: Find these references and ensure they are closed

# and deleted synchronously rather than forcing a gc

# collection.

if is_win:

gc.collect()

gitdb.util.mman.collect()

if is_win:

gc.collect()

def __eq__(self, rhs: object) -> bool:

if isinstance(rhs, Repo) and self.git_dir:

return self.git_dir == rhs.git_dir

return False

def __ne__(self, rhs: object) -> bool:

return not self.__eq__(rhs)

def __hash__(self) -> int:

return hash(self.git_dir)

# Description property

def _get_description(self) -> str:

if self.git_dir:

filename = osp.join(self.git_dir, 'description')

with open(filename, 'rb') as fp:

return fp.read().rstrip().decode(defenc)

def _set_description(self, descr: str) -> None:

if self.git_dir:

filename = osp.join(self.git_dir, 'description')

with open(filename, 'wb') as fp:

fp.write((descr + '\n').encode(defenc))

description = property(_get_description, _set_description,

doc="the project's description")

del _get_description

del _set_description

@property

def working_tree_dir(self) -> Optional[PathLike]:

""":return: The working tree directory of our git repository. If this is a bare repository, None is returned.

return self._working_tree_dir

@property

def common_dir(self) -> PathLike:

"""

if self._common_dir:

return self._common_dir

elif self.git_dir:

return self.git_dir

else:

# or could return ""

raise InvalidGitRepositoryError()

@property

def bare(self) -> bool:

""":return: True if the repository is bare"""

return self._bare

@property

def heads(self) -> 'IterableList[Head]':

"""A list of ``Head`` objects representing the branch heads in

return Head.list_items(self)

@property

def references(self) -> 'IterableList[Reference]':

"""A list of Reference objects representing tags, heads and remote references.

return Reference.list_items(self)

# alias for references

refs = references

# alias for heads

branches = heads

@property

def index(self) -> 'IndexFile':

""":return: IndexFile representing this repository's index.

return IndexFile(self)

@property

def head(self) -> 'HEAD':

""":return: HEAD Object pointing to the current head reference"""

return HEAD(self, 'HEAD')

@property

def remotes(self) -> 'IterableList[Remote]':

"""A list of Remote objects allowing to access and manipulate remotes

return Remote.list_items(self)

def remote(self, name: str = 'origin') -> 'Remote':

""":return: Remote with the specified name

r = Remote(self, name)

if not r.exists():

raise ValueError("Remote named '%s' didn't exist" % name)

return r

#{ Submodules

@property

def submodules(self) -> 'IterableList[Submodule]':

"""

return Submodule.list_items(self)

def submodule(self, name: str) -> 'Submodule':

""" :return: Submodule with the given name

try:

return self.submodules[name]

except IndexError as e:

raise ValueError("Didn't find submodule named %r" % name) from e

# END exception handling

def create_submodule(self, *args: Any, **kwargs: Any) -> Submodule:

"""Create a new submodule

return Submodule.add(self, *args, **kwargs)

def iter_submodules(self, *args: Any, **kwargs: Any) -> Iterator[Submodule]:

"""An iterator yielding Submodule instances, see Traversable interface

return RootModule(self).traverse(*args, **kwargs)

def submodule_update(self, *args: Any, **kwargs: Any) -> Iterator[Submodule]:

"""Update the submodules, keeping the repository consistent as it will

return RootModule(self).update(*args, **kwargs)

#}END submodules

@property

def tags(self) -> 'IterableList[TagReference]':

"""A list of ``Tag`` objects that are available in this repo

return TagReference.list_items(self)

def tag(self, path: PathLike) -> TagReference:

""":return: TagReference Object, reference pointing to a Commit or Tag

full_path = self._to_full_tag_path(path)

return TagReference(self, full_path)

@staticmethod

def _to_full_tag_path(path: PathLike) -> str:

path_str = str(path)

if path_str.startswith(TagReference._common_path_default + '/'):

return path_str

if path_str.startswith(TagReference._common_default + '/'):

return Reference._common_path_default + '/' + path_str

else:

return TagReference._common_path_default + '/' + path_str

def create_head(self, path: PathLike, commit: str = 'HEAD',

force: bool = False, logmsg: Optional[str] = None

) -> 'Head':

"""Create a new head within the repository.

return Head.create(self, path, commit, logmsg, force)

def delete_head(self, *heads: 'Union[str, Head]', **kwargs: Any) -> None:

"""Delete the given heads

return Head.delete(self, *heads, **kwargs)

def create_tag(self, path: PathLike, ref: str = 'HEAD',

message: Optional[str] = None, force: bool = False, **kwargs: Any

) -> TagReference:

"""Create a new tag reference.

return TagReference.create(self, path, ref, message, force, **kwargs)

def delete_tag(self, *tags: TagReference) -> None:

"""Delete the given tag references"""

return TagReference.delete(self, *tags)

def create_remote(self, name: str, url: str, **kwargs: Any) -> Remote:

"""Create a new remote.

return Remote.create(self, name, url, **kwargs)

def delete_remote(self, remote: 'Remote') -> str:

"""Delete the given remote."""

return Remote.remove(self, remote)

def _get_config_path(self, config_level: Lit_config_levels) -> str:

# we do not support an absolute path of the gitconfig on windows ,

# use the global config instead

if is_win and config_level == "system":

config_level = "global"

if config_level == "system":

return "/etc/gitconfig"

elif config_level == "user":

config_home = os.environ.get("XDG_CONFIG_HOME") or osp.join(os.environ.get("HOME", '~'), ".config")

return osp.normpath(osp.expanduser(osp.join(config_home, "git", "config")))

elif config_level == "global":

return osp.normpath(osp.expanduser("~/.gitconfig"))

elif config_level == "repository":

repo_dir = self._common_dir or self.git_dir

if not repo_dir:

raise NotADirectoryError

else:

return osp.normpath(osp.join(repo_dir, "config"))

else:

assert_never(config_level, # type:ignore[unreachable]

ValueError(f"Invalid configuration level: {config_level!r}"))

def config_reader(self, config_level: Optional[Lit_config_levels] = None,

) -> GitConfigParser:

"""

files = None

if config_level is None:

files = [self._get_config_path(cast(Lit_config_levels, f))

for f in self.config_level if cast(Lit_config_levels, f)]

else:

files = [self._get_config_path(config_level)]

return GitConfigParser(files, read_only=True, repo=self)

def config_writer(self, config_level: Lit_config_levels = "repository"

) -> GitConfigParser:

"""

return GitConfigParser(self._get_config_path(config_level), read_only=False, repo=self)

def commit(self, rev: Union[str, Commit_ish, None] = None

) -> Commit:

"""The Commit object for the specified revision

if rev is None:

return self.head.commit

return self.rev_parse(str(rev) + "^0")

def iter_trees(self, *args: Any, **kwargs: Any) -> Iterator['Tree']:

""":return: Iterator yielding Tree objects

return (c.tree for c in self.iter_commits(*args, **kwargs))

def tree(self, rev: Union[Tree_ish, str, None] = None) -> 'Tree':

"""The Tree object for the given treeish revision

if rev is None:

return self.head.commit.tree

return self.rev_parse(str(rev) + "^{tree}")

def iter_commits(self, rev: Union[str, Commit, 'SymbolicReference', None] = None,

paths: Union[PathLike, Sequence[PathLike]] = '',

**kwargs: Any) -> Iterator[Commit]:

"""A list of Commit objects representing the history of a given ref/commit

if rev is None:

rev = self.head.commit

return Commit.iter_items(self, rev, paths, **kwargs)

def merge_base(self, *rev: TBD, **kwargs: Any

) -> List[Union[Commit_ish, None]]:

"""Find the closest common ancestor for the given revision (e.g. Commits, Tags, References, etc)

if len(rev) < 2:

raise ValueError("Please specify at least two revs, got only %i" % len(rev))

# end handle input

res: List[Union[Commit_ish, None]] = []

try:

lines = self.git.merge_base(*rev, **kwargs).splitlines() # List[str]

except GitCommandError as err:

if err.status == 128:

raise

# end handle invalid rev

# Status code 1 is returned if there is no merge-base

# (see https://github.com/git/git/blob/master/builtin/merge-base.c#L16)

return res

# end exception handling

for line in lines:

res.append(self.commit(line))

# end for each merge-base

return res

def is_ancestor(self, ancestor_rev: 'Commit', rev: 'Commit') -> bool:

"""Check if a commit is an ancestor of another

try:

self.git.merge_base(ancestor_rev, rev, is_ancestor=True)

except GitCommandError as err:

if err.status == 1:

return False

raise

return True

def is_valid_object(self, sha: str, object_type: Union[str, None] = None) -> bool:

try:

complete_sha = self.odb.partial_to_complete_sha_hex(sha)

object_info = self.odb.info(complete_sha)

if object_type:

if object_info.type == object_type.encode():

return True

else:

log.debug("Commit hash points to an object of type '%s'. Requested were objects of type '%s'",

object_info.type.decode(), object_type)

return False

else:

return True

except BadObject:

log.debug("Commit hash is invalid.")

return False

def _get_daemon_export(self) -> bool:

if self.git_dir:

filename = osp.join(self.git_dir, self.DAEMON_EXPORT_FILE)

return osp.exists(filename)

def _set_daemon_export(self, value: object) -> None:

if self.git_dir:

filename = osp.join(self.git_dir, self.DAEMON_EXPORT_FILE)

fileexists = osp.exists(filename)

if value and not fileexists:

touch(filename)

elif not value and fileexists:

os.unlink(filename)

daemon_export = property(_get_daemon_export, _set_daemon_export,

doc="If True, git-daemon may export this repository")

del _get_daemon_export

del _set_daemon_export

def _get_alternates(self) -> List[str]:

"""The list of alternates for this repo from which objects can be retrieved

if self.git_dir:

alternates_path = osp.join(self.git_dir, 'objects', 'info', 'alternates')

if osp.exists(alternates_path):

with open(alternates_path, 'rb') as f:

alts = f.read().decode(defenc)

return alts.strip().splitlines()

return []

def _set_alternates(self, alts: List[str]) -> None:

"""Sets the alternates

alternates_path = osp.join(self.common_dir, 'objects', 'info', 'alternates')

if not alts:

if osp.isfile(alternates_path):

os.remove(alternates_path)

else:

with open(alternates_path, 'wb') as f:

f.write("\n".join(alts).encode(defenc))

alternates = property(_get_alternates, _set_alternates,

doc="Retrieve a list of alternates paths or set a list paths to be used as alternates")

def is_dirty(self, index: bool = True, working_tree: bool = True, untracked_files: bool = False,

submodules: bool = True, path: Optional[PathLike] = None) -> bool:

"""

if self._bare:

# Bare repositories with no associated working directory are

# always consired to be clean.

return False

# start from the one which is fastest to evaluate

default_args = ['--abbrev=40', '--full-index', '--raw']

if not submodules:

default_args.append('--ignore-submodules')

if path:

default_args.extend(["--", str(path)])

if index:

# diff index against HEAD

if osp.isfile(self.index.path) and \

len(self.git.diff('--cached', *default_args)):

return True

# END index handling

if working_tree:

# diff index against working tree

if len(self.git.diff(*default_args)):

return True

# END working tree handling

if untracked_files:

if len(self._get_untracked_files(path, ignore_submodules=not submodules)):

return True

# END untracked files

return False

@property

def untracked_files(self) -> List[str]:

"""

return self._get_untracked_files()

def _get_untracked_files(self, *args: Any, **kwargs: Any) -> List[str]:

# make sure we get all files, not only untracked directories

proc = self.git.status(*args,

porcelain=True,

untracked_files=True,

as_process=True,

**kwargs)

# Untracked files preffix in porcelain mode

prefix = "?? "

untracked_files = []

for line in proc.stdout:

line = line.decode(defenc)

if not line.startswith(prefix):

continue

filename = line[len(prefix):].rstrip('\n')

# Special characters are escaped

if filename[0] == filename[-1] == '"':

filename = filename[1:-1]

# WHATEVER ... it's a mess, but works for me

filename = filename.encode('ascii').decode('unicode_escape').encode('latin1').decode(defenc)

untracked_files.append(filename)

finalize_process(proc)

return untracked_files

def ignored(self, *paths: PathLike) -> List[str]:

"""Checks if paths are ignored via .gitignore

try:

proc: str = self.git.check_ignore(*paths)

except GitCommandError:

return []

return proc.replace("\\\\", "\\").replace('"', "").split("\n")

@property

def active_branch(self) -> Head:

"""The name of the currently active branch.

# reveal_type(self.head.reference) # => Reference

return self.head.reference

def blame_incremental(self, rev: str | HEAD, file: str, **kwargs: Any) -> Iterator['BlameEntry']:

"""Iterator for blame information for the given file at the given revision.

data: bytes = self.git.blame(rev, '--', file, p=True, incremental=True, stdout_as_string=False, **kwargs)

commits: Dict[bytes, Commit] = {}

stream = (line for line in data.split(b'\n') if line)

while True:

try:

line = next(stream) # when exhausted, causes a StopIteration, terminating this function

except StopIteration:

return

split_line = line.split()

hexsha, orig_lineno_b, lineno_b, num_lines_b = split_line

lineno = int(lineno_b)

num_lines = int(num_lines_b)

orig_lineno = int(orig_lineno_b)

if hexsha not in commits:

# Now read the next few lines and build up a dict of properties

# for this commit

props: Dict[bytes, bytes] = {}

while True:

try:

line = next(stream)

except StopIteration:

return

if line == b'boundary':

# "boundary" indicates a root commit and occurs

# instead of the "previous" tag

continue

tag, value = line.split(b' ', 1)

props[tag] = value

if tag == b'filename':

# "filename" formally terminates the entry for --incremental

orig_filename = value

break

c = Commit(self, hex_to_bin(hexsha),

author=Actor(safe_decode(props[b'author']),

safe_decode(props[b'author-mail'].lstrip(b'<').rstrip(b'>'))),

authored_date=int(props[b'author-time']),

committer=Actor(safe_decode(props[b'committer']),

safe_decode(props[b'committer-mail'].lstrip(b'<').rstrip(b'>'))),

committed_date=int(props[b'committer-time']))

commits[hexsha] = c

else:

# Discard all lines until we find "filename" which is

# guaranteed to be the last line

while True:

try:

line = next(stream) # will fail if we reach the EOF unexpectedly

except StopIteration:

return

tag, value = line.split(b' ', 1)

if tag == b'filename':

orig_filename = value

break

yield BlameEntry(commits[hexsha],

range(lineno, lineno + num_lines),

safe_decode(orig_filename),

range(orig_lineno, orig_lineno + num_lines))

def blame(self, rev: Union[str, HEAD], file: str, incremental: bool = False, **kwargs: Any

) -> List[List[Commit | List[str | bytes] | None]] | Iterator[BlameEntry] | None:

"""The blame information for the given file at the given revision.

if incremental:

return self.blame_incremental(rev, file, **kwargs)

data: bytes = self.git.blame(rev, '--', file, p=True, stdout_as_string=False, **kwargs)

commits: Dict[str, Commit] = {}

blames: List[List[Commit | List[str | bytes] | None]] = []

class InfoTD(TypedDict, total=False):

sha: str

id: str

filename: str

summary: str

author: str

author_email: str

author_date: int

committer: str

committer_email: str

committer_date: int

info: InfoTD = {}

keepends = True

for line_bytes in data.splitlines(keepends):

try:

line_str = line_bytes.rstrip().decode(defenc)

except UnicodeDecodeError:

firstpart = ''

parts = []

is_binary = True

else:

# As we don't have an idea when the binary data ends, as it could contain multiple newlines

# in the process. So we rely on being able to decode to tell us what is is.

# This can absolutely fail even on text files, but even if it does, we should be fine treating it

# as binary instead

parts = self.re_whitespace.split(line_str, 1)

firstpart = parts[0]

is_binary = False

# end handle decode of line

if self.re_hexsha_only.search(firstpart):

# handles

# 634396b2f541a9f2d58b00be1a07f0c358b999b3 1 1 7 - indicates blame-data start

# 634396b2f541a9f2d58b00be1a07f0c358b999b3 2 2 - indicates

# another line of blame with the same data

digits = parts[-1].split(" ")

if len(digits) == 3:

info = {'id': firstpart}

blames.append([None, []])

elif info['id'] != firstpart:

info = {'id': firstpart}

blames.append([commits.get(firstpart), []])

# END blame data initialization

else:

m = self.re_author_committer_start.search(firstpart)

if m:

# handles:

# author Tom Preston-Werner

# author-mail <tom@mojombo.com>

# author-time 1192271832

# author-tz -0700

# committer Tom Preston-Werner

# committer-mail <tom@mojombo.com>

# committer-time 1192271832

# committer-tz -0700 - IGNORED BY US

role = m.group(0)

if role == 'author':

if firstpart.endswith('-mail'):

info["author_email"] = parts[-1]

elif firstpart.endswith('-time'):

info["author_date"] = int(parts[-1])

elif role == firstpart:

info["author"] = parts[-1]

elif role == 'committer':

if firstpart.endswith('-mail'):

info["committer_email"] = parts[-1]

elif firstpart.endswith('-time'):

info["committer_date"] = int(parts[-1])

elif role == firstpart:

info["committer"] = parts[-1]

# END distinguish mail,time,name

else:

# handle

# filename lib/grit.rb

# summary add Blob

# <and rest>

if firstpart.startswith('filename'):

info['filename'] = parts[-1]

elif firstpart.startswith('summary'):

info['summary'] = parts[-1]

elif firstpart == '':

if info:

sha = info['id']

c = commits.get(sha)

if c is None:

c = Commit(self, hex_to_bin(sha),

author=Actor._from_string(f"{info['author']} {info['author_email']}"),

authored_date=info['author_date'],

committer=Actor._from_string(

f"{info['committer']} {info['committer_email']}"),

committed_date=info['committer_date'])

commits[sha] = c

blames[-1][0] = c

# END if commit objects needs initial creation

if blames[-1][1] is not None:

line: str | bytes

if not is_binary:

if line_str and line_str[0] == '\t':

line_str = line_str[1:]

line = line_str

else:

line = line_bytes

# NOTE: We are actually parsing lines out of binary data, which can lead to the

# binary being split up along the newline separator. We will append this to the

# blame we are currently looking at, even though it should be concatenated with

# the last line we have seen.