# GitPython Exploration: Complete Replacement Analysis
## Overview
This document explores replacing `commitizen.git` entirely with GitPython in the Commitizen MCP Connector. The analysis covers implementation changes, benefits, challenges, and compatibility with normal commit behavior.
## Current Implementation Analysis
### Current commitizen.git Usage
```python
from commitizen.git import (
is_git_project, # Repository validation
is_staging_clean, # Check if staging area is clean
commit, # Execute git commit
add, # Stage files
get_commits, # Get commit history
GitCommit # Commit object wrapper
)
```
### Current Limitations
1. **Directory Changes Required**: All operations require `os.chdir(repo_path)`
2. **Subprocess-Based**: Wraps git commands with subprocess calls
3. **Limited Information**: Basic git operation results
4. **Error Handling**: Generic subprocess error handling
5. **Thread Safety**: Directory changes can cause race conditions
## GitPython Alternative Implementation
### Core Architecture Changes
#### 1. Repository Object Management
```python
# BEFORE (commitizen.git)
import os
from commitizen.git import is_git_project, commit
original_cwd = os.getcwd()
os.chdir(repo_path)
if is_git_project():
result = commit(message, args, committer_date)
os.chdir(original_cwd)
# AFTER (GitPython)
from git import Repo, InvalidGitRepositoryError
try:
repo = Repo(repo_path)
commit_obj = repo.index.commit(message)
except InvalidGitRepositoryError:
raise NotAGitProjectError(f"Not a git repository: {repo_path}")
```
#### 2. Enhanced GitService Class
```python
"""
GitPython-based GitService - Complete replacement implementation
"""
import logging
from pathlib import Path
from typing import Dict, Any, List, Optional, Union
from datetime import datetime
from git import Repo, InvalidGitRepositoryError, GitCommandError
from git.exc import GitError, NoSuchPathError
from git.objects import Commit
from git import Actor
logger = logging.getLogger(__name__)
class GitPythonService:
"""
Pure GitPython implementation replacing all commitizen.git functionality.
Key improvements:
- No directory changes required
- Rich git object access
- Better error handling
- Thread-safe operations
- Enhanced repository information
"""
def __init__(self, repo_path: Optional[Union[str, Path]] = None):
"""Initialize with GitPython Repo object."""
self.repo_path = Path(repo_path) if repo_path else Path.cwd()
try:
self.repo = Repo(self.repo_path)
logger.info(f"GitPython repository initialized: {self.repo_path}")
except InvalidGitRepositoryError:
raise NotAGitProjectError(f"Not a git repository: {self.repo_path}")
except Exception as e:
raise GitOperationError(f"Failed to initialize repository: {e}")
# Repository validation methods
def is_git_project(self) -> bool:
"""Replace commitizen.git.is_git_project()"""
return self.repo is not None and self.repo.git_dir is not None
def is_staging_clean(self) -> bool:
"""Replace commitizen.git.is_staging_clean()"""
try:
# Check if index differs from HEAD (has staged changes)
return len(self.repo.index.diff("HEAD")) == 0
except Exception:
return True # Assume clean on error
# Enhanced file operations
def get_staged_files(self) -> List[str]:
"""Get list of staged files using GitPython."""
try:
return [item.a_path for item in self.repo.index.diff("HEAD")]
except Exception as e:
logger.warning(f"Could not get staged files: {e}")
return []
def get_unstaged_files(self) -> List[str]:
"""Get list of unstaged files."""
try:
return [item.a_path for item in self.repo.index.diff(None)]
except Exception as e:
logger.warning(f"Could not get unstaged files: {e}")
return []
def get_untracked_files(self) -> List[str]:
"""Get list of untracked files."""
try:
return self.repo.untracked_files
except Exception as e:
logger.warning(f"Could not get untracked files: {e}")
return []
# Enhanced repository status
def get_repository_status(self) -> Dict[str, Any]:
"""Enhanced repository status using pure GitPython."""
try:
staged_files = self.get_staged_files()
unstaged_files = self.get_unstaged_files()
untracked_files = self.get_untracked_files()
# Get current branch info
try:
current_branch = self.repo.active_branch.name
branch_commit = self.repo.active_branch.commit
except Exception:
current_branch = "HEAD (detached)"
branch_commit = self.repo.head.commit
# Get recent commits with rich information
recent_commits = []
try:
for commit in self.repo.iter_commits(max_count=5):
recent_commits.append({
"sha": commit.hexsha,
"short_sha": commit.hexsha[:8],
"message": commit.message.strip(),
"summary": commit.summary,
"author_name": commit.author.name,
"author_email": commit.author.email,
"committer_name": commit.committer.name,
"committer_email": commit.committer.email,
"authored_date": commit.authored_datetime.isoformat(),
"committed_date": commit.committed_datetime.isoformat(),
"parents": [parent.hexsha[:8] for parent in commit.parents],
"stats": {
"total_insertions": commit.stats.total["insertions"],
"total_deletions": commit.stats.total["deletions"],
"files_changed": commit.stats.total["files"]
}
})
except Exception as e:
logger.warning(f"Could not get recent commits: {e}")
# Get repository statistics
try:
total_commits = sum(1 for _ in self.repo.iter_commits())
total_branches = len(list(self.repo.branches))
total_tags = len(list(self.repo.tags))
except Exception:
total_commits = total_branches = total_tags = 0
return {
"repository_path": str(self.repo_path),
"is_git_repository": True,
"staging_clean": len(staged_files) == 0,
"staged_files": staged_files,
"staged_files_count": len(staged_files),
"unstaged_files": unstaged_files,
"unstaged_files_count": len(unstaged_files),
"untracked_files": untracked_files,
"untracked_files_count": len(untracked_files),
"current_branch": current_branch,
"head_commit": {
"sha": branch_commit.hexsha[:8],
"message": branch_commit.summary,
"author": branch_commit.author.name,
"committed_date": branch_commit.committed_datetime.isoformat()
},
"recent_commits": recent_commits,
"repository_stats": {
"total_commits": total_commits,
"total_branches": total_branches,
"total_tags": total_tags
}
}
except Exception as e:
logger.error(f"Failed to get repository status: {e}")
raise GitOperationError(f"Repository status check failed: {e}")
# Enhanced file staging
def add_files(self, *files: str, force_execute: bool = False) -> Dict[str, Any]:
"""Replace commitizen.git.add() with GitPython."""
if not force_execute:
return {
"success": False,
"error": "force_execute=True required for actual file staging",
"files": list(files),
"executed": False,
"preview": self._preview_add_files(*files)
}
try:
# Validate file paths
validated_files = []
for file_path in files:
validated_path = self._validate_file_path(file_path)
validated_files.append(validated_path)
# Add files using GitPython
self.repo.index.add(validated_files)
# Get updated status
updated_staged = self.get_staged_files()
return {
"success": True,
"files": validated_files,
"executed": True,
"repository_path": str(self.repo_path),
"updated_staged_files": updated_staged,
"staged_files_count": len(updated_staged)
}
except GitError as e:
logger.error(f"Git add failed: {e}")
return {
"success": False,
"error": f"Git add failed: {e}",
"files": list(files),
"executed": False
}
except Exception as e:
logger.error(f"Add execution failed: {e}")
return {
"success": False,
"error": f"Add execution failed: {e}",
"files": list(files),
"executed": False
}
# Enhanced commit execution
def execute_commit(
self,
message: str,
force_execute: bool = False,
author_name: Optional[str] = None,
author_email: Optional[str] = None,
committer_name: Optional[str] = None,
committer_email: Optional[str] = None,
commit_date: Optional[datetime] = None,
sign_off: bool = False,
**kwargs
) -> Dict[str, Any]:
"""Replace commitizen.git.commit() with enhanced GitPython implementation."""
if not force_execute:
return {
"success": False,
"error": "force_execute=True required for actual commit execution",
"message": message,
"executed": False,
"preview": self.preview_commit(message, **kwargs)
}
try:
# Sanitize message
sanitized_message = self._sanitize_commit_message(message)
# Add sign-off if requested
if sign_off:
# Get git config for sign-off
try:
user_name = self.repo.config_reader().get_value("user", "name")
user_email = self.repo.config_reader().get_value("user", "email")
sanitized_message += f"\n\nSigned-off-by: {user_name} <{user_email}>"
except Exception:
logger.warning("Could not add sign-off - git user config not found")
# Check for staged changes
if self.is_staging_clean():
return {
"success": False,
"error": "No staged changes to commit",
"message": sanitized_message,
"executed": False
}
# Prepare commit parameters
commit_kwargs = {}
# Set author if provided
if author_name and author_email:
commit_kwargs['author'] = Actor(author_name, author_email)
# Set committer if provided
if committer_name and committer_email:
commit_kwargs['committer'] = Actor(committer_name, committer_email)
# Set commit date if provided
if commit_date:
commit_kwargs['commit_date'] = commit_date
commit_kwargs['author_date'] = commit_date
# Execute commit using GitPython
commit_obj = self.repo.index.commit(sanitized_message, **commit_kwargs)
# Get commit statistics
commit_stats = commit_obj.stats.total
return {
"success": True,
"message": sanitized_message,
"original_message": message,
"executed": True,
"commit_hash": commit_obj.hexsha,
"commit_short_hash": commit_obj.hexsha[:8],
"author": {
"name": commit_obj.author.name,
"email": commit_obj.author.email
},
"committer": {
"name": commit_obj.committer.name,
"email": commit_obj.committer.email
},
"authored_date": commit_obj.authored_datetime.isoformat(),
"committed_date": commit_obj.committed_datetime.isoformat(),
"stats": {
"files_changed": commit_stats["files"],
"insertions": commit_stats["insertions"],
"deletions": commit_stats["deletions"]
},
"parents": [parent.hexsha[:8] for parent in commit_obj.parents],
"repository_path": str(self.repo_path)
}
except GitError as e:
logger.error(f"Git commit failed: {e}")
return {
"success": False,
"error": f"Git commit failed: {e}",
"message": message,
"executed": False
}
except Exception as e:
logger.error(f"Commit execution failed: {e}")
return {
"success": False,
"error": f"Commit execution failed: {e}",
"message": message,
"executed": False
}
# Enhanced commit preview
def preview_commit(self, message: str, **kwargs) -> Dict[str, Any]:
"""Enhanced commit preview using GitPython's diff capabilities."""
try:
status = self.get_repository_status()
if status["staging_clean"]:
return {
"success": False,
"error": "No staged changes to commit",
"staged_files": [],
"message": message,
"would_execute": False
}
# Get detailed diff information
staged_diff = self.repo.index.diff("HEAD")
changes_detail = []
total_insertions = 0
total_deletions = 0
for item in staged_diff:
try:
# Get diff statistics
diff_text = item.diff.decode('utf-8', errors='ignore')
insertions = diff_text.count('\n+') - diff_text.count('\n+++')
deletions = diff_text.count('\n-') - diff_text.count('\n---')
changes_detail.append({
"file": item.a_path or item.b_path,
"change_type": item.change_type,
"insertions": max(0, insertions),
"deletions": max(0, deletions),
"is_binary": item.diff == b'',
"old_file": item.a_path,
"new_file": item.b_path
})
total_insertions += max(0, insertions)
total_deletions += max(0, deletions)
except Exception as e:
logger.warning(f"Could not analyze diff for {item.a_path}: {e}")
changes_detail.append({
"file": item.a_path or item.b_path,
"change_type": item.change_type,
"insertions": 0,
"deletions": 0,
"is_binary": True,
"error": str(e)
})
return {
"success": True,
"message": message,
"staged_files": status["staged_files"],
"staged_files_count": status["staged_files_count"],
"changes_detail": changes_detail,
"total_insertions": total_insertions,
"total_deletions": total_deletions,
"total_changes": total_insertions + total_deletions,
"would_execute": True,
"repository_path": str(self.repo_path),
"current_branch": status["current_branch"],
"head_commit": status["head_commit"]
}
except Exception as e:
logger.error(f"Commit preview failed: {e}")
return {
"success": False,
"error": str(e),
"message": message,
"would_execute": False
}
# Enhanced commit history
def get_commits(self, max_count: int = 10, since: Optional[str] = None) -> List[Dict[str, Any]]:
"""Replace commitizen.git.get_commits() with enhanced GitPython implementation."""
try:
commits = []
# Build iterator arguments
iter_kwargs = {"max_count": max_count}
if since:
iter_kwargs["since"] = since
for commit in self.repo.iter_commits(**iter_kwargs):
# Get commit statistics
try:
stats = commit.stats.total
files_changed = stats["files"]
insertions = stats["insertions"]
deletions = stats["deletions"]
except Exception:
files_changed = insertions = deletions = 0
commits.append({
"sha": commit.hexsha,
"short_sha": commit.hexsha[:8],
"message": commit.message.strip(),
"summary": commit.summary,
"author_name": commit.author.name,
"author_email": commit.author.email,
"committer_name": commit.committer.name,
"committer_email": commit.committer.email,
"authored_date": commit.authored_datetime.isoformat(),
"committed_date": commit.committed_datetime.isoformat(),
"parents": [parent.hexsha[:8] for parent in commit.parents],
"stats": {
"files_changed": files_changed,
"insertions": insertions,
"deletions": deletions
}
})
return commits
except Exception as e:
logger.error(f"Failed to get commits: {e}")
return []
# Utility methods
def _validate_file_path(self, file_path: str) -> str:
"""Validate file path is within repository bounds."""
if not file_path or not file_path.strip():
raise ValueError("File path cannot be empty")
try:
# Resolve path and ensure it's within repository
full_path = (self.repo_path / file_path).resolve()
repo_root_resolved = self.repo_path.resolve()
# Check if path is within repository bounds
if not str(full_path).startswith(str(repo_root_resolved)):
raise ValueError(f"File path outside repository: {file_path}")
return file_path.strip()
except Exception as e:
raise ValueError(f"Invalid file path '{file_path}': {e}")
def _sanitize_commit_message(self, message: str) -> str:
"""Sanitize commit message for security."""
if not message or not message.strip():
raise ValueError("Commit message cannot be empty")
# Remove dangerous shell metacharacters
dangerous_chars = ['`', '$', ';', '|', '&', '>', '<', '\x00']
sanitized = message
for char in dangerous_chars:
sanitized = sanitized.replace(char, '')
# Basic length limit
if len(sanitized) > 1000:
raise ValueError("Commit message too long (max 1000 characters)")
# Remove excessive whitespace
import re
sanitized = re.sub(r'\s+', ' ', sanitized.strip())
if not sanitized:
raise ValueError("Commit message becomes empty after sanitization")
return sanitized
def _preview_add_files(self, *files: str) -> Dict[str, Any]:
"""Preview file staging operation."""
try:
file_info = []
for file_path in files:
try:
validated_path = self._validate_file_path(file_path)
full_path = self.repo_path / validated_path
if full_path.exists():
file_info.append({
"file": validated_path,
"exists": True,
"size": full_path.stat().st_size,
"is_tracked": validated_path in [item.a_path for item in self.repo.index.diff(None)]
})
else:
file_info.append({
"file": validated_path,
"exists": False,
"error": "File does not exist"
})
except ValueError as e:
file_info.append({
"file": file_path,
"exists": False,
"error": str(e)
})
return {
"files": file_info,
"total_files": len(files),
"valid_files": len([f for f in file_info if f.get("exists", False)])
}
except Exception as e:
return {
"error": str(e),
"files": list(files)
}
```
## Key Benefits of GitPython Replacement
### 1. **No Directory Changes Required**
- **Before**: Every operation requires `os.chdir(repo_path)` and restoration
- **After**: Repository object maintains context automatically
- **Benefit**: Thread-safe, cleaner code, no race conditions
### 2. **Richer Repository Information**
- **Before**: Basic subprocess output parsing
- **After**: Direct access to git objects with rich metadata
- **Benefit**: Better user experience with detailed information
### 3. **Enhanced Error Handling**
```python
# GitPython provides specific exception types
try:
commit_obj = repo.index.commit(message)
except GitCommandError as e:
# Specific git command failures
except InvalidGitRepositoryError as e:
# Repository validation issues
except GitError as e:
# General git errors
```
### 4. **Better Performance**
- **Before**: Subprocess overhead for each git operation
- **After**: Direct git library calls
- **Benefit**: Faster operations, especially for status checks
### 5. **Advanced Git Features**
- Detailed diff analysis with insertions/deletions
- Rich commit object access
- Branch and tag management
- Repository statistics
## Normal Commit Behavior Compatibility
### ✅ **Fully Compatible**
GitPython maintains complete compatibility with normal git behavior:
1. **Standard Git Operations**: All normal git commands work identically
2. **Git Hooks**: Fully supports pre-commit, post-commit, and other git hooks
3. **Repository State**: Maintains all git repository state and history
4. **Interoperability**: Works seamlessly with command-line git and other tools
5. **Configuration**: Respects git configuration files and settings
### Enhanced Capabilities
The GitPython implementation provides **additional** capabilities without breaking normal behavior:
- **Detailed Commit Statistics**: File changes, insertions, deletions
- **Rich Commit History**: Enhanced commit information with metadata
- **Better Diff Analysis**: Line-by-line change analysis
- **Repository Introspection**: Branch, tag, and repository statistics
## Implementation Comparison
### Current vs GitPython Implementation
| Aspect | Current (commitizen.git) | GitPython Alternative |
|--------|-------------------------|----------------------|
| **Directory Changes** | Required (`os.chdir`) | Not required |
| **Thread Safety** | Limited (directory changes) | Full thread safety |
| **Error Handling** | Generic subprocess errors | Specific git exceptions |
| **Repository Info** | Basic status | Rich metadata |
| **Performance** | Subprocess overhead | Direct library calls |
| **Diff Analysis** | Limited | Detailed line-by-line |
| **Commit Objects** | Basic wrapper | Full git object access |
| **Configuration** | Limited access | Full git config access |
### MCP Tool Enhancement Example
```python
# Enhanced MCP tool with GitPython
@mcp.tool()
def get_repository_status_enhanced(repo_path: str) -> Dict[str, Any]:
"""Enhanced repository status with GitPython."""
try:
git_service = GitPythonService(repo_path)
status = git_service.get_repository_status()
return {
"success": True,
"repository_status": status,
"enhanced_features": {
"detailed_file_status": True,
"commit_statistics": True,
"branch_information": True,
"repository_metrics": True
}
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
@mcp.tool()
def preview_commit_enhanced(
message: str,
repo_path: str
) -> Dict[str, Any]:
"""Enhanced commit preview with detailed diff analysis."""
try:
git_service = GitPythonService(repo_path)
preview = git_service.preview_commit(message)
return {
"success": True,
"preview": preview,
"enhanced_features": {
"line_by_line_diff": True,
"insertion_deletion_count": True,
"file_change_analysis": True,
"binary_file_detection": True
}
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
```
## Migration Strategy
### Phase 1: Add GitPython Dependency
```toml
dependencies = [
"mcp[cli]>=1.10.0",
"commitizen>=3.0.0", # Keep for message generation
"GitPython>=3.1.40" # Add GitPython
]
```
### Phase 2: Implement GitPython Service
1. Create `GitPythonService` class
2. Implement all methods to replace `commitizen.git` functions
3. Add comprehensive error handling
4. Maintain same interfaces for MCP tools
### Phase 3: Update Imports and Integration
```python
# Remove commitizen.git imports
# from commitizen.git import is_git_project, commit, add
# Add GitPython imports
from git import Repo, InvalidGitRepositoryError, GitCommandError
# Update service initialization
class CommitzenService:
def __init__(self, repo_path: Optional[str] = None):
# Keep Commitizen for message generation
self.config = BaseConfig()
self.plugin_adapter = PluginAdapter(self.config)
# Use GitPython for git operations
try:
self.git_service = GitPythonService(repo_path)
self.git_enabled = True
except InvalidGitRepositoryError:
self.git_service = None
self.git_enabled = False
```
### Phase 4: Enhanced Features
- Leverage GitPython's advanced capabilities
- Add richer MCP tool responses
- Implement better error messages
- Add repository analytics
## Testing Strategy
### Compatibility Tests
```python
def test_gitpython_vs_commitizen_compatibility():
"""Ensure GitPython implementation produces same results as commitizen.git."""
# Test repository validation
assert gitpython_service.is_git_project() == commitizen_is_git_project()
# Test staging status
assert gitpython_service.is_staging_clean() == commitizen_is_staging_clean()
# Test commit execution (in test environment)
# Verify same commit hash and metadata
```
### Enhanced Feature Tests
```python
def test_gitpython_enhanced_features():
"""Test GitPython-specific enhanced features."""
status = gitpython_service.get_repository_status()
# Test enhanced information
assert "repository_stats" in status
assert "recent_commits" in status
assert "current_branch" in status
# Test detailed diff analysis
preview = gitpython_service.preview_commit("test message")
assert "total_insertions" in preview
assert "total_deletions" in preview
assert "changes_detail" in preview
```
## Conclusion
**Complete replacement of `commitizen.git` with GitPython would provide significant benefits:**
1. **✅ Maintains Full Compatibility**: All normal git behavior preserved
2. **✅ Eliminates Complexity**: No more directory changes or subprocess management
3. **✅ Enhances Capabilities**: Richer information and better user experience
4. **✅ Improves Performance**: Direct library calls vs subprocess overhead
5. **✅ Better Maintainability**: Cleaner code with specific error handling
**The implementation would be transparent to users** while providing enhanced functionality through the same MCP tool interfaces. This represents a significant architectural improvement that maintains backward compatibility while enabling future enhancements.
**Recommendation**: Proceed with GitPython replacement as it offers substantial benefits with no downsides for normal commit behavior.
