Skip to content

Utils

Git utilities for CodeMap.

logger module-attribute 

logger = getLogger(__name__)

GitDiff dataclass

Represents a Git diff chunk.

Source code in src/codemap/git/utils.py 
21
22
23
24
25
26
27
28
@dataclass
class GitDiff:
	"""Represents a Git diff chunk."""

	files: list[str]
	content: str
	is_staged: bool = False
	is_untracked: bool = False

__init__ 

__init__(
	files: list[str],
	content: str,
	is_staged: bool = False,
	is_untracked: bool = False,
) -> None

files instance-attribute 

files: list[str]

content instance-attribute 

content: str

is_staged class-attribute instance-attribute 

is_staged: bool = False

is_untracked class-attribute instance-attribute 

is_untracked: bool = False

GitError

Bases: Exception

Custom exception for Git-related errors.

Source code in src/codemap/git/utils.py 
31
32
class GitError(Exception):
	"""Custom exception for Git-related errors."""

ExtendedGitRepoContext

Bases: GitRepoContext

Extended context for Git operations using pygit2.

Source code in src/codemap/git/utils.py 
 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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class ExtendedGitRepoContext(GitRepoContext):
	"""Extended context for Git operations using pygit2."""

	_extended_instance: ExtendedGitRepoContext | None = None

	@classmethod
	def get_instance(cls) -> ExtendedGitRepoContext:
		"""Get an instance of the ExtendedGitRepoContext class."""
		if cls._extended_instance is None:
			cls._extended_instance = cls()
		return cls._extended_instance

	def __init__(self) -> None:
		"""Initialize the ExtendedGitRepoContext with the given repository path."""
		super().__init__()

	@classmethod
	def validate_repo_path(cls, path: Path | None = None) -> Path | None:
		"""Validate and return the repository path, or None if not valid."""
		try:
			if path is None:
				path = Path.cwd()
			return cls.get_repo_root(path)
		except GitError:
			return None

	def get_staged_diff(self) -> GitDiff:
		"""Get the diff of staged changes as a GitDiff object."""
		commit = self.repo.head.peel(Commit)
		diff = self.repo.diff(commit.tree, cached=True)

		files = []
		content = ""
		if isinstance(diff, Diff):
			files = [delta.delta.new_file.path for delta in diff]
			content = diff.patch
		elif isinstance(diff, Patch):
			files = [diff.delta.new_file.path]
			content = diff.text
		return GitDiff(files=files, content=content or "", is_staged=True)

	def get_unstaged_diff(self) -> GitDiff:
		"""Get the diff of unstaged changes as a GitDiff object."""
		diff = self.repo.diff()
		files = []

		content = ""

		if isinstance(diff, Diff):
			files = [delta.delta.new_file.path for delta in diff]
			content = diff.patch
		elif isinstance(diff, Patch):
			files = [diff.delta.new_file.path]
			content = diff.text

		return GitDiff(files=files, content=content or "", is_staged=False)

	def get_other_staged_files(self, targeted_files: list[str]) -> list[str]:
		"""Get staged files that are not part of the targeted files."""
		all_staged = self.get_staged_diff().files
		return [f for f in all_staged if f not in targeted_files]

	def stash_staged_changes(self, exclude_files: list[str]) -> bool:
		"""Temporarily stash staged changes except for specified files."""
		try:
			other_files = self.get_other_staged_files(exclude_files)
			if not other_files:
				return False
			self.stage_files(other_files)
		except GitError as e:
			msg = "Failed to stash other staged changes"
			raise GitError(msg) from e
		else:
			return True

	def unstash_changes(self) -> None:
		"""Restore previously stashed changes."""
		try:
			stash_list = self.get_other_staged_files([])
			if "CodeMap: temporary stash for commit" in stash_list:
				self.unstage_files(stash_list)
		except GitError as e:
			msg = "Failed to restore stashed changes; you may need to manually run 'git stash pop'"
			raise GitError(msg) from e

	def commit_only_files(
		self,
		files: list[str],
		message: str,
		ignore_hooks: bool = False,
	) -> list[str]:
		"""
		Commit only the specified files with the given message.

		Runs the pre-commit, commit-msg, and post-commit hooks unless ignore_hooks is True.
		"""
		import tempfile

		# Run pre-commit hook if not ignored
		if not ignore_hooks and hook_exists("pre-commit"):
			exit_code = run_hook("pre-commit")
			if exit_code != 0:
				error_msg = "pre-commit hook failed, aborting commit."
				logger.error(error_msg)
				raise RuntimeError(error_msg)
		try:
			# Prepare commit-msg hook: write message to a temp file if needed
			commit_msg_file = None
			if not ignore_hooks and hook_exists("commit-msg"):
				with tempfile.NamedTemporaryFile("w+", delete=False) as f:
					f.write(message)
					commit_msg_file = f.name
				exit_code = run_hook("commit-msg", repo_root=None)  # Could pass file as env var if needed
				if exit_code != 0:
					error_msg = "commit-msg hook failed, aborting commit."
					logger.error(error_msg)
					if commit_msg_file:
						Path(commit_msg_file).unlink()
					raise RuntimeError(error_msg)
			# self.stage_files(files) # Removed: Index is already prepared by the caller
			other_staged = self.get_other_staged_files(files)
			try:
				self.commit(message)
				logger.info("Created commit with message: %s", message)
			except GitError as e:
				error_msg = "Git commit command failed"
				logger.exception(error_msg)
				raise GitError(error_msg) from e
			# Run post-commit hook if not ignored
			if not ignore_hooks and hook_exists("post-commit"):
				exit_code = run_hook("post-commit")
				if exit_code != 0:
					logger.warning("post-commit hook failed (commit already created)")
			if commit_msg_file:
				Path(commit_msg_file).unlink()
			return other_staged
		except GitError:
			raise
		except Exception as e:
			error_msg = f"Error in commit_only_files: {e!s}"
			logger.exception(error_msg)
			raise GitError(error_msg) from e

	def get_per_file_diff(self, file_path: str, staged: bool = False) -> GitDiff:
		"""
		Get the diff for a single file, either staged or unstaged.

		Args:
			file_path: The path to the file to diff (relative to repo root).
			staged: If True, get the staged diff; otherwise, get the unstaged diff.

		Returns:
			GitDiff: The diff for the specified file.

		Raises:
			GitError: If the diff cannot be generated.
		"""
		logger.debug("get_per_file_diff called with file_path: '%s', staged: %s", file_path, staged)
		try:
			if staged:
				commit = self.repo.head.peel(Commit)
				diff = self.repo.diff(commit.tree, cached=True)
				is_staged = True
			else:
				diff = self.repo.diff()
				is_staged = False

			file_path_set = {file_path}
			if isinstance(diff, Diff):
				for patch in diff:
					new_file_path = patch.delta.new_file.path
					old_file_path = patch.delta.old_file.path
					logger.debug(
						"  Patch details - New: '%s', Old: '%s'",
						new_file_path,
						old_file_path,
					)
					if {new_file_path, old_file_path} & file_path_set:
						content = patch.text or ""
						logger.debug("    Patch text (first 200 chars): %s", repr(content[:200]))
						files = [new_file_path]
						git_diff_obj = GitDiff(files=files, content=content, is_staged=is_staged)
						logger.debug(
							"    Returning GitDiff for '%s', content length: %d",
							file_path,
							len(git_diff_obj.content),
						)
						return git_diff_obj
				logger.debug("  No matching patch found in Diff for '%s'. Returning empty GitDiff.", file_path)
				return GitDiff(files=[file_path], content="", is_staged=is_staged)
			if isinstance(diff, Patch):
				new_file_path = diff.delta.new_file.path
				old_file_path = diff.delta.old_file.path
				logger.debug(
					"  Patch details (standalone) - New: '%s', Old: '%s'",
					new_file_path,
					old_file_path,
				)
				if {new_file_path, old_file_path} & file_path_set:
					content = diff.text or ""
					logger.debug("    Patch text (first 200 chars): %s", repr(content[:200]))
					files = [new_file_path]
					git_diff_obj = GitDiff(files=files, content=content, is_staged=is_staged)
					logger.debug(
						"    Returning GitDiff for '%s' (standalone patch), content length: %d",
						file_path,
						len(git_diff_obj.content),
					)
					return git_diff_obj
				logger.debug("  Standalone Patch does not match '%s'. Returning empty GitDiff.", file_path)
				return GitDiff(files=[file_path], content="", is_staged=is_staged)
			logger.debug("  Diff object is neither Diff nor Patch for '%s'. Returning empty GitDiff.", file_path)
			return GitDiff(files=[file_path], content="", is_staged=is_staged)
		except Exception as e:
			logger.exception("Failed to get %s diff for %s", "staged" if staged else "unstaged", file_path)
			msg = f"Failed to get {'staged' if staged else 'unstaged'} diff for {file_path}: {e}"
			raise GitError(msg) from e

get_instance classmethod 

get_instance() -> ExtendedGitRepoContext

Get an instance of the ExtendedGitRepoContext class.

Source code in src/codemap/git/utils.py 
40
41
42
43
44
45
@classmethod
def get_instance(cls) -> ExtendedGitRepoContext:
	"""Get an instance of the ExtendedGitRepoContext class."""
	if cls._extended_instance is None:
		cls._extended_instance = cls()
	return cls._extended_instance

__init__ 

__init__() -> None

Initialize the ExtendedGitRepoContext with the given repository path.

Source code in src/codemap/git/utils.py 
47
48
49
def __init__(self) -> None:
	"""Initialize the ExtendedGitRepoContext with the given repository path."""
	super().__init__()

validate_repo_path classmethod 

validate_repo_path(path: Path | None = None) -> Path | None

Validate and return the repository path, or None if not valid.

Source code in src/codemap/git/utils.py 
51
52
53
54
55
56
57
58
59
@classmethod
def validate_repo_path(cls, path: Path | None = None) -> Path | None:
	"""Validate and return the repository path, or None if not valid."""
	try:
		if path is None:
			path = Path.cwd()
		return cls.get_repo_root(path)
	except GitError:
		return None

get_staged_diff 

get_staged_diff() -> GitDiff

Get the diff of staged changes as a GitDiff object.

Source code in src/codemap/git/utils.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def get_staged_diff(self) -> GitDiff:
	"""Get the diff of staged changes as a GitDiff object."""
	commit = self.repo.head.peel(Commit)
	diff = self.repo.diff(commit.tree, cached=True)

	files = []
	content = ""
	if isinstance(diff, Diff):
		files = [delta.delta.new_file.path for delta in diff]
		content = diff.patch
	elif isinstance(diff, Patch):
		files = [diff.delta.new_file.path]
		content = diff.text
	return GitDiff(files=files, content=content or "", is_staged=True)

get_unstaged_diff 

get_unstaged_diff() -> GitDiff

Get the diff of unstaged changes as a GitDiff object.

Source code in src/codemap/git/utils.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def get_unstaged_diff(self) -> GitDiff:
	"""Get the diff of unstaged changes as a GitDiff object."""
	diff = self.repo.diff()
	files = []

	content = ""

	if isinstance(diff, Diff):
		files = [delta.delta.new_file.path for delta in diff]
		content = diff.patch
	elif isinstance(diff, Patch):
		files = [diff.delta.new_file.path]
		content = diff.text

	return GitDiff(files=files, content=content or "", is_staged=False)

get_other_staged_files 

get_other_staged_files(
	targeted_files: list[str],
) -> list[str]

Get staged files that are not part of the targeted files.

Source code in src/codemap/git/utils.py 
92
93
94
95
def get_other_staged_files(self, targeted_files: list[str]) -> list[str]:
	"""Get staged files that are not part of the targeted files."""
	all_staged = self.get_staged_diff().files
	return [f for f in all_staged if f not in targeted_files]

stash_staged_changes 

stash_staged_changes(exclude_files: list[str]) -> bool

Temporarily stash staged changes except for specified files.

Source code in src/codemap/git/utils.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
def stash_staged_changes(self, exclude_files: list[str]) -> bool:
	"""Temporarily stash staged changes except for specified files."""
	try:
		other_files = self.get_other_staged_files(exclude_files)
		if not other_files:
			return False
		self.stage_files(other_files)
	except GitError as e:
		msg = "Failed to stash other staged changes"
		raise GitError(msg) from e
	else:
		return True

unstash_changes 

unstash_changes() -> None

Restore previously stashed changes.

Source code in src/codemap/git/utils.py 
110
111
112
113
114
115
116
117
118
def unstash_changes(self) -> None:
	"""Restore previously stashed changes."""
	try:
		stash_list = self.get_other_staged_files([])
		if "CodeMap: temporary stash for commit" in stash_list:
			self.unstage_files(stash_list)
	except GitError as e:
		msg = "Failed to restore stashed changes; you may need to manually run 'git stash pop'"
		raise GitError(msg) from e

commit_only_files 

commit_only_files(
	files: list[str],
	message: str,
	ignore_hooks: bool = False,
) -> list[str]

Commit only the specified files with the given message.

Runs the pre-commit, commit-msg, and post-commit hooks unless ignore_hooks is True.

Source code in src/codemap/git/utils.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
def commit_only_files(
	self,
	files: list[str],
	message: str,
	ignore_hooks: bool = False,
) -> list[str]:
	"""
	Commit only the specified files with the given message.

	Runs the pre-commit, commit-msg, and post-commit hooks unless ignore_hooks is True.
	"""
	import tempfile

	# Run pre-commit hook if not ignored
	if not ignore_hooks and hook_exists("pre-commit"):
		exit_code = run_hook("pre-commit")
		if exit_code != 0:
			error_msg = "pre-commit hook failed, aborting commit."
			logger.error(error_msg)
			raise RuntimeError(error_msg)
	try:
		# Prepare commit-msg hook: write message to a temp file if needed
		commit_msg_file = None
		if not ignore_hooks and hook_exists("commit-msg"):
			with tempfile.NamedTemporaryFile("w+", delete=False) as f:
				f.write(message)
				commit_msg_file = f.name
			exit_code = run_hook("commit-msg", repo_root=None)  # Could pass file as env var if needed
			if exit_code != 0:
				error_msg = "commit-msg hook failed, aborting commit."
				logger.error(error_msg)
				if commit_msg_file:
					Path(commit_msg_file).unlink()
				raise RuntimeError(error_msg)
		# self.stage_files(files) # Removed: Index is already prepared by the caller
		other_staged = self.get_other_staged_files(files)
		try:
			self.commit(message)
			logger.info("Created commit with message: %s", message)
		except GitError as e:
			error_msg = "Git commit command failed"
			logger.exception(error_msg)
			raise GitError(error_msg) from e
		# Run post-commit hook if not ignored
		if not ignore_hooks and hook_exists("post-commit"):
			exit_code = run_hook("post-commit")
			if exit_code != 0:
				logger.warning("post-commit hook failed (commit already created)")
		if commit_msg_file:
			Path(commit_msg_file).unlink()
		return other_staged
	except GitError:
		raise
	except Exception as e:
		error_msg = f"Error in commit_only_files: {e!s}"
		logger.exception(error_msg)
		raise GitError(error_msg) from e

get_per_file_diff 

get_per_file_diff(
	file_path: str, staged: bool = False
) -> GitDiff

Get the diff for a single file, either staged or unstaged.

Parameters:

Name Type Description Default
file_path str

The path to the file to diff (relative to repo root).

 required
staged bool

If True, get the staged diff; otherwise, get the unstaged diff.

 False

Returns:

Name Type Description
GitDiff GitDiff

The diff for the specified file.

Raises:

Type Description
GitError

If the diff cannot be generated.
Source code in src/codemap/git/utils.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def get_per_file_diff(self, file_path: str, staged: bool = False) -> GitDiff:
	"""
	Get the diff for a single file, either staged or unstaged.

	Args:
		file_path: The path to the file to diff (relative to repo root).
		staged: If True, get the staged diff; otherwise, get the unstaged diff.

	Returns:
		GitDiff: The diff for the specified file.

	Raises:
		GitError: If the diff cannot be generated.
	"""
	logger.debug("get_per_file_diff called with file_path: '%s', staged: %s", file_path, staged)
	try:
		if staged:
			commit = self.repo.head.peel(Commit)
			diff = self.repo.diff(commit.tree, cached=True)
			is_staged = True
		else:
			diff = self.repo.diff()
			is_staged = False

		file_path_set = {file_path}
		if isinstance(diff, Diff):
			for patch in diff:
				new_file_path = patch.delta.new_file.path
				old_file_path = patch.delta.old_file.path
				logger.debug(
					"  Patch details - New: '%s', Old: '%s'",
					new_file_path,
					old_file_path,
				)
				if {new_file_path, old_file_path} & file_path_set:
					content = patch.text or ""
					logger.debug("    Patch text (first 200 chars): %s", repr(content[:200]))
					files = [new_file_path]
					git_diff_obj = GitDiff(files=files, content=content, is_staged=is_staged)
					logger.debug(
						"    Returning GitDiff for '%s', content length: %d",
						file_path,
						len(git_diff_obj.content),
					)
					return git_diff_obj
			logger.debug("  No matching patch found in Diff for '%s'. Returning empty GitDiff.", file_path)
			return GitDiff(files=[file_path], content="", is_staged=is_staged)
		if isinstance(diff, Patch):
			new_file_path = diff.delta.new_file.path
			old_file_path = diff.delta.old_file.path
			logger.debug(
				"  Patch details (standalone) - New: '%s', Old: '%s'",
				new_file_path,
				old_file_path,
			)
			if {new_file_path, old_file_path} & file_path_set:
				content = diff.text or ""
				logger.debug("    Patch text (first 200 chars): %s", repr(content[:200]))
				files = [new_file_path]
				git_diff_obj = GitDiff(files=files, content=content, is_staged=is_staged)
				logger.debug(
					"    Returning GitDiff for '%s' (standalone patch), content length: %d",
					file_path,
					len(git_diff_obj.content),
				)
				return git_diff_obj
			logger.debug("  Standalone Patch does not match '%s'. Returning empty GitDiff.", file_path)
			return GitDiff(files=[file_path], content="", is_staged=is_staged)
		logger.debug("  Diff object is neither Diff nor Patch for '%s'. Returning empty GitDiff.", file_path)
		return GitDiff(files=[file_path], content="", is_staged=is_staged)
	except Exception as e:
		logger.exception("Failed to get %s diff for %s", "staged" if staged else "unstaged", file_path)
		msg = f"Failed to get {'staged' if staged else 'unstaged'} diff for {file_path}: {e}"
		raise GitError(msg) from e