feat: add git control endpoints (status, branch, commit, fetch, pull, push, merge)
- 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 ✓
This commit is contained in:
Fusion
@@ -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()
|
||||
Reference in New Issue
Block a user