__all__ = ["IndexFile", "CheckoutError", "StageType"]

import contextlib

import datetime

import glob

from io import BytesIO

import os

import os.path as osp

from stat import S_ISLNK, S_ISREG

import subprocess

import shutil

import tempfile

from types import MappingProxyType

from gitdb.base import IStream

from git import _backend

from git.compat import defenc, safe_decode

from git.cmd import Git

import git.diff as git_diff

from git.exc import CheckoutError, GitCommandError, GitError, InvalidGitRepositoryError, UnmergedEntriesError

from git.objects import Blob, Commit, Object, Submodule, Tree

from git.util import (

Actor,

LazyMixin,

LockedFD,

join_path_native,

_is_path_rooted,

_to_relative_path,

_validate_repo_path,

to_native_path_linux,

unbare_repo,

)

from .fun import (

S_IFGITLINK,

entry_key,

run_commit_hook,

stat_mode_to_index_mode,

)

from .typ import BaseIndexEntry, IndexEntry, StageType, CE_STAGESHIFT, CE_VALID, CE_EXT_SKIP_WORKTREE

from .util import default_index, git_working_dir

from typing import (

Any,

BinaryIO,

Callable,

cast,

Dict,

Generator,

Iterable,

Iterator,

List,

Mapping,

NoReturn,

Sequence,

TYPE_CHECKING,

Tuple,

Union,

)

from git.types import Literal, PathLike

if TYPE_CHECKING:

from subprocess import Popen

from git.refs.reference import Reference

from git.repo import Repo

Treeish = Union[Tree, Commit, str, bytes, Sequence[Union[Tree, Commit, str, bytes]]]

class IndexFile(LazyMixin, git_diff.Diffable):

"""An index whose storage and merge operations are implemented by Git.

unsafe_git_checkout_index_options = ["--prefix"]

unsafe_git_read_tree_options = ["--index-output"]

__slots__ = ("repo", "_file_path", "_data", "_split_index")

S_IFGITLINK = S_IFGITLINK

"""Flags for 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._data: Union[bytes, None] = None

self._split_index = False

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

def _read_entries(self, path: PathLike) -> Iterator[IndexEntry]:

output = self.repo.git._call_process_safe(

"ls_files",

"--stage",

"-v",

"-z",

"--full-name",

env={"GIT_INDEX_FILE": osp.abspath(path)},

stdout_as_string=False,

)

for record in output.split(b"\0"):

if not record:

continue

metadata, path_bytes = record.split(b"\t", 1)

tag, mode, oid, stage = metadata.split()

entry_path = safe_decode(path_bytes)

binsha = bytes.fromhex(oid.decode("ascii"))

if len(binsha) != self.repo._oid_size or int(stage) not in range(4):

raise ValueError("Invalid index entry returned by Git")

flags = int(stage) << CE_STAGESHIFT

if tag.islower():

flags |= CE_VALID

if tag.upper() == b"S":

flags |= CE_EXT_SKIP_WORKTREE << 16

entry = IndexEntry((int(mode, 8), binsha, flags, entry_path))

yield entry

def iter_entries(self) -> Iterator[IndexEntry]:

"""Yield immutable entries from the current on-disk or pending index."""

if self._data is None:

yield from self._read_entries(self.path)

else:

with self._materialized_index() as path:

yield from self._read_entries(path)

@property

def entries(self) -> Mapping[Tuple[PathLike, StageType], IndexEntry]:

"""Return an immutable snapshot keyed by path and merge stage.

return MappingProxyType({entry_key(entry): entry for entry in self.iter_entries()})

def entry(self, path: PathLike, stage: StageType = 0) -> IndexEntry:

"""Return the exact path/stage entry; raise ``KeyError`` if absent."""

for entry in self.iter_entries():

if entry.path == os.fspath(path) and entry.stage == stage:

return entry

raise KeyError((path, stage))

@property

def version(self) -> int:

"""The index version reported by Git, or 2 for an unwritten index."""

if self._data is None and not osp.exists(self.path):

return 2

if self._data is None:

return int(

self.repo.git._call_process_safe(

"update_index", "--show-index-version", env={"GIT_INDEX_FILE": osp.abspath(self.path)}

)

)

with self._materialized_index() as path:

return int(

self.repo.git._call_process_safe("update_index", "--show-index-version", env={"GIT_INDEX_FILE": path})

)

@contextlib.contextmanager

def _materialized_index(self) -> Generator[str, None, None]:

"""Give Git a private file, leaving published and pending state untouched."""

with tempfile.TemporaryDirectory(prefix="gitpython-index-") as directory:

path = osp.join(directory, "index")

if self._data is not None:

with open(path, "wb") as stream:

stream.write(self._data)

elif osp.isfile(self.path):

shutil.copyfile(self.path, path)

else:

self.repo.git._call_process_safe("read_tree", "--empty", env={"GIT_INDEX_FILE": path})

yield path

def _accept_index(self, path: str, write: bool) -> None:

env = {"GIT_INDEX_FILE": path}

split = self._split_index or bool(self.repo.git._call_process_safe("rev_parse", "--shared-index-path", env=env))

if split:

# Pending bytes must not depend on a shared index that Git may expire.

self.repo.git._call_process_safe("update_index", "--no-split-index", env=env)

with open(path, "rb") as stream:

data = stream.read()

if write:

if split:

self.repo.git._call_process_safe("update_index", "--split-index", env=env)

lock = LockedFD(self.path)

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

try:

assert not isinstance(stream, int)

with open(path, "rb") as source:

shutil.copyfileobj(source, stream)

lock.commit()

except BaseException:

lock.rollback()

raise

self._data = None

self._split_index = False

else:

self._data = data

self._split_index = split

def _update_entries(

self, entries: Sequence[BaseIndexEntry], clear: Sequence[PathLike] = (), *, write: bool = False

) -> None:

"""Send a batch to Git and reject successful-but-ignored updates."""

records = []

for name in clear:

if "\0" in os.fspath(name):

raise ValueError("Invalid index path")

records.append(b"0 " + self.repo._null_hexsha.encode("ascii") + b"\t" + os.fsencode(name) + b"\0")

for entry in entries:

if "\0" in os.fspath(entry.path) or len(entry.binsha) != self.repo._oid_size:

raise ValueError("Invalid index path or object ID")

records.append(

("%o %s %d\t" % (entry.mode, entry.hexsha, entry.stage)).encode("ascii")

+ os.fsencode(entry.path)

+ b"\0"

)

with self._materialized_index() as path:

env = {"GIT_INDEX_FILE": path}

with tempfile.TemporaryFile() as stream:

stream.write(b"".join(records))

stream.seek(0)

self.repo.git._call_process_safe(

"update_index",

"-z",

"--index-info",

istream=stream,

env=env,

_config=["core.protectHFS=true", "core.protectNTFS=true"],

)

actual = {(entry.path, entry.stage): entry for entry in self._read_entries(path)}

requested = {(entry.path, entry.stage): entry for entry in entries}

if any(key not in actual or actual[key][:2] != entry[:2] for key, entry in requested.items()) or any(

(name, stage) in actual and (name, stage) not in requested for name in clear for stage in range(4)

):

raise ValueError("Git did not retain the requested index entries; filenames may be unsupported")

for option, mask in (("--assume-unchanged", CE_VALID), ("--skip-worktree", CE_EXT_SKIP_WORKTREE << 16)):

names = [

os.fsencode(entry.path) + b"\0" for entry in entries if entry.stage == 0 and entry.flags & mask

]

if names:

with tempfile.TemporaryFile() as stream:

stream.write(b"".join(names))

stream.seek(0)

self.repo.git._call_process_safe(

"update_index", option, "-z", "--stdin", istream=stream, env=env

)

self._accept_index(path, write)

def _tree_oid(self, tree: Treeish) -> str:

# The documented merge workflow passes the singleton result of merge_base().

if isinstance(tree, (list, tuple)):

if len(tree) != 1:

raise ValueError("A tree revision sequence must contain exactly one revision")

tree = tree[0]

if isinstance(tree, bytes):

tree = tree.hex() if len(tree) == self.repo._oid_size else tree.decode("ascii")

value = str(tree)

Git._check_operand(value, "tree revision")

return self.repo.git._call_process_safe("rev_parse", "--verify", "--end-of-options", value + "^{tree}")

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 write(self, file_path: Union[None, PathLike] = None, ignore_extension_data: bool = False) -> None:

"""Publish the current index atomically; Git owns extension data.

original_path = self.path

with self._materialized_index() as path:

self._file_path = file_path or original_path

try:

self._accept_index(path, True)

except BaseException:

self._file_path = original_path

raise

@default_index

def merge_tree(

self,

rhs: Treeish,

base: Union[None, Treeish] = None,

allow_unsafe_options: bool = False,

) -> "IndexFile":

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

if not allow_unsafe_options:

Git.check_unsafe_options(

options=Git._option_candidates([base, rhs]),

unsafe_options=self.unsafe_git_read_tree_options,

)

# -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(self._tree_oid(base))

args.append(self._tree_oid(rhs))

with self._materialized_index() as path:

self.repo.git._call_process_safe("read_tree", args, env={"GIT_INDEX_FILE": path})

self._accept_index(path, True)

return self

@classmethod

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

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

if not 1 <= len(tree_sha) <= 3:

raise ValueError("Specify between one and three trees")

return cls.from_tree(repo, *(tree_sha[-1:] if len(tree_sha) < 3 else tree_sha))

@classmethod

def from_tree(

cls,

repo: "Repo",

*treeish: Treeish,

allow_unsafe_options: bool = False,

**kwargs: Any,

) -> "IndexFile":

R"""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))

if not allow_unsafe_options:

Git.check_unsafe_options(

options=Git._option_candidates(treeish, kwargs),

unsafe_options=cls.unsafe_git_read_tree_options,

)

if set(kwargs) - {"trivial", "aggressive", "verbose"}:

raise ValueError("Unsupported read-tree options")

with tempfile.TemporaryDirectory(prefix="gitpython-merge-") as directory:

path = osp.join(directory, "index")

index = cls(repo, path)

args = [index._tree_oid(tree) for tree in treeish]

if len(args) > 1:

args[:0] = ["--reset", "--aggressive", "-i"]

repo.git._call_process_safe("read_tree", *args, env={"GIT_INDEX_FILE": path}, **kwargs)

index._accept_index(path, False)

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

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 = os.fspath(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 dirname in dirs[:]:

directory = osp.join(root, dirname)

try:

_validate_repo_path(to_native_path_linux(osp.relpath(directory, r)))

except ValueError:

dirs.remove(dirname)

continue

if osp.islink(directory):

dirs.remove(dirname)

yield osp.relpath(directory, r)

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:

# It 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

fprogress(filepath, False, item)

rval: Union[None, str] = None

if proc.stdin is not None:

try:

proc.stdin.write(os.fspath(filepath).encode(defenc, "surrogateescape") + b"\0")

except OSError 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.iter_entries():

blob = entry.to_blob(self.repo)

output = (entry.stage, blob)

if predicate(output):

yield output

# END for each entry

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

"""

def is_unmerged_blob(t: Tuple[StageType, Blob]) -> bool:

return 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.

entries = []

names = set()

for blob in iter_blobs:

if blob.path in names or any(e.path == blob.path and e.stage == 0 for e in self.iter_entries()):

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

names.add(blob.path)

entries.append(BaseIndexEntry.from_blob(blob))

self._update_entries(entries, list(names))

return self

def update(self) -> "IndexFile":

"""Discard pending edits and query the published file on subsequent reads."""

self._data = None

self._split_index = False

return self

def write_tree(self) -> Tree:

"""Write this index to a corresponding :class:`~git.objects.tree.Tree` object

entries = list(self.iter_entries())

for entry in entries:

_validate_repo_path(entry.path, entry.mode)

if entry.stage:

raise UnmergedEntriesError(entry)

native = _backend.write_tree(

self.repo.git, [(os.fsencode(entry.path), entry.mode, entry.hexsha) for entry in entries]

)

if native is not NotImplemented:

return Tree(self.repo, bytes.fromhex(native), path="")

with self._materialized_index() as path:

oid = self.repo.git._call_process_safe("write_tree", "--missing-ok", env={"GIT_INDEX_FILE": path})

return Tree(self.repo, bytes.fromhex(oid), path="")

def _process_diff_args(

self,

args: List[Union[PathLike, "git_diff.Diffable"]],

) -> List[Union[PathLike, "git_diff.Diffable"]]:

try:

args.pop(args.index(self))

except IndexError:

pass

# END remove self

return args

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

"""

if self.repo.bare:

drive, _tail = osp.splitdrive(os.fspath(path))

if drive or _is_path_rooted(path):

raise InvalidGitRepositoryError("paths with a drive or root require a non-bare repository")

return path

return _to_relative_path(cast(PathLike, self.repo.working_tree_dir), path)

def _preprocess_add_items(

self, items: Union[PathLike, 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, os.PathLike)):

items = [items]

for item in items:

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

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

# Source paths must be safe to read, but their recorded names may be rewritten.

# Recorded destinations are validated by Git after rewriting.

for entry in entries:

_validate_repo_path(entry.path)

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.

filepath = self._to_relative_path(filepath)

_validate_repo_path(filepath)

parent = osp.realpath(self.repo.working_dir)

for component in os.fspath(filepath).split("/")[:-1]:

parent = osp.join(parent, component)

if osp.islink(parent) or osp.normcase(osp.realpath(parent)) != osp.normcase(osp.abspath(parent)):

raise ValueError("Cannot stage a path beyond a symbolic link: %r" % filepath)

st = os.lstat(filepath)

if not S_ISLNK(st.st_mode) and not S_ISREG(st.st_mode):

raise ValueError("Can only stage a regular file or symbolic link: %r" % filepath)

stream_size = st.st_size

if S_ISLNK(st.st_mode):

target = os.fsencode(os.readlink(filepath))

stream_size = len(target)

def open_stream() -> BinaryIO:

return BytesIO(target)

else:

def open_stream() -> BinaryIO:

# Do not follow a final symlink or block on a FIFO substituted

# between lstat and open on platforms supporting these flags.

def opener(path: str, flags: int) -> int:

return os.open(path, flags | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_NONBLOCK", 0))

return open(filepath, "rb", opener=opener)

with open_stream() as stream:

if not S_ISLNK(st.st_mode):

st = os.fstat(stream.fileno())

if not S_ISREG(st.st_mode):

raise ValueError("Can only stage a regular file: %r" % filepath)

stream_size = st.st_size

fprogress(filepath, False, filepath)

istream = self.repo.odb.store(IStream(Blob.type, stream_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: Union[Callable, None],

fprogress: Callable,

entries: List[BaseIndexEntry],

) -> List[BaseIndexEntry]:

entries_added: List[BaseIndexEntry] = []

if path_rewriter:

working_tree_dir = self.repo.working_tree_dir

if working_tree_dir is None:

raise InvalidGitRepositoryError("Cannot rewrite paths without a working tree")

working_tree_dir = str(working_tree_dir)

for path in paths:

if osp.isabs(path):

abspath = path

gitrelative_path = path[len(working_tree_dir) + 1 :]

else:

gitrelative_path = path

abspath = osp.join(working_tree_dir, gitrelative_path)

# END obtain relative and absolute paths

blob = Blob(

self.repo,

self.repo._null_binsha,

stat_mode_to_index_mode(os.lstat(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: Union[PathLike, 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]:

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

# 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, so 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

# HANDLE ENTRY OBJECT CREATION

# Create objects if required, otherwise go with the existing shas.

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

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.flags,

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.flags, path_rewriter(e)))