From 762e0de44c47760a9c8cf5c08990f3c8b6d34b38 Mon Sep 17 00:00:00 2001 From: Fusion Date: Tue, 19 May 2026 14:49:22 +0200 Subject: [PATCH] feat: add git control endpoints (status, branch, commit, fetch, pull, push, merge) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add GitStatus dataclass and git control utilities - Add endpoints: - GET /status - working directory status - POST /branches - create branch - DELETE /branches/{name} - delete branch - POST /checkout - checkout branch - POST /commit - commit changes - POST /fetch - fetch from remote - POST /pull - pull updates - POST /push - push changes - POST /merge - merge branches Quality gates: ruff ✓, mypy ✓ --- apps/api/src/api/git_repositories.py | 331 +++++++++++++++++++++++++++ apps/api/src/utils/git_control.py | 282 +++++++++++++++++++++++ 2 files changed, 613 insertions(+) create mode 100644 apps/api/src/utils/git_control.py diff --git a/apps/api/src/api/git_repositories.py b/apps/api/src/api/git_repositories.py index ac4d067..9b19a8f 100644 --- a/apps/api/src/api/git_repositories.py +++ b/apps/api/src/api/git_repositories.py @@ -20,6 +20,17 @@ from src.utils.git_files import ( list_branches, list_tree, ) +from src.utils.git_control import ( + checkout_branch, + commit_changes, + create_branch, + delete_branch, + fetch, + get_status, + merge, + pull, + push, +) from src.utils.git_history import get_commit_detail, get_commit_history from src.utils.git_url_parser import parse_git_url @@ -466,3 +477,323 @@ async def update_repository_file( ) except RuntimeError as e: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +# Git Control Endpoints + + +class StatusResponse(BaseModel): + branch: str + modified: list[str] + added: list[str] + deleted: list[str] + untracked: list[str] + renamed: list[str] + ahead: int + behind: int + + +@router.get("/{project_id}/repositories/{repo_id}/status", response_model=StatusResponse) +async def get_repository_status( + project_id: uuid.UUID, + repo_id: uuid.UUID, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> StatusResponse: + """Get the working directory status.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + status_result = get_status(repo.path) + return StatusResponse( + branch=status_result.branch, + modified=status_result.modified, + added=status_result.added, + deleted=status_result.deleted, + untracked=status_result.untracked, + renamed=status_result.renamed, + ahead=status_result.ahead, + behind=status_result.behind, + ) + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +class BranchCreateRequest(BaseModel): + name: str + base_branch: str = "HEAD" + + +class CheckoutRequest(BaseModel): + branch: str + + +@router.post("/{project_id}/repositories/{repo_id}/branches") +async def create_repository_branch( + project_id: uuid.UUID, + repo_id: uuid.UUID, + data: BranchCreateRequest, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> dict: + """Create a new branch.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + create_branch(repo.path, data.name, data.base_branch) + return {"message": f"Branch '{data.name}' created", "branch": data.name} + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +@router.delete("/{project_id}/repositories/{repo_id}/branches/{branch_name}") +async def delete_repository_branch( + project_id: uuid.UUID, + repo_id: uuid.UUID, + branch_name: str, + force: bool = False, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> dict: + """Delete a branch.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + delete_branch(repo.path, branch_name, force) + return {"message": f"Branch '{branch_name}' deleted"} + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +@router.post("/{project_id}/repositories/{repo_id}/checkout") +async def checkout_repository_branch( + project_id: uuid.UUID, + repo_id: uuid.UUID, + data: CheckoutRequest, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> dict: + """Checkout a branch.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + checkout_branch(repo.path, data.branch) + return {"message": f"Checked out branch '{data.branch}'", "branch": data.branch} + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +class CommitRequest(BaseModel): + message: str + files: list[str] | None = None + + +class CommitResponse(BaseModel): + commit_hash: str + message: str + + +@router.post("/{project_id}/repositories/{repo_id}/commit", response_model=CommitResponse) +async def commit_repository_changes( + project_id: uuid.UUID, + repo_id: uuid.UUID, + data: CommitRequest, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> CommitResponse: + """Commit changes to the repository.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + # Get user info for commit + user = await _get_user(session, user_id) + author_name = user.name or "Unknown" + author_email = user.email or "unknown@example.com" + + try: + commit_hash = commit_changes( + repo_path=repo.path, + message=data.message, + author_name=author_name, + author_email=author_email, + files=data.files, + ) + return CommitResponse( + commit_hash=commit_hash, + message=data.message, + ) + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +class FetchResponse(BaseModel): + message: str + + +@router.post("/{project_id}/repositories/{repo_id}/fetch", response_model=FetchResponse) +async def fetch_repository( + project_id: uuid.UUID, + repo_id: uuid.UUID, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> FetchResponse: + """Fetch from remote.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + fetch(repo.path) + return FetchResponse(message="Fetched from remote") + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +class PullResponse(BaseModel): + message: str + + +@router.post("/{project_id}/repositories/{repo_id}/pull", response_model=PullResponse) +async def pull_repository( + project_id: uuid.UUID, + repo_id: uuid.UUID, + branch: str | None = None, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> PullResponse: + """Pull updates from remote.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + pull(repo.path, branch) + return PullResponse(message="Pulled from remote") + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +class PushResponse(BaseModel): + message: str + + +@router.post("/{project_id}/repositories/{repo_id}/push", response_model=PushResponse) +async def push_repository( + project_id: uuid.UUID, + repo_id: uuid.UUID, + branch: str | None = None, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> PushResponse: + """Push changes to remote.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + push(repo.path, branch) + return PushResponse(message="Pushed to remote") + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) + + +class MergeRequest(BaseModel): + source_branch: str + target_branch: str | None = None + message: str | None = None + + +class MergeResponse(BaseModel): + commit_hash: str + message: str + + +@router.post("/{project_id}/repositories/{repo_id}/merge", response_model=MergeResponse) +async def merge_repository_branches( + project_id: uuid.UUID, + repo_id: uuid.UUID, + data: MergeRequest, + user_id: uuid.UUID = Depends(get_current_user_id), + session: AsyncSession = Depends(get_db_session), +) -> MergeResponse: + """Merge branches.""" + _user = await _get_user(session, user_id) + _project = await _get_owned_project(project_id, user_id, session) + + repo = await session.get(GitRepository, repo_id) + if repo is None or repo.project_id != project_id: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found") + + if not os.path.exists(repo.path): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="repository not found on disk") + + try: + commit_hash = merge( + repo_path=repo.path, + source_branch=data.source_branch, + target_branch=data.target_branch, + message=data.message, + ) + return MergeResponse( + commit_hash=commit_hash, + message=data.message or f"Merge {data.source_branch}", + ) + except RuntimeError as e: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e)) diff --git a/apps/api/src/utils/git_control.py b/apps/api/src/utils/git_control.py new file mode 100644 index 0000000..9b5a85a --- /dev/null +++ b/apps/api/src/utils/git_control.py @@ -0,0 +1,282 @@ +"""Git control utilities for repository operations.""" + +import subprocess +from dataclasses import dataclass, field +from typing import Any + + +def _run_git_command(repo_path: str, *args: str) -> str: + """Run a git command in the repository directory.""" + result = subprocess.run( + ["git", *args], + cwd=repo_path, + capture_output=True, + text=True, + ) + if result.returncode != 0: + raise RuntimeError(f"Git command failed: {result.stderr}") + return result.stdout + + +@dataclass +class GitStatus: + """Represents the working directory status.""" + + branch: str + modified: list[str] = field(default_factory=list) + added: list[str] = field(default_factory=list) + deleted: list[str] = field(default_factory=list) + untracked: list[str] = field(default_factory=list) + renamed: list[str] = field(default_factory=list) + ahead: int = 0 + behind: int = 0 + + +def get_status(repo_path: str) -> GitStatus: + """Get the working directory status. + + Args: + repo_path: Path to the git repository + + Returns: + GitStatus with changes + """ + # Get current branch + try: + branch = _run_git_command(repo_path, "rev-parse", "--abbrev-ref", "HEAD").strip() + except RuntimeError: + branch = "HEAD" + + status = GitStatus(branch=branch) + + # Get status with porcelain format + try: + output = _run_git_command(repo_path, "status", "--porcelain", "--branch") + except RuntimeError: + return status + + for line in output.strip().split("\n"): + if not line: + continue + + # Branch info line starts with ## + if line.startswith("## "): + branch_info = line[3:] + # Parse ahead/behind info + if "[ahead " in branch_info: + ahead_str = branch_info.split("[ahead ")[1].split("]")[0] + status.ahead = int(ahead_str.split(",")[0]) + if "[behind " in branch_info: + behind_str = branch_info.split("[behind ")[1].split("]")[0] + status.behind = int(behind_str.split(",")[0]) + continue + + # Parse status code + if len(line) < 3: + continue + + index_status = line[0] + worktree_status = line[1] + filename = line[3:] + + # Untracked files + if index_status == "?" and worktree_status == "?": + status.untracked.append(filename) + continue + + # Added files + if index_status == "A" or worktree_status == "A": + status.added.append(filename) + continue + + # Deleted files + if index_status == "D" or worktree_status == "D": + status.deleted.append(filename) + continue + + # Renamed files + if index_status == "R" or worktree_status == "R": + status.renamed.append(filename) + continue + + # Modified files + if index_status == "M" or worktree_status == "M": + status.modified.append(filename) + continue + + return status + + +def create_branch(repo_path: str, name: str, base_branch: str = "HEAD") -> None: + """Create a new branch. + + Args: + repo_path: Path to the git repository + name: Branch name + base_branch: Base branch to create from (default: HEAD) + + Raises: + RuntimeError: If branch creation fails + """ + _run_git_command(repo_path, "branch", name, base_branch) + + +def delete_branch(repo_path: str, name: str, force: bool = False) -> None: + """Delete a branch. + + Args: + repo_path: Path to the git repository + name: Branch name + force: Force delete even if not merged + + Raises: + RuntimeError: If branch deletion fails + """ + flag = "-D" if force else "-d" + _run_git_command(repo_path, "branch", flag, name) + + +def checkout_branch(repo_path: str, name: str) -> None: + """Checkout a branch. + + Args: + repo_path: Path to the git repository + name: Branch name + + Raises: + RuntimeError: If checkout fails + """ + _run_git_command(repo_path, "checkout", name) + + +def commit_changes( + repo_path: str, + message: str, + author_name: str, + author_email: str, + files: list[str] | None = None, +) -> str: + """Commit changes to the repository. + + Args: + repo_path: Path to the git repository + message: Commit message + author_name: Author name + author_email: Author email + files: Specific files to commit (None = all staged) + + Returns: + Commit hash + + Raises: + RuntimeError: If commit fails + """ + # Stage files if specified + if files: + for file in files: + _run_git_command(repo_path, "add", file) + else: + _run_git_command(repo_path, "add", "-A") + + # Commit + _run_git_command( + repo_path, + "commit", + "-m", + message, + f"--author={author_name} <{author_email}>", + ) + + # Return commit hash + return _run_git_command(repo_path, "rev-parse", "HEAD").strip() + + +def fetch(repo_path: str) -> None: + """Fetch from remote. + + Args: + repo_path: Path to the git repository + + Raises: + RuntimeError: If fetch fails + """ + _run_git_command(repo_path, "fetch", "--all") + + +def pull(repo_path: str, branch: str | None = None) -> None: + """Pull updates from remote. + + Args: + repo_path: Path to the git repository + branch: Branch to pull (default: current branch) + + Raises: + RuntimeError: If pull fails + """ + args = ["pull"] + if branch: + args.extend(["origin", branch]) + _run_git_command(repo_path, *args) + + +def push(repo_path: str, branch: str | None = None) -> None: + """Push changes to remote. + + Args: + repo_path: Path to the git repository + branch: Branch to push (default: current branch) + + Raises: + RuntimeError: If push fails + """ + args = ["push"] + if branch: + args.extend(["origin", branch]) + _run_git_command(repo_path, *args) + + +def merge( + repo_path: str, + source_branch: str, + target_branch: str | None = None, + message: str | None = None, +) -> str: + """Merge a branch into the current branch. + + Args: + repo_path: Path to the git repository + source_branch: Branch to merge from + target_branch: Branch to merge into (default: current branch) + message: Merge commit message + + Returns: + Merge commit hash + + Raises: + RuntimeError: If merge fails (including conflicts) + """ + # Checkout target branch if specified + if target_branch: + checkout_branch(repo_path, target_branch) + + # Merge + args = ["merge", source_branch] + if message: + args.extend(["-m", message]) + + _run_git_command(repo_path, *args) + + # Return merge commit hash + return _run_git_command(repo_path, "rev-parse", "HEAD").strip() + + +def get_current_branch(repo_path: str) -> str: + """Get the current branch name. + + Args: + repo_path: Path to the git repository + + Returns: + Current branch name + """ + return _run_git_command(repo_path, "rev-parse", "--abbrev-ref", "HEAD").strip()