Add config, git_manager, exceptions

src/promptforge/core/git_manager.py

@@ -0,0 +1,101 @@
+from pathlib import Path
+from typing import List, Optional
+from datetime import datetime
+
+from git import Repo, Commit, GitCommandError
+
+from .exceptions import GitError
+
+
+class GitManager:
+    def __init__(self, prompts_dir: Path):
+        self.prompts_dir = Path(prompts_dir)
+        self.repo: Optional[Repo] = None
+
+    def init(self) -> bool:
+        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:
+        try:
+            self.repo = Repo(str(self.prompts_dir))
+            return not self.repo.bare
+        except Exception:
+            return False
+
+    def _configure_gitignore(self) -> None:
+        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:
+        self._ensure_repo()
+        try:
+            self.repo.index.add(["*"])
+            return self.repo.index.commit(message, author=author)
+        except GitCommandError as e:
+            raise GitError(f"Failed to commit changes: {e}")
+
+    def log(self, max_count: int = 20) -> List[Commit]:
+        self._ensure_repo()
+        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 create_branch(self, branch_name: str) -> None:
+        self._ensure_repo()
+        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:
+        self._ensure_repo()
+        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]:
+        self._ensure_repo()
+        return [head.name for head in self.repo.heads]
+
+    def status(self) -> str:
+        self._ensure_repo()
+        return self.repo.git.status()
+
+    def diff(self) -> str:
+        self._ensure_repo()
+        return self.repo.git.diff()
+
+    def get_file_history(self, filename: str) -> List[dict]:
+        self._ensure_repo()
+        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:
+        if self.repo is None:
+            if not self._is_git_repo():
+                raise GitError("Git repository not initialized. Run 'pf init' first.")