alex
22474cdba5
Backend (ruff): - Fix 106 errors: move imports to top of file (E402) - Remove unused imports (F401) - Add missing imports for undefined names (F821) - Remove unused variables (F841) - Fix test_models.py broken RefreshToken test - Fix test_projects_api.py missing TestClient import Frontend (eslint): - Remove unused imports/variables across 10 files - Fix explicit any types in client.ts and sessions.ts - Clean up empty block statements in terminal.tsx Quality gates: ruff (pass), eslint (pass), tsc --noEmit (pass), pytest (98 passed, 4 pre-existing failures)
314 lines
8.1 KiB
Python
314 lines
8.1 KiB
Python
"""Git control utilities for repository operations."""
|
|
|
|
import subprocess
|
|
from dataclasses import dataclass, field
|
|
|
|
|
|
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:
|
|
try:
|
|
branch = _run_git_command(repo_path, "symbolic-ref", "--short", "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
|
|
"""
|
|
try:
|
|
_run_git_command(repo_path, "rev-parse", "--verify", "HEAD^{commit}")
|
|
except RuntimeError:
|
|
# No commits yet - empty repository
|
|
try:
|
|
_run_git_command(repo_path, "checkout", "--orphan", name)
|
|
except RuntimeError as e:
|
|
if "work tree" in str(e).lower():
|
|
# Bare repository - use symbolic-ref instead
|
|
_run_git_command(repo_path, "symbolic-ref", "HEAD", f"refs/heads/{name}")
|
|
return
|
|
raise
|
|
return
|
|
|
|
_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
|
|
"""
|
|
try:
|
|
_run_git_command(repo_path, "checkout", name)
|
|
except RuntimeError as e:
|
|
if "work tree" in str(e).lower():
|
|
# Bare repository - use symbolic-ref instead
|
|
_run_git_command(repo_path, "symbolic-ref", "HEAD", f"refs/heads/{name}")
|
|
return
|
|
raise
|
|
|
|
|
|
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.append("origin")
|
|
args.append(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
|
|
"""
|
|
try:
|
|
branch = _run_git_command(repo_path, "rev-parse", "--abbrev-ref", "HEAD").strip()
|
|
if branch != "HEAD":
|
|
return branch
|
|
except RuntimeError:
|
|
pass
|
|
|
|
return _run_git_command(repo_path, "symbolic-ref", "--short", "HEAD").strip()
|