1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
"""Working tree status and index helpers for the pgt TUI."""
from __future__ import annotations
import subprocess
from pygit2 import Commit, GitError, Index, IndexEntry, Repository, Tree
from pygit2.enums import FileStatus
from pygittools.hooks_push import git_executable
_INDEX_FLAGS = (
FileStatus.INDEX_NEW
| FileStatus.INDEX_MODIFIED
| FileStatus.INDEX_DELETED
| FileStatus.INDEX_RENAMED
| FileStatus.INDEX_TYPECHANGE
)
_WT_FLAGS = (
FileStatus.WT_MODIFIED
| FileStatus.WT_DELETED
| FileStatus.WT_TYPECHANGE
| FileStatus.WT_RENAMED
| FileStatus.WT_UNREADABLE
)
def categorize_status(repo: Repository) -> tuple[list[str], list[str], list[str]]:
"""Return staged, unstaged, and untracked paths from ``repo.status()``."""
status = repo.status()
staged = sorted(path for path, flags in status.items() if flags & _INDEX_FLAGS)
unstaged = sorted(path for path, flags in status.items() if flags & _WT_FLAGS)
untracked = sorted(path for path, flags in status.items() if flags & FileStatus.WT_NEW)
return staged, unstaged, untracked
def _index_path(path: str) -> str:
"""Normalize a path for libgit2 index operations.
pygit2 reports untracked directories with a trailing slash, but
``index.add`` rejects those paths.
"""
return path.removesuffix("/")
def stage_path(repo: Repository, path: str) -> None:
stage_paths(repo, [path])
def stage_paths(repo: Repository, paths: list[str]) -> None:
if not paths:
return
index = repo.index
for path in paths:
status_flags = repo.status().get(path, 0)
if status_flags & FileStatus.WT_DELETED:
index.remove(path)
else:
index.add(_index_path(path))
index.write()
def unstage_path(repo: Repository, path: str) -> None:
index = repo.index
_unstage_index_path(index, _head_tree(repo), path)
index.write()
def unstage_paths(repo: Repository, paths: list[str]) -> None:
if not paths:
return
index = repo.index
head_tree = _head_tree(repo)
for path in paths:
_unstage_index_path(index, head_tree, path)
index.write()
def _head_tree(repo: Repository) -> Tree | None:
if repo.head_is_unborn:
return None
try:
return repo.head.peel(Commit).tree
except GitError:
return None
def _unstage_index_path(index: Index, head_tree: Tree | None, path: str) -> None:
if head_tree is not None and path in head_tree:
entry = head_tree[path]
index.add(IndexEntry(path, entry.id, entry.filemode))
else:
index.remove(path)
def commit_staged(repo: Repository, message: str) -> tuple[bool, str]:
"""Create a commit via the git CLI so client hooks run. Returns (ok, error)."""
trimmed = message.strip()
if not trimmed:
return False, "Commit message is empty"
workdir = repo.workdir
if workdir is None:
return False, "Repository has no working directory"
staged, _, _ = categorize_status(repo)
if not staged:
return False, "Nothing staged to commit"
result = subprocess.run(
[git_executable(), "commit", "-m", trimmed],
cwd=workdir,
capture_output=True,
text=True,
check=False,
)
if result.returncode == 0:
return True, ""
output = (result.stderr or result.stdout or "Commit failed").strip()
if not output:
return False, "Commit failed"
return False, output.splitlines()[-1]