diff --git a/app/src/promptforge/core/git_manager.py b/app/src/promptforge/core/git_manager.py new file mode 100644 index 0000000..cac3f77 --- /dev/null +++ b/app/src/promptforge/core/git_manager.py @@ -0,0 +1,202 @@ +from pathlib import Path +from typing import Any, List, Optional +from datetime import datetime + +from git import Repo, Commit, GitCommandError + +from .exceptions import GitError + + +class GitManager: + """Manage git operations for prompt directories.""" + + def __init__(self, prompts_dir: Path): + """Initialize git manager for a prompts directory. + + Args: + prompts_dir: Path to the prompts directory. + """ + self.prompts_dir = Path(prompts_dir) + self.repo: Optional[Repo] = None + + def init(self) -> bool: + """Initialize a git repository in the prompts directory. + + Returns: + True if repository was created, False if it already exists. + """ + try: + if not self.prompts_dir.exists(): + self.prompts_dir.mkdir(parents=True, exist_ok=True) + + if self._is_git_repo(): + return False + + self.repo = Repo.init(str(self.prompts_dir)) + self._configure_gitignore() + self.repo.index.commit("Initial commit: PromptForge repository") + return True + except Exception as e: + raise GitError(f"Failed to initialize git repository: {e}") + + def _is_git_repo(self) -> bool: + """Check if prompts_dir is a git repository.""" + try: + self.repo = Repo(str(self.prompts_dir)) + return not self.repo.bare + except Exception: + return False + + def _configure_gitignore(self) -> None: + """Create .gitignore for prompts directory.""" + gitignore_path = self.prompts_dir / ".gitignore" + if not gitignore_path.exists(): + gitignore_path.write_text("*.lock\n.temp*\n") + + def commit(self, message: str, author: Optional[str] = None) -> Commit: + """Commit all changes to prompts. + + Args: + message: Commit message. + author: Author string (e.g., "Name "). + + Returns: + Created commit object. + + Raises: + GitError: If commit fails. + """ + self._ensure_repo() + assert self.repo is not None + try: + self.repo.index.add(["*"]) + author_arg: Any = author # type: ignore[assignment] + return self.repo.index.commit(message, author=author_arg) + except GitCommandError as e: + raise GitError(f"Failed to commit changes: {e}") + + def log(self, max_count: int = 20) -> List[Commit]: + """Get commit history. + + Args: + max_count: Maximum number of commits to return. + + Returns: + List of commit objects. + """ + self._ensure_repo() + assert self.repo is not None + try: + return list(self.repo.iter_commits(max_count=max_count)) + except GitCommandError as e: + raise GitError(f"Failed to get commit log: {e}") + + def show_commit(self, commit_sha: str) -> str: + """Show content of a specific commit. + + Args: + commit_sha: SHA of the commit. + + Returns: + Commit diff as string. + """ + self._ensure_repo() + assert self.repo is not None + try: + commit = self.repo.commit(commit_sha) + diff_result = commit.diff("HEAD~1" if commit_sha == "HEAD" else f"{commit_sha}^") + return str(diff_result) if diff_result else "" + except Exception as e: + raise GitError(f"Failed to show commit: {e}") + + def create_branch(self, branch_name: str) -> None: + """Create a new branch for prompt variations. + + Args: + branch_name: Name of the new branch. + + Raises: + GitError: If branch creation fails. + """ + self._ensure_repo() + assert self.repo is not None + try: + self.repo.create_head(branch_name) + except GitCommandError as e: + raise GitError(f"Failed to create branch: {e}") + + def switch_branch(self, branch_name: str) -> None: + """Switch to a different branch. + + Args: + branch_name: Name of the branch to switch to. + + Raises: + GitError: If branch switch fails. + """ + self._ensure_repo() + assert self.repo is not None + try: + self.repo.heads[branch_name].checkout() + except GitCommandError as e: + raise GitError(f"Failed to switch branch: {e}") + + def list_branches(self) -> List[str]: + """List all branches. + + Returns: + List of branch names. + """ + self._ensure_repo() + assert self.repo is not None + return [head.name for head in self.repo.heads] + + def status(self) -> str: + """Get git status. + + Returns: + Status string. + """ + self._ensure_repo() + assert self.repo is not None + return self.repo.git.status() + + def diff(self) -> str: + """Show uncommitted changes. + + Returns: + Diff as string. + """ + self._ensure_repo() + assert self.repo is not None + return self.repo.git.diff() + + def get_file_history(self, filename: str) -> List[dict]: + """Get commit history for a specific file. + + Args: + filename: Name of the file. + + Returns: + List of commit info dictionaries. + """ + self._ensure_repo() + assert self.repo is not None + commits = [] + try: + for commit in self.repo.iter_commits("--all", filename): + commits.append({ + "sha": commit.hexsha, + "author": str(commit.author), + "date": datetime.fromtimestamp(commit.authored_date).isoformat(), + "message": commit.message, + }) + except GitCommandError: + pass + return commits + + def _ensure_repo(self) -> None: + """Ensure repository is initialized.""" + if self.repo is None: + if not self._is_git_repo(): + raise GitError("Git repository not initialized. Run 'pf init' first.")