|
| 1 | +from git import Repo, InvalidGitRepositoryError, GitCommandError |
| 2 | +import shutil |
| 3 | +import os |
| 4 | +import tempfile |
| 5 | +from typing import List, Union |
| 6 | + |
| 7 | + |
| 8 | +def get_current_repo() -> Repo: |
| 9 | + """ |
| 10 | + Get the current Git repository from the current working directory or its parent directories. |
| 11 | +
|
| 12 | + Returns: |
| 13 | + The Repo object representing the current Git repository. |
| 14 | +
|
| 15 | + Raises: |
| 16 | + InvalidGitRepositoryError: If the current directory is not within a Git repository. |
| 17 | + """ |
| 18 | + try: |
| 19 | + return Repo(os.getcwd(), search_parent_directories=True) |
| 20 | + except InvalidGitRepositoryError: |
| 21 | + raise InvalidGitRepositoryError( |
| 22 | + "Current directory is not within a Git repository." |
| 23 | + ) |
| 24 | + |
| 25 | + |
| 26 | +def checkout_new_branch(branch_name: str) -> None: |
| 27 | + """ |
| 28 | + Create and checkout a new Git branch with the given name. |
| 29 | + This operation will fail if the branch already exists. |
| 30 | +
|
| 31 | + Args: |
| 32 | + branch_name: Name of the branch to create and checkout. |
| 33 | +
|
| 34 | + Raises: |
| 35 | + GitCommandError: If the branch already exists. |
| 36 | + """ |
| 37 | + repo = get_current_repo() |
| 38 | + git = repo.git |
| 39 | + # Attempt to create and checkout the new branch |
| 40 | + git.checkout(b=branch_name) |
| 41 | + |
| 42 | + |
| 43 | +def add_and_commit(paths: List[str], message: str) -> str: |
| 44 | + """ |
| 45 | + Add changes to the Git index for the specified paths and commit them with a message. |
| 46 | +
|
| 47 | + Args: |
| 48 | + paths: List of paths to add to the Git index. |
| 49 | + message: Commit message. |
| 50 | +
|
| 51 | + Returns: |
| 52 | + The commit ID of the new commit. |
| 53 | + """ |
| 54 | + repo = get_current_repo() |
| 55 | + index = repo.index |
| 56 | + index.add(paths) |
| 57 | + commit = index.commit(message) |
| 58 | + return commit.hexsha |
| 59 | + |
| 60 | + |
| 61 | +def get_current_head() -> Union[str, None]: |
| 62 | + """ |
| 63 | + Get the current HEAD of the repository, which is either the branch name or the commit SHA. |
| 64 | +
|
| 65 | + Returns: |
| 66 | + The branch name if on a branch, otherwise the commit SHA. |
| 67 | + """ |
| 68 | + repo = get_current_repo() |
| 69 | + if repo.head.is_detached: |
| 70 | + return str(repo.head.commit.hexsha) |
| 71 | + else: |
| 72 | + return str(repo.active_branch.name) |
| 73 | + |
| 74 | + |
| 75 | +def copy_files_to_new_branch_or_commit(paths: List[str], new_ref: str) -> None: |
| 76 | + """ |
| 77 | + Copy specified files from the current branch, switch to a new branch or commit, and |
| 78 | + copy those files into the new branch or commit. |
| 79 | +
|
| 80 | + Args: |
| 81 | + paths: List of file paths to copy. |
| 82 | + new_ref: The name of the new branch to create or the commit SHA to checkout. |
| 83 | +
|
| 84 | + Raises: |
| 85 | + FileNotFoundError: If any of the specified paths do not exist. |
| 86 | + GitCommandError: If there are issues with Git operations. |
| 87 | + """ |
| 88 | + repo = get_current_repo() |
| 89 | + current_files = {path: os.path.join(repo.working_tree_dir, path) for path in paths} |
| 90 | + |
| 91 | + # Create a temporary directory to store backups |
| 92 | + with tempfile.TemporaryDirectory() as temp_dir: |
| 93 | + backup_files = { |
| 94 | + path: os.path.join(temp_dir, os.path.basename(path)) for path in paths |
| 95 | + } |
| 96 | + |
| 97 | + # Copy files to temporary backup |
| 98 | + for path in paths: |
| 99 | + if not os.path.exists(current_files[path]): |
| 100 | + raise FileNotFoundError(f"{path} does not exist in the current branch.") |
| 101 | + shutil.copy2(current_files[path], backup_files[path]) |
| 102 | + |
| 103 | + # Checkout the new branch or commit |
| 104 | + try: |
| 105 | + repo.git.checkout(new_ref) |
| 106 | + except GitCommandError as e: |
| 107 | + raise e |
| 108 | + |
| 109 | + # Restore files from backup |
| 110 | + for path in paths: |
| 111 | + shutil.copy2(backup_files[path], current_files[path]) |
0 commit comments