diff --git a/apps/api/app/git/__init__.py b/apps/api/app/git/__init__.py index 9fb2f09..247ed77 100644 --- a/apps/api/app/git/__init__.py +++ b/apps/api/app/git/__init__.py @@ -2,6 +2,7 @@ from app.git.connection import ConnectionManager, RepositoryConnectionData from app.git.credentials import AccessTokenCredential, CredentialStorage, GitCredential +from app.git.operations import GitOperations, LocalGitOperations from app.git.provider import GitProvider from app.git.ssh_key import SshKeyLifecycle, SshKeyPair from app.git.types import ConnectionStatus, CredentialKind, ProviderKind, SshKeyStatus @@ -13,7 +14,9 @@ __all__ = [ "CredentialKind", "CredentialStorage", "GitCredential", + "GitOperations", "GitProvider", + "LocalGitOperations", "ProviderKind", "RepositoryConnectionData", "SshKeyLifecycle", diff --git a/apps/api/app/git/operations.py b/apps/api/app/git/operations.py new file mode 100644 index 0000000..ea2057e --- /dev/null +++ b/apps/api/app/git/operations.py @@ -0,0 +1,101 @@ +"""Local Git subprocess interface.""" + +import abc +import subprocess +from pathlib import Path +from typing import Any + + +class GitOperations(abc.ABC): + """Local Git subprocess interface. + + This abstraction is separate from :class:`~app.git.provider.GitProvider`, + which handles remote provider API operations. + """ + + @abc.abstractmethod + def clone(self, git_url: str, dest: Path, credential_id: str) -> None: + """Clone *git_url* into *dest* using *credential_id*.""" + + @abc.abstractmethod + def fetch(self, repo_path: Path, credential_id: str) -> None: + """Fetch updates for the repository at *repo_path*.""" + + @abc.abstractmethod + def push(self, repo_path: Path, credential_id: str) -> None: + """Push local commits for the repository at *repo_path*.""" + + @abc.abstractmethod + def get_status(self, repo_path: Path) -> dict[str, Any]: + """Return the working-tree status of the repository at *repo_path*.""" + + +class LocalGitOperations(GitOperations): + """Git operations backed by the local ``git`` CLI.""" + + def clone(self, git_url: str, dest: Path, credential_id: str) -> None: + raise NotImplementedError( + "Credential-aware subprocess invocation will be implemented in a follow-up task" + ) + + def fetch(self, repo_path: Path, credential_id: str) -> None: + raise NotImplementedError( + "Credential-aware subprocess invocation will be implemented in a follow-up task" + ) + + def push(self, repo_path: Path, credential_id: str) -> None: + raise NotImplementedError( + "Credential-aware subprocess invocation will be implemented in a follow-up task" + ) + + def get_status(self, repo_path: Path) -> dict[str, Any]: + if not repo_path.exists() or not (repo_path / ".git").is_dir(): + raise RuntimeError("Not a git repository") + + branch_result = subprocess.run( + ["git", "-C", str(repo_path), "branch", "--show-current"], + capture_output=True, + text=True, + check=True, + ) + branch = branch_result.stdout.strip() + + status_result = subprocess.run( + ["git", "-C", str(repo_path), "status", "--porcelain"], + capture_output=True, + text=True, + check=True, + ) + + untracked: list[str] = [] + modified: list[str] = [] + staged: list[str] = [] + deleted: list[str] = [] + + for line in status_result.stdout.splitlines(): + if len(line) < 3: + continue + index_status = line[0] + worktree_status = line[1] + filename = line[3:] + + if index_status == "?" and worktree_status == "?": + untracked.append(filename) + elif index_status in ("M", "A"): + staged.append(filename) + elif index_status == "D" or worktree_status == "D": + deleted.append(filename) + + if worktree_status == "M": + modified.append(filename) + + clean = not (untracked or modified or staged or deleted) + + return { + "branch": branch, + "clean": clean, + "untracked": untracked, + "modified": modified, + "staged": staged, + "deleted": deleted, + }