# Commitizen API Analysis for Git Integration
## Exploring Unused Functionality in commitizen>=3.0.0
### Overview
This document provides a detailed analysis of the Commitizen library's internal APIs that are currently unused in the MCP Connector but could be leveraged to implement actual git commit functionality.
### Commitizen Internal Structure
#### Core Modules Available
1. **`commitizen.git`** - Git operations wrapper
2. **`commitizen.commands.commit`** - High-level commit command implementation
3. **`commitizen.exceptions`** - Git-specific exceptions
4. **`commitizen.config`** - Configuration management (already used)
5. **`commitizen.factory`** - Plugin factory (already used)
### Detailed API Analysis
#### 1. `commitizen.git` Module
##### GitCommit Class
```python
from commitizen.git import GitCommit
class GitCommit:
"""Wrapper for git commit operations."""
def __init__(self, config: BaseConfig):
self.config = config
def commit(
self,
message: str,
args: str = "",
retry: bool = False
) -> subprocess.CompletedProcess:
"""Execute git commit with message."""
# Returns subprocess result with commit hash, status
def write_commit_message_to_file(
self,
commit_msg: str,
commit_msg_file: str
) -> None:
"""Write commit message to file for git hooks."""
def get_tag_names(self) -> List[str]:
"""Get list of git tags."""
def is_staging_clean(self) -> bool:
"""Check if staging area is clean."""
```
##### GitTag Class
```python
from commitizen.git import GitTag
class GitTag:
"""Git tag operations."""
def create_tag(
self,
tag_name: str,
message: str = "",
signed: bool = False
) -> subprocess.CompletedProcess:
"""Create git tag."""
```
##### Utility Functions
```python
from commitizen.git import (
get_commits,
get_commit_range,
get_latest_tag_name,
is_git_project,
get_current_branch,
get_staged_files_remotes
)
# Repository validation
def is_git_project(path: str = ".") -> bool:
"""Check if directory is a git repository."""
# Commit history
def get_commits(
start: str = "",
end: str = "HEAD",
args: str = ""
) -> List[GitCommit]:
"""Get commit history between references."""
# File operations
def get_staged_files_remotes() -> List[str]:
"""Get list of staged files."""
```
#### 2. `commitizen.commands.commit` Module
##### Commit Command Class
```python
from commitizen.commands.commit import Commit
class Commit:
"""High-level commit command implementation."""
def __init__(self, config: BaseConfig, arguments: dict):
self.config = config
self.arguments = arguments
self.commit_parser = factory.commit_parser_factory(config)
self.committer = factory.committer_factory(config)
def __call__(self) -> None:
"""Execute the commit command."""
# Full commit workflow implementation
def _get_commit_message(self) -> str:
"""Generate commit message interactively."""
def _prepare_commit_message_file(self, message: str) -> str:
"""Prepare commit message file for git."""
def _commit_with_message(self, message: str) -> None:
"""Execute git commit with prepared message."""
```
#### 3. Exception Handling
```python
from commitizen.exceptions import (
CommitizenException,
GitCommandError,
NoStagedFilesError,
NotAGitProjectError,
NoCommitsFoundError
)
# Git-specific exceptions
class GitCommandError(CommitizenException):
"""Git command execution failed."""
class NoStagedFilesError(CommitizenException):
"""No files staged for commit."""
class NotAGitProjectError(CommitizenException):
"""Directory is not a git repository."""
```
### Implementation Strategy
#### 1. GitService Implementation
Based on the API analysis, here's how we can implement the GitService:
```python
"""
Git Service Implementation using Commitizen APIs
"""
import logging
from typing import Dict, List, Any, Optional
from pathlib import Path
import subprocess
from commitizen.git import GitCommit, GitTag, is_git_project, get_staged_files_remotes
from commitizen.commands.commit import Commit
from commitizen.exceptions import (
CommitizenException,
GitCommandError,
NoStagedFilesError,
NotAGitProjectError
)
from commitizen.config import BaseConfig
logger = logging.getLogger(__name__)
class GitService:
"""Service class leveraging Commitizen's git functionality."""
def __init__(self, repo_path: Optional[str] = None):
"""Initialize with Commitizen's git classes."""
self.repo_path = Path(repo_path) if repo_path else Path.cwd()
# Initialize Commitizen components
self.config = BaseConfig()
self.git_commit = GitCommit(self.config)
self.git_tag = GitTag()
# Validate repository
if not is_git_project(str(self.repo_path)):
raise NotAGitProjectError(f"Not a git repository: {self.repo_path}")
def validate_repository(self) -> Dict[str, Any]:
"""Validate git repository state using Commitizen APIs."""
try:
validation = {
"is_git_repo": is_git_project(str(self.repo_path)),
"has_staged_files": not self.git_commit.is_staging_clean(),
"staged_files": self.get_staged_files(),
"current_branch": self._get_current_branch(),
"repository_path": str(self.repo_path)
}
return {
"valid": validation["is_git_repo"],
"details": validation
}
except Exception as e:
logger.error(f"Repository validation failed: {e}")
return {
"valid": False,
"error": str(e)
}
def get_staged_files(self) -> List[str]:
"""Get staged files using Commitizen API."""
try:
return get_staged_files_remotes()
except Exception as e:
logger.error(f"Failed to get staged files: {e}")
return []
def get_repository_status(self) -> Dict[str, Any]:
"""Get comprehensive repository status."""
try:
staged_files = self.get_staged_files()
status = {
"staged_files": staged_files,
"staged_count": len(staged_files),
"staging_clean": self.git_commit.is_staging_clean(),
"repository_valid": is_git_project(str(self.repo_path)),
"current_branch": self._get_current_branch()
}
return status
except Exception as e:
logger.error(f"Failed to get repository status: {e}")
raise RuntimeError(f"Repository status check failed: {e}")
def preview_commit(
self,
message: str,
stage_all: bool = False,
sign_off: bool = False,
additional_args: Optional[List[str]] = None
) -> Dict[str, Any]:
"""Preview commit without executing (dry-run)."""
try:
# Validate repository state
repo_status = self.get_repository_status()
if stage_all:
# Would stage all files (preview only)
preview_files = self._get_unstaged_files()
else:
preview_files = repo_status["staged_files"]
# Build git command that would be executed
git_args = []
if sign_off:
git_args.append("--signoff")
if additional_args:
git_args.extend(additional_args)
preview = {
"message": message,
"files_to_commit": preview_files,
"file_count": len(preview_files),
"git_args": git_args,
"repository_status": repo_status,
"would_stage_all": stage_all,
"dry_run": True
}
return preview
except Exception as e:
logger.error(f"Commit preview failed: {e}")
raise RuntimeError(f"Commit preview failed: {e}")
def execute_commit(
self,
message: str,
stage_all: bool = False,
sign_off: bool = False,
dry_run: bool = True,
additional_args: Optional[List[str]] = None
) -> Dict[str, Any]:
"""Execute git commit using Commitizen's GitCommit class."""
try:
# Safety check - validate repository
validation = self.validate_repository()
if not validation["valid"]:
raise GitCommandError(f"Repository validation failed: {validation.get('error')}")
# Check for staged files
if not stage_all and self.git_commit.is_staging_clean():
raise NoStagedFilesError("No files staged for commit")
# Build git arguments
git_args = []
if stage_all:
git_args.append("-a") # Stage all changes
if sign_off:
git_args.append("--signoff")
if additional_args:
git_args.extend(additional_args)
args_string = " ".join(git_args)
if dry_run:
# Return what would be executed
return {
"dry_run": True,
"message": message,
"args": args_string,
"would_execute": f"git commit {args_string} -m '{message}'",
"repository_status": self.get_repository_status()
}
else:
# Execute actual commit using Commitizen's API
result = self.git_commit.commit(
message=message,
args=args_string
)
return {
"success": result.returncode == 0,
"commit_hash": self._extract_commit_hash(result),
"message": message,
"args": args_string,
"git_output": result.stdout.decode() if result.stdout else "",
"git_error": result.stderr.decode() if result.stderr else "",
"return_code": result.returncode
}
except (GitCommandError, NoStagedFilesError, NotAGitProjectError) as e:
logger.error(f"Git commit failed: {e}")
return {
"success": False,
"error": str(e),
"error_type": type(e).__name__
}
except Exception as e:
logger.error(f"Unexpected error during commit: {e}")
return {
"success": False,
"error": str(e),
"error_type": "UnexpectedError"
}
def stage_files(self, files: List[str]) -> Dict[str, Any]:
"""Stage specific files for commit."""
try:
# Use subprocess for git add (Commitizen doesn't expose this directly)
import subprocess
cmd = ["git", "add"] + files
result = subprocess.run(
cmd,
cwd=self.repo_path,
capture_output=True,
text=True
)
return {
"success": result.returncode == 0,
"staged_files": files,
"git_output": result.stdout,
"git_error": result.stderr
}
except Exception as e:
logger.error(f"Failed to stage files: {e}")
return {
"success": False,
"error": str(e)
}
def _get_current_branch(self) -> str:
"""Get current git branch."""
try:
from commitizen.git import get_current_branch
return get_current_branch()
except Exception:
return "unknown"
def _get_unstaged_files(self) -> List[str]:
"""Get list of unstaged files."""
try:
import subprocess
result = subprocess.run(
["git", "diff", "--name-only"],
cwd=self.repo_path,
capture_output=True,
text=True
)
return result.stdout.strip().split('\n') if result.stdout.strip() else []
except Exception:
return []
def _extract_commit_hash(self, result: subprocess.CompletedProcess) -> Optional[str]:
"""Extract commit hash from git commit output."""
try:
output = result.stdout.decode() if result.stdout else ""
# Git commit output typically starts with commit hash
lines = output.strip().split('\n')
for line in lines:
if line.startswith('[') and ']' in line:
# Format: [branch commit_hash] message
parts = line.split(']')[0].split()
if len(parts) >= 2:
return parts[-1] # Last part should be hash
return None
except Exception:
return None
```
#### 2. Enhanced CommitzenService Integration
```python
# Extension to existing CommitzenService class
from .git_service import GitService
class CommitzenService:
def __init__(self):
# Existing initialization...
try:
self.git_service = GitService()
self.git_enabled = True
logger.info("Git service initialized successfully")
except NotAGitProjectError:
self.git_service = None
self.git_enabled = False
logger.info("Git service disabled - not in a git repository")
except Exception as e:
self.git_service = None
self.git_enabled = False
logger.warning(f"Git service initialization failed: {e}")
def prepare_commit(
self,
message: str,
stage_all: bool = False,
sign_off: bool = False
) -> Dict[str, Any]:
"""Prepare commit with validation and preview."""
if not self.git_enabled:
return {
"error": "Git operations not available - not in a git repository",
"git_enabled": False
}
try:
# Validate the message first
is_valid = self.validate_message(message)
if not is_valid:
return {
"error": "Invalid commit message format",
"message": message,
"is_valid": False
}
# Get commit preview
preview = self.git_service.preview_commit(
message=message,
stage_all=stage_all,
sign_off=sign_off
)
return {
"message": message,
"is_valid": is_valid,
"preview": preview,
"git_enabled": True
}
except Exception as e:
logger.error(f"Failed to prepare commit: {e}")
return {
"error": str(e),
"message": message,
"git_enabled": True
}
def execute_commit_with_approval(
self,
message: str,
stage_all: bool = False,
sign_off: bool = False,
force_execute: bool = False
) -> Dict[str, Any]:
"""Execute commit after validation and safety checks."""
if not self.git_enabled:
return {
"error": "Git operations not available",
"success": False
}
try:
# Always do dry run first unless force_execute is True
dry_run = not force_execute
result = self.git_service.execute_commit(
message=message,
stage_all=stage_all,
sign_off=sign_off,
dry_run=dry_run
)
return result
except Exception as e:
logger.error(f"Failed to execute commit: {e}")
return {
"error": str(e),
"success": False
}
```
### Key Benefits of Using Commitizen APIs
1. **Consistency**: Uses the same git operations as Commitizen CLI
2. **Error Handling**: Leverages Commitizen's robust error handling
3. **Configuration**: Respects existing Commitizen configuration
4. **Validation**: Built-in repository and commit validation
5. **Compatibility**: Works with all Commitizen plugins and configurations
### Integration Points
#### Existing MCP Tools Enhancement
The current MCP tools can be enhanced to include git operations:
```python
@mcp.tool()
def generate_commit_message_with_preview(
type: str,
subject: str,
body: Optional[str] = None,
scope: Optional[str] = None,
breaking: Optional[bool] = False,
footer: Optional[str] = None,
include_git_preview: bool = True
) -> Dict[str, Any]:
"""Enhanced version that includes git preview."""
# Generate message (existing functionality)
result = generate_commit_message(type, subject, body, scope, breaking, footer)
# Add git preview if requested and available
if include_git_preview and service.git_enabled:
preview = service.prepare_commit(result["message"])
result["git_preview"] = preview
return result
```
### Testing Strategy with Real APIs
```python
# Test file: tests/test_git_integration.py
import pytest
from unittest.mock import Mock, patch
from commitizen.exceptions import NotAGitProjectError, NoStagedFilesError
from commitizen_mcp_connector.git_service import GitService
class TestGitService:
@patch('commitizen.git.is_git_project')
def test_init_valid_repo(self, mock_is_git):
"""Test initialization in valid git repository."""
mock_is_git.return_value = True
service = GitService()
assert service.repo_path is not None
@patch('commitizen.git.is_git_project')
def test_init_invalid_repo(self, mock_is_git):
"""Test initialization in non-git directory."""
mock_is_git.return_value = False
with pytest.raises(NotAGitProjectError):
GitService()
@patch('commitizen.git.get_staged_files_remotes')
def test_get_staged_files(self, mock_staged_files):
"""Test getting staged files."""
mock_staged_files.return_value = ['file1.py', 'file2.py']
# Test implementation
def test_preview_commit_dry_run(self):
"""Test commit preview functionality."""
# Test implementation with mocked git operations
def test_execute_commit_safety_checks(self):
"""Test that safety checks prevent accidental commits."""
# Test implementation
```
This analysis shows that Commitizen provides comprehensive git integration APIs that can be safely leveraged to add actual commit functionality while maintaining consistency with the existing Commitizen ecosystem.
