|
| 1 | +"""Git helper function.""" |
| 2 | + |
| 3 | +import logging |
| 4 | +from pathlib import Path |
| 5 | +from typing import ClassVar, Self |
| 6 | + |
| 7 | +from ..constants import GITLOGGER_NAME |
| 8 | +from ..models.repo import Repo |
| 9 | +from ..utils.shell import execute_git_command |
| 10 | + |
| 11 | +log = logging.getLogger(GITLOGGER_NAME) |
| 12 | + |
| 13 | + |
| 14 | +class ManagedGitRepo: |
| 15 | + """Manages a bare repository and its temporary worktrees.""" |
| 16 | + |
| 17 | + #: Class variable to indicate the update state of a repo |
| 18 | + _is_updated: ClassVar[dict[Repo, bool]] = {} |
| 19 | + |
| 20 | + def __init__(self: Self, remote_url: str, permanent_root: Path) -> None: |
| 21 | + """Initialize the managed repository. |
| 22 | +
|
| 23 | + :param remote_url: The remote URL of the repository. |
| 24 | + :param permanent_root: The root directory for storing permanent bare clones. |
| 25 | + """ |
| 26 | + self._repo_model = Repo(remote_url) |
| 27 | + self._permanent_root = permanent_root |
| 28 | + # The Repo model handles the "sluggification" of the URL |
| 29 | + self.bare_repo_path = self._permanent_root / self._repo_model.slug |
| 30 | + # Initialize attribute for output: |
| 31 | + self.stdout = self.stderr = None |
| 32 | + # Add repo into class variable |
| 33 | + type(self)._is_updated.setdefault(self._repo_model, False) |
| 34 | + |
| 35 | + def __repr__(self: Self) -> str: |
| 36 | + """Return a string representation of the ManagedGitRepo.""" |
| 37 | + return ( |
| 38 | + f'{self.__class__.__name__}(remote_url={self.remote_url!r}, ' |
| 39 | + f"bare_repo_path='{self.bare_repo_path!s}')" |
| 40 | + ) |
| 41 | + |
| 42 | + @property |
| 43 | + def slug(self: Self) -> str: |
| 44 | + """Return the slug of the repository.""" |
| 45 | + return self._repo_model.slug |
| 46 | + |
| 47 | + @property |
| 48 | + def remote_url(self: Self) -> str: |
| 49 | + """Return the remote URL of the repository.""" |
| 50 | + return self._repo_model.url |
| 51 | + |
| 52 | + @property |
| 53 | + def permanent_root(self: Self) -> Path: |
| 54 | + """Return the permanent root directory for the repository.""" |
| 55 | + return self._permanent_root |
| 56 | + |
| 57 | + async def _initial_clone(self: Self) -> bool: |
| 58 | + """Execute the initial 'git clone --bare' command. |
| 59 | +
|
| 60 | + This is a helper for `clone_bare` and assumes the destination |
| 61 | + directory does not exist. |
| 62 | +
|
| 63 | + :returns: True if the clone was successful, False on error. |
| 64 | + """ |
| 65 | + url = self._repo_model.url |
| 66 | + try: |
| 67 | + self.stdout, self.stderr = await execute_git_command( |
| 68 | + 'clone', |
| 69 | + '--bare', |
| 70 | + '--progress', |
| 71 | + str(url), |
| 72 | + str(self.bare_repo_path), |
| 73 | + cwd=self._permanent_root, |
| 74 | + ) |
| 75 | + log.info("Cloned '%s' successfully", url) |
| 76 | + return True |
| 77 | + |
| 78 | + except RuntimeError as e: |
| 79 | + log.error("Failed to clone '%s': %s", url, e) |
| 80 | + return False |
| 81 | + |
| 82 | + async def clone_bare(self: Self) -> bool: |
| 83 | + """Clone the remote repository as a bare repository. |
| 84 | +
|
| 85 | + If the repository already exists, it logs a message and returns. |
| 86 | +
|
| 87 | + :returns: True if successful, False otherwise. |
| 88 | + """ |
| 89 | + url = self._repo_model.url |
| 90 | + cls = type(self) |
| 91 | + |
| 92 | + if cls._is_updated.get(self._repo_model, False): |
| 93 | + log.info('Repository %r already processed this run.', self._repo_model.name) |
| 94 | + return True |
| 95 | + |
| 96 | + result = False |
| 97 | + if self.bare_repo_path.exists(): |
| 98 | + log.info( |
| 99 | + 'Repository already exists, fetching updates for %r', |
| 100 | + self._repo_model.name, |
| 101 | + ) |
| 102 | + result = await self.fetch_updates() |
| 103 | + else: |
| 104 | + log.info("Cloning '%s' into '%s'...", url, self.bare_repo_path) |
| 105 | + result = await self._initial_clone() |
| 106 | + |
| 107 | + if result: |
| 108 | + cls._is_updated[self._repo_model] = True |
| 109 | + |
| 110 | + return result |
| 111 | + |
| 112 | + async def create_worktree( |
| 113 | + self: Self, |
| 114 | + target_dir: Path, |
| 115 | + branch: str, |
| 116 | + *, |
| 117 | + is_local: bool = True, |
| 118 | + options: list[str] | None = None, |
| 119 | + ) -> None: |
| 120 | + """Create a temporary worktree from the bare repository.""" |
| 121 | + if not self.bare_repo_path.exists(): |
| 122 | + raise FileNotFoundError( |
| 123 | + 'Cannot create worktree. Bare repository does not exist at: ' |
| 124 | + f'{self.bare_repo_path}' |
| 125 | + ) |
| 126 | + |
| 127 | + clone_args = ['clone'] |
| 128 | + if is_local: |
| 129 | + clone_args.append('--local') |
| 130 | + clone_args.extend(['--branch', branch]) |
| 131 | + if options: |
| 132 | + clone_args.extend(options) |
| 133 | + clone_args.extend([str(self.bare_repo_path), str(target_dir)]) |
| 134 | + |
| 135 | + self.stdout, self.stderr = await execute_git_command( |
| 136 | + *clone_args, cwd=target_dir.parent |
| 137 | + ) |
| 138 | + |
| 139 | + async def fetch_updates(self: Self) -> bool: |
| 140 | + """Fetch updates from the remote to the bare repository. |
| 141 | +
|
| 142 | + :return: True if successful, False otherwise. |
| 143 | + """ |
| 144 | + if not self.bare_repo_path.exists(): |
| 145 | + log.warning( |
| 146 | + 'Cannot fetch updates: Bare repository does not exist at %s', |
| 147 | + self.bare_repo_path, |
| 148 | + ) |
| 149 | + return False |
| 150 | + |
| 151 | + log.info("Fetching updates for '%s'", self.slug) |
| 152 | + try: |
| 153 | + self.stdout, self.stderr = await execute_git_command( |
| 154 | + 'fetch', '--all', cwd=self.bare_repo_path |
| 155 | + ) |
| 156 | + log.info("Successfully fetched updates for '%s'", self.slug) |
| 157 | + return True |
| 158 | + |
| 159 | + except RuntimeError as e: |
| 160 | + log.error("Failed to fetch updates for '%s': %s", self.slug, e) |
| 161 | + return False |
0 commit comments