"""

__slots__ = ("repo", "version", "entries", "_extension_data", "_file_path")

_VERSION = 2 # latest version we support

S_IFGITLINK = S_IFGITLINK # a submodule

def __init__(self, repo: 'Repo', file_path: Union[PathLike, None] = None) -> None:

"""Initialize this Index instance, optionally from the given ``file_path``.

self.repo = repo

self.version = self._VERSION

self._extension_data = b''

self._file_path: PathLike = file_path or self._index_path()

def _set_cache_(self, attr: str) -> None:

if attr == "entries":

try:

fd = os.open(self._file_path, os.O_RDONLY)

except OSError:

# in new repositories, there may be no index, which means we are empty

self.entries: Dict[Tuple[PathLike, StageType], IndexEntry] = {}

return None

# END exception handling

try:

stream = file_contents_ro(fd, stream=True, allow_mmap=True)

finally:

os.close(fd)

self._deserialize(stream)

else:

super(IndexFile, self)._set_cache_(attr)

def _index_path(self) -> PathLike:

if self.repo.git_dir:

return join_path_native(self.repo.git_dir, "index")

else:

raise GitCommandError("No git directory given to join index path")

@property

def path(self) -> PathLike:

""" :return: Path to the index file we are representing """

return self._file_path

def _delete_entries_cache(self) -> None:

"""Safely clear the entries cache so it can be recreated"""

try:

del(self.entries)

except AttributeError:

# fails in python 2.6.5 with this exception

pass

# END exception handling

#{ Serializable Interface

def _deserialize(self, stream: IO) -> 'IndexFile':

"""Initialize this instance with index values read from the given stream"""

self.version, self.entries, self._extension_data, _conten_sha = read_cache(stream)

return self

def _entries_sorted(self) -> List[IndexEntry]:

""":return: list of entries, in a sorted fashion, first by path, then by stage"""

return sorted(self.entries.values(), key=lambda e: (e.path, e.stage))

def _serialize(self, stream: IO, ignore_extension_data: bool = False) -> 'IndexFile':

entries = self._entries_sorted()

extension_data = self._extension_data # type: Union[None, bytes]

if ignore_extension_data:

extension_data = None

write_cache(entries, stream, extension_data)

return self

#} END serializable interface

def write(self, file_path: Union[None, PathLike] = None, ignore_extension_data: bool = False) -> None:

"""Write the current state to our file path or to the given one

# make sure we have our entries read before getting a write lock

# else it would be done when streaming. This can happen

# if one doesn't change the index, but writes it right away

self.entries

lfd = LockedFD(file_path or self._file_path)

stream = lfd.open(write=True, stream=True)

ok = False

try:

self._serialize(stream, ignore_extension_data)

ok = True

finally:

if not ok:

lfd.rollback()

lfd.commit()

# make sure we represent what we have written

if file_path is not None:

self._file_path = file_path

@post_clear_cache

@default_index

def merge_tree(self, rhs: Treeish, base: Union[None, Treeish] = None) -> 'IndexFile':

"""Merge the given rhs treeish into the current index, possibly taking

# -i : ignore working tree status

# --aggressive : handle more merge cases

# -m : do an actual merge

args: List[Union[Treeish, str]] = ["--aggressive", "-i", "-m"]

if base is not None:

args.append(base)

args.append(rhs)

self.repo.git.read_tree(args)

return self

@classmethod

def new(cls, repo: 'Repo', *tree_sha: Union[str, Tree]) -> 'IndexFile':

""" Merge the given treeish revisions into a new index which is returned.

tree_sha_bytes: List[bytes] = [to_bin_sha(str(t)) for t in tree_sha]

base_entries = aggressive_tree_merge(repo.odb, tree_sha_bytes)

inst = cls(repo)

# convert to entries dict

entries: Dict[Tuple[PathLike, int], IndexEntry] = dict(zip(

((e.path, e.stage) for e in base_entries),

(IndexEntry.from_base(e) for e in base_entries)))

inst.entries = entries

return inst

@classmethod

def from_tree(cls, repo: 'Repo', *treeish: Treeish, **kwargs: Any) -> 'IndexFile':

"""Merge the given treeish revisions into a new index which is returned.

if len(treeish) == 0 or len(treeish) > 3:

raise ValueError("Please specify between 1 and 3 treeish, got %i" % len(treeish))

arg_list: List[Union[Treeish, str]] = []

# ignore that working tree and index possibly are out of date

if len(treeish) > 1:

# drop unmerged entries when reading our index and merging

arg_list.append("--reset")

# handle non-trivial cases the way a real merge does

arg_list.append("--aggressive")

# END merge handling

# tmp file created in git home directory to be sure renaming

# works - /tmp/ dirs could be on another device

tmp_index = tempfile.mktemp('', '', repo.git_dir)

arg_list.append("--index-output=%s" % tmp_index)

arg_list.extend(treeish)

# move current index out of the way - otherwise the merge may fail

# as it considers existing entries. moving it essentially clears the index.

# Unfortunately there is no 'soft' way to do it.

# The TemporaryFileSwap assure the original file get put back

if repo.git_dir:

index_handler = TemporaryFileSwap(join_path_native(repo.git_dir, 'index'))

try:

repo.git.read_tree(*arg_list, **kwargs)

index = cls(repo, tmp_index)

index.entries # force it to read the file as we will delete the temp-file

del(index_handler) # release as soon as possible

finally:

if osp.exists(tmp_index):

os.remove(tmp_index)

# END index merge handling

return index

# UTILITIES

@unbare_repo

def _iter_expand_paths(self: 'IndexFile', paths: Sequence[PathLike]) -> Iterator[PathLike]:

"""Expand the directories in list of paths to the corresponding paths accordingly,

def raise_exc(e: Exception) -> NoReturn:

raise e

r = str(self.repo.working_tree_dir)

rs = r + os.sep

for path in paths:

abs_path = str(path)

if not osp.isabs(abs_path):

abs_path = osp.join(r, path)

# END make absolute path

try:

st = os.lstat(abs_path) # handles non-symlinks as well

except OSError:

# the lstat call may fail as the path may contain globs as well

pass

else:

if S_ISLNK(st.st_mode):

yield abs_path.replace(rs, '')

continue

# end check symlink

# if the path is not already pointing to an existing file, resolve globs if possible

if not os.path.exists(abs_path) and ('?' in abs_path or '*' in abs_path or '[' in abs_path):

resolved_paths = glob.glob(abs_path)

# not abs_path in resolved_paths:

# a glob() resolving to the same path we are feeding it with

# is a glob() that failed to resolve. If we continued calling

# ourselves we'd endlessly recurse. If the condition below

# evaluates to true then we are likely dealing with a file

# whose name contains wildcard characters.

if abs_path not in resolved_paths:

for f in self._iter_expand_paths(glob.glob(abs_path)):

yield str(f).replace(rs, '')

continue

# END glob handling

try:

for root, _dirs, files in os.walk(abs_path, onerror=raise_exc):

for rela_file in files:

# add relative paths only

yield osp.join(root.replace(rs, ''), rela_file)

# END for each file in subdir

# END for each subdirectory

except OSError:

# was a file or something that could not be iterated

yield abs_path.replace(rs, '')

# END path exception handling

# END for each path

def _write_path_to_stdin(self, proc: 'Popen', filepath: PathLike, item: PathLike, fmakeexc: Callable[..., GitError],

fprogress: Callable[[PathLike, bool, PathLike], None],

read_from_stdout: bool = True) -> Union[None, str]:

"""Write path to proc.stdin and make sure it processes the item, including progress.

fprogress(filepath, False, item)

rval: Union[None, str] = None

if proc.stdin is not None:

try:

proc.stdin.write(("%s\n" % filepath).encode(defenc))

except IOError as e:

# pipe broke, usually because some error happened

raise fmakeexc() from e

# END write exception handling

proc.stdin.flush()

if read_from_stdout and proc.stdout is not None:

rval = proc.stdout.readline().strip()

fprogress(filepath, True, item)

return rval

def iter_blobs(self, predicate: Callable[[Tuple[StageType, Blob]], bool] = lambda t: True

) -> Iterator[Tuple[StageType, Blob]]:

"""

for entry in self.entries.values():

blob = entry.to_blob(self.repo)

blob.size = entry.size

output = (entry.stage, blob)

if predicate(output):

yield output

# END for each entry

def unmerged_blobs(self) -> Dict[PathLike, List[Tuple[StageType, Blob]]]:

"""

is_unmerged_blob = lambda t: t[0] != 0

path_map: Dict[PathLike, List[Tuple[StageType, Blob]]] = {}

for stage, blob in self.iter_blobs(is_unmerged_blob):

path_map.setdefault(blob.path, []).append((stage, blob))

# END for each unmerged blob

for line in path_map.values():

line.sort()

return path_map

@ classmethod

def entry_key(cls, *entry: Union[BaseIndexEntry, PathLike, StageType]) -> Tuple[PathLike, StageType]:

return entry_key(*entry)

def resolve_blobs(self, iter_blobs: Iterator[Blob]) -> 'IndexFile':

"""Resolve the blobs given in blob iterator. This will effectively remove the

for blob in iter_blobs:

stage_null_key = (blob.path, 0)

if stage_null_key in self.entries:

raise ValueError("Path %r already exists at stage 0" % str(blob.path))

# END assert blob is not stage 0 already

# delete all possible stages

for stage in (1, 2, 3):

try:

del(self.entries[(blob.path, stage)])

except KeyError:

pass

# END ignore key errors

# END for each possible stage

self.entries[stage_null_key] = IndexEntry.from_blob(blob)

# END for each blob

return self

def update(self) -> 'IndexFile':

"""Reread the contents of our index file, discarding all cached information

self._delete_entries_cache()

# allows to lazily reread on demand

return self

def write_tree(self) -> Tree:

"""Writes this index to a corresponding Tree object into the repository's

# we obtain no lock as we just flush our contents to disk as tree

# If we are a new index, the entries access will load our data accordingly

mdb = MemoryDB()

entries = self._entries_sorted()

binsha, tree_items = write_tree_from_cache(entries, mdb, slice(0, len(entries)))

# copy changed trees only

mdb.stream_copy(mdb.sha_iter(), self.repo.odb)

# note: additional deserialization could be saved if write_tree_from_cache

# would return sorted tree entries

root_tree = Tree(self.repo, binsha, path='')

root_tree._cache = tree_items

return root_tree

def _process_diff_args(self, # type: ignore[override]

args: List[Union[str, 'git_diff.Diffable', Type['git_diff.Diffable.Index']]]

) -> List[Union[str, 'git_diff.Diffable', Type['git_diff.Diffable.Index']]]:

try:

args.pop(args.index(self))

except IndexError:

pass

# END remove self

return args

def _to_relative_path(self, path: PathLike) -> PathLike:

"""

if not osp.isabs(path):

return path

if self.repo.bare:

raise InvalidGitRepositoryError("require non-bare repository")

if not str(path).startswith(str(self.repo.working_tree_dir)):

raise ValueError("Absolute path %r is not in git repository at %r" % (path, self.repo.working_tree_dir))

return os.path.relpath(path, self.repo.working_tree_dir)

def _preprocess_add_items(self, items: Sequence[Union[PathLike, Blob, BaseIndexEntry, 'Submodule']]

) -> Tuple[List[PathLike], List[BaseIndexEntry]]:

""" Split the items into two lists of path strings and BaseEntries. """

paths = []

entries = []

# if it is a string put in list

if isinstance(items, str):

items = [items]

for item in items:

if isinstance(item, str):

paths.append(self._to_relative_path(item))

elif isinstance(item, (Blob, Submodule)):

entries.append(BaseIndexEntry.from_blob(item))

elif isinstance(item, BaseIndexEntry):

entries.append(item)

else:

raise TypeError("Invalid Type: %r" % item)

# END for each item

return paths, entries

def _store_path(self, filepath: PathLike, fprogress: Callable) -> BaseIndexEntry:

"""Store file at filepath in the database and return the base index entry

st = os.lstat(filepath) # handles non-symlinks as well

if S_ISLNK(st.st_mode):

# in PY3, readlink is string, but we need bytes. In PY2, it's just OS encoded bytes, we assume UTF-8

open_stream: Callable[[], BinaryIO] = lambda: BytesIO(force_bytes(os.readlink(filepath),

encoding=defenc))

else:

open_stream = lambda: open(filepath, 'rb')

with open_stream() as stream:

fprogress(filepath, False, filepath)

istream = self.repo.odb.store(IStream(Blob.type, st.st_size, stream))

fprogress(filepath, True, filepath)

return BaseIndexEntry((stat_mode_to_index_mode(st.st_mode),

istream.binsha, 0, to_native_path_linux(filepath)))

@ unbare_repo

@ git_working_dir

def _entries_for_paths(self, paths: List[str], path_rewriter: Callable, fprogress: Callable,

entries: List[BaseIndexEntry]) -> List[BaseIndexEntry]:

entries_added: List[BaseIndexEntry] = []

if path_rewriter:

for path in paths:

if osp.isabs(path):

abspath = path

gitrelative_path = path[len(str(self.repo.working_tree_dir)) + 1:]

else:

gitrelative_path = path

if self.repo.working_tree_dir:

abspath = osp.join(self.repo.working_tree_dir, gitrelative_path)

# end obtain relative and absolute paths

blob = Blob(self.repo, Blob.NULL_BIN_SHA,

stat_mode_to_index_mode(os.stat(abspath).st_mode),

to_native_path_linux(gitrelative_path))

# TODO: variable undefined

entries.append(BaseIndexEntry.from_blob(blob))

# END for each path

del(paths[:])

# END rewrite paths

# HANDLE PATHS

assert len(entries_added) == 0

for filepath in self._iter_expand_paths(paths):

entries_added.append(self._store_path(filepath, fprogress))

# END for each filepath

# END path handling

return entries_added

def add(self, items: Sequence[Union[PathLike, Blob, BaseIndexEntry, 'Submodule']], force: bool = True,

fprogress: Callable = lambda *args: None, path_rewriter: Union[Callable[..., PathLike], None] = None,

write: bool = True, write_extension_data: bool = False) -> List[BaseIndexEntry]:

"""Add files from the working tree, specific blobs or BaseIndexEntries

# sort the entries into strings and Entries, Blobs are converted to entries

# automatically

# paths can be git-added, for everything else we use git-update-index

paths, entries = self._preprocess_add_items(items)

entries_added: List[BaseIndexEntry] = []

# This code needs a working tree, therefore we try not to run it unless required.

# That way, we are OK on a bare repository as well.

# If there are no paths, the rewriter has nothing to do either

if paths:

entries_added.extend(self._entries_for_paths(paths, path_rewriter, fprogress, entries))

# HANDLE ENTRIES

if entries:

null_mode_entries = [e for e in entries if e.mode == 0]

if null_mode_entries:

raise ValueError(

"At least one Entry has a null-mode - please use index.remove to remove files for clarity")

# END null mode should be remove

# HANLDE ENTRY OBJECT CREATION

# create objects if required, otherwise go with the existing shas

null_entries_indices = [i for i, e in enumerate(entries) if e.binsha == Object.NULL_BIN_SHA]

if null_entries_indices:

@ git_working_dir

def handle_null_entries(self: 'IndexFile') -> None:

for ei in null_entries_indices:

null_entry = entries[ei]

new_entry = self._store_path(null_entry.path, fprogress)

# update null entry

entries[ei] = BaseIndexEntry(

(null_entry.mode, new_entry.binsha, null_entry.stage, null_entry.path))

# END for each entry index

# end closure

handle_null_entries(self)

# END null_entry handling

# REWRITE PATHS

# If we have to rewrite the entries, do so now, after we have generated

# all object sha's

if path_rewriter:

for i, e in enumerate(entries):

entries[i] = BaseIndexEntry((e.mode, e.binsha, e.stage, path_rewriter(e)))

# END for each entry

# END handle path rewriting

# just go through the remaining entries and provide progress info

for i, entry in enumerate(entries):

progress_sent = i in null_entries_indices

if not progress_sent:

fprogress(entry.path, False, entry)

fprogress(entry.path, True, entry)

# END handle progress

# END for each enty

entries_added.extend(entries)

# END if there are base entries

# FINALIZE

# add the new entries to this instance

for entry in entries_added:

self.entries[(entry.path, 0)] = IndexEntry.from_base(entry)

if write:

self.write(ignore_extension_data=not write_extension_data)

# END handle write

return entries_added

def _items_to_rela_paths(self, items: Union[PathLike, Sequence[Union[PathLike, BaseIndexEntry, Blob, Submodule]]]

) -> List[PathLike]:

"""Returns a list of repo-relative paths from the given items which

paths = []

# if string put in list

if isinstance(items, (str, os.PathLike)):

items = [items]

for item in items:

if isinstance(item, (BaseIndexEntry, (Blob, Submodule))):

paths.append(self._to_relative_path(item.path))

elif isinstance(item, str):

paths.append(self._to_relative_path(item))

else:

raise TypeError("Invalid item type: %r" % item)

# END for each item

return paths

@ post_clear_cache

@ default_index

def remove(self, items: Sequence[Union[PathLike, Blob, BaseIndexEntry, 'Submodule']], working_tree: bool = False,

**kwargs: Any) -> List[str]:

"""Remove the given items from the index and optionally from

args = []

if not working_tree:

args.append("--cached")

args.append("--")

# preprocess paths

paths = self._items_to_rela_paths(items)

removed_paths = self.repo.git.rm(args, paths, **kwargs).splitlines()

# process output to gain proper paths

# rm 'path'

return [p[4:-1] for p in removed_paths]

@ post_clear_cache

@ default_index

def move(self, items: Sequence[Union[PathLike, Blob, BaseIndexEntry, 'Submodule']], skip_errors: bool = False,

**kwargs: Any) -> List[Tuple[str, str]]:

"""Rename/move the items, whereas the last item is considered the destination of

args = []

if skip_errors:

args.append('-k')

paths = self._items_to_rela_paths(items)

if len(paths) < 2:

raise ValueError("Please provide at least one source and one destination of the move operation")

was_dry_run = kwargs.pop('dry_run', kwargs.pop('n', None))

kwargs['dry_run'] = True

# first execute rename in dryrun so the command tells us what it actually does

# ( for later output )

out = []

mvlines = self.repo.git.mv(args, paths, **kwargs).splitlines()

# parse result - first 0:n/2 lines are 'checking ', the remaining ones

# are the 'renaming' ones which we parse

for ln in range(int(len(mvlines) / 2), len(mvlines)):

tokens = mvlines[ln].split(' to ')

assert len(tokens) == 2, "Too many tokens in %s" % mvlines[ln]

# [0] = Renaming x

# [1] = y

out.append((tokens[0][9:], tokens[1]))

# END for each line to parse

# either prepare for the real run, or output the dry-run result

if was_dry_run:

return out

# END handle dryrun

# now apply the actual operation

kwargs.pop('dry_run')

self.repo.git.mv(args, paths, **kwargs)

return out

def commit(self,

message: str,

parent_commits: Union[Commit_ish, None] = None,

head: bool = True,

author: Union[None, 'Actor'] = None,

committer: Union[None, 'Actor'] = None,

author_date: Union[str, None] = None,

commit_date: Union[str, None] = None,

skip_hooks: bool = False) -> Commit:

"""Commit the current default index file, creating a commit object.

if not skip_hooks:

run_commit_hook('pre-commit', self)

self._write_commit_editmsg(message)

run_commit_hook('commit-msg', self, self._commit_editmsg_filepath())

message = self._read_commit_editmsg()

self._remove_commit_editmsg()

tree = self.write_tree()

rval = Commit.create_from_tree(self.repo, tree, message, parent_commits,

head, author=author, committer=committer,

author_date=author_date, commit_date=commit_date)

if not skip_hooks:

run_commit_hook('post-commit', self)

return rval

def _write_commit_editmsg(self, message: str) -> None:

with open(self._commit_editmsg_filepath(), "wb") as commit_editmsg_file: