alex
3728c245d3
- Add GitHubAdapter and GitLabAdapter with URL parsing - Create provider factory in apps/api/app/git/providers/ - Implement clone, fetch, push in LocalGitOperations - Add repository_connections router with CRUD and SSH key endpoints - Create RepositoryConnection schema with validation - Update models and routers __init__.py for new components - Add comprehensive tests for git operations
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
import abc
|
|
import subprocess
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
class GitOperations(abc.ABC):
|
|
@abc.abstractmethod
|
|
def clone(self, git_url: str, dest: Path, credential_id: str) -> None:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def fetch(self, repo_path: Path, credential_id: str) -> None:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def push(self, repo_path: Path, credential_id: str) -> None:
|
|
pass
|
|
|
|
@abc.abstractmethod
|
|
def get_status(self, repo_path: Path) -> dict[str, Any]:
|
|
pass
|
|
|
|
|
|
class LocalGitOperations(GitOperations):
|
|
def clone(self, git_url: str, dest: Path, credential_id: str) -> None:
|
|
cmd = ["git", "clone", git_url, str(dest)]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Git clone failed: {result.stderr}")
|
|
|
|
def fetch(self, repo_path: Path, credential_id: str) -> None:
|
|
cmd = ["git", "-C", str(repo_path), "fetch", "--all"]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Git fetch failed: {result.stderr}")
|
|
|
|
def push(self, repo_path: Path, credential_id: str) -> None:
|
|
cmd = ["git", "-C", str(repo_path), "push"]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(f"Git push failed: {result.stderr}")
|
|
|
|
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")
|
|
|
|
try:
|
|
branch_result = subprocess.run(
|
|
["git", "-C", str(repo_path), "branch", "--show-current"],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
check=True,
|
|
)
|
|
branch = branch_result.stdout.strip()
|
|
|
|
status_result = subprocess.run(
|
|
["git", "-C", str(repo_path), "status", "--porcelain"],
|
|
capture_output=True,
|
|
text=True,
|
|
encoding="utf-8",
|
|
check=True,
|
|
)
|
|
except subprocess.CalledProcessError as exc:
|
|
raise RuntimeError("Git command failed") from exc
|
|
|
|
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)
|
|
|
|
if 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,
|
|
}
|