|
| 1 | +from __future__ import annotations |
| 2 | + |
| 3 | +from pathlib import Path |
| 4 | + |
| 5 | +from dulwich.objects import Tree |
| 6 | +from dulwich.repo import Repo |
| 7 | + |
| 8 | + |
| 9 | +class GitRepoManager: |
| 10 | + def __init__(self, root_directory: str, branch: str = "main"): |
| 11 | + self.root_directory = root_directory |
| 12 | + self.git: Repo | None = None |
| 13 | + self.branch = branch |
| 14 | + |
| 15 | + self.initialize_repo() |
| 16 | + |
| 17 | + def initialize_repo(self) -> None: |
| 18 | + # Check if the directory already has a repository |
| 19 | + |
| 20 | + root_path = Path(self.root_directory) |
| 21 | + |
| 22 | + if root_path.exists() and (root_path / ".git").is_dir(): |
| 23 | + self.git = Repo(self.root_directory) # Open existing repo |
| 24 | + else: |
| 25 | + self.git = Repo.init(self.root_directory, default_branch=self.branch.encode("utf-8")) |
| 26 | + self.create_initial_commit() |
| 27 | + # Ensure the repository is valid |
| 28 | + if not self.git: |
| 29 | + raise ValueError("Failed to initialize or open a repository.") |
| 30 | + |
| 31 | + def create_initial_commit(self) -> None: |
| 32 | + if not self.git: |
| 33 | + raise ValueError("Git repository not initialized.") |
| 34 | + |
| 35 | + """Create an initial commit if no commits exist.""" |
| 36 | + # Create an empty tree object |
| 37 | + tree = Tree() |
| 38 | + self.git.object_store.add_object(tree) |
| 39 | + |
| 40 | + # Create the initial commit without an index (use empty tree) |
| 41 | + author = committer = b"Initial Commit <[email protected]>" |
| 42 | + commit_msg = b"Initial commit" |
| 43 | + |
| 44 | + ref = f"refs/heads/{self.branch}".encode() |
| 45 | + |
| 46 | + self.git.do_commit(author=author, committer=committer, message=commit_msg, ref=ref) |
| 47 | + |
| 48 | + # Set HEAD reference to point to the main branch |
| 49 | + self.git.refs.set_symbolic_ref(b"HEAD", ref) |
| 50 | + |
| 51 | + @property |
| 52 | + def active_branch(self) -> str | None: |
| 53 | + """Get the name of the current active branch.""" |
| 54 | + if not self.git: |
| 55 | + raise ValueError("Repository is not initialized.") |
| 56 | + |
| 57 | + # Read the symbolic reference of HEAD to get the current commit SHA |
| 58 | + head_ref = self.git.refs[b"HEAD"] |
| 59 | + if head_ref: |
| 60 | + commit_sha = head_ref.decode("utf-8") # Commit SHA from HEAD |
| 61 | + # Now look for the branch that points to this commit SHA |
| 62 | + for ref in self.git.refs.as_dict().keys(): |
| 63 | + if ref.startswith(b"refs/heads/"): |
| 64 | + branch_ref = ref.decode("utf-8") |
| 65 | + # Check if this branch points to the same commit SHA |
| 66 | + if self.git.refs[ref] == commit_sha.encode("utf-8"): |
| 67 | + return branch_ref.split("/")[-1] |
| 68 | + return None # If no matching branch is found |
| 69 | + raise ValueError("No HEAD reference found, cannot determine active branch.") |
0 commit comments