feat: add repository workspace as default project view
- Create git file utilities (list_tree, get_file_content, list_branches, commit_file) - Add file browsing API endpoints (list, content, branches, update) - Create RepoWorkspace page with sidebar + main content layout - Add FileTree component with directory navigation - Add FileViewer component for viewing file contents - Update project list to link to workspace - Add workspace CSS styles - Update router with workspace route Quality gates: ruff ✓, mypy ✓, typecheck ✓, build ✓
This commit is contained in:
Fusion
@@ -14,6 +14,12 @@ from src.config import Settings
|
||||
from src.models.git_repository import GitRepository
|
||||
from src.models.project import Project
|
||||
from src.models.user import User
|
||||
from src.utils.git_files import (
|
||||
commit_file,
|
||||
get_file_content,
|
||||
list_branches,
|
||||
list_tree,
|
||||
)
|
||||
from src.utils.git_history import get_commit_detail, get_commit_history
|
||||
from src.utils.git_url_parser import parse_git_url
|
||||
|
||||
@@ -265,3 +271,198 @@ async def get_repository_commit(
|
||||
return detail
|
||||
except (RuntimeError, ValueError) as e:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
||||
|
||||
|
||||
# File browsing endpoints
|
||||
|
||||
|
||||
class FileListResponse(BaseModel):
|
||||
path: str
|
||||
branch: str
|
||||
entries: list[dict]
|
||||
|
||||
|
||||
class FileContentResponse(BaseModel):
|
||||
path: str
|
||||
branch: str
|
||||
content: str
|
||||
size: int
|
||||
encoding: str
|
||||
language: str | None
|
||||
is_binary: bool
|
||||
last_commit: dict | None
|
||||
|
||||
|
||||
class BranchesResponse(BaseModel):
|
||||
branches: list[dict]
|
||||
default_branch: str
|
||||
|
||||
|
||||
class FileUpdateRequest(BaseModel):
|
||||
path: str
|
||||
branch: str
|
||||
content: str
|
||||
commit_message: str
|
||||
|
||||
|
||||
class FileUpdateResponse(BaseModel):
|
||||
commit_hash: str
|
||||
message: str
|
||||
branch: str
|
||||
|
||||
|
||||
@router.get("/{project_id}/repositories/{repo_id}/files", response_model=FileListResponse)
|
||||
async def list_repository_files(
|
||||
project_id: uuid.UUID,
|
||||
repo_id: uuid.UUID,
|
||||
branch: str = "main",
|
||||
path: str = "",
|
||||
user_id: uuid.UUID = Depends(get_current_user_id),
|
||||
session: AsyncSession = Depends(get_db_session),
|
||||
) -> FileListResponse:
|
||||
"""List files and directories in a repository path."""
|
||||
_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:
|
||||
entries = list_tree(repo.path, branch=branch, path=path)
|
||||
return FileListResponse(
|
||||
path=path,
|
||||
branch=branch,
|
||||
entries=[
|
||||
{
|
||||
"name": e.name,
|
||||
"type": e.type,
|
||||
"path": e.path,
|
||||
"size": e.size,
|
||||
"mode": e.mode,
|
||||
"last_commit": e.last_commit,
|
||||
}
|
||||
for e in entries
|
||||
],
|
||||
)
|
||||
except RuntimeError as e:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
||||
|
||||
|
||||
@router.get("/{project_id}/repositories/{repo_id}/files/content", response_model=FileContentResponse)
|
||||
async def get_repository_file_content(
|
||||
project_id: uuid.UUID,
|
||||
repo_id: uuid.UUID,
|
||||
branch: str,
|
||||
path: str,
|
||||
user_id: uuid.UUID = Depends(get_current_user_id),
|
||||
session: AsyncSession = Depends(get_db_session),
|
||||
) -> FileContentResponse:
|
||||
"""Get the content of a file."""
|
||||
_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:
|
||||
file_content = get_file_content(repo.path, branch=branch, path=path)
|
||||
return FileContentResponse(
|
||||
path=file_content.path,
|
||||
branch=file_content.branch,
|
||||
content=file_content.content,
|
||||
size=file_content.size,
|
||||
encoding=file_content.encoding,
|
||||
language=file_content.language,
|
||||
is_binary=file_content.is_binary,
|
||||
last_commit=file_content.last_commit,
|
||||
)
|
||||
except FileNotFoundError:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="file not found")
|
||||
except RuntimeError as e:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
||||
|
||||
|
||||
@router.get("/{project_id}/repositories/{repo_id}/branches", response_model=BranchesResponse)
|
||||
async def get_repository_branches(
|
||||
project_id: uuid.UUID,
|
||||
repo_id: uuid.UUID,
|
||||
user_id: uuid.UUID = Depends(get_current_user_id),
|
||||
session: AsyncSession = Depends(get_db_session),
|
||||
) -> BranchesResponse:
|
||||
"""List all branches in 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")
|
||||
|
||||
try:
|
||||
branches, default_branch = list_branches(repo.path)
|
||||
return BranchesResponse(
|
||||
branches=[
|
||||
{
|
||||
"name": b.name,
|
||||
"is_default": b.is_default,
|
||||
"last_commit": b.last_commit,
|
||||
}
|
||||
for b in branches
|
||||
],
|
||||
default_branch=default_branch,
|
||||
)
|
||||
except RuntimeError as e:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
||||
|
||||
|
||||
@router.post("/{project_id}/repositories/{repo_id}/files/content", response_model=FileUpdateResponse)
|
||||
async def update_repository_file(
|
||||
project_id: uuid.UUID,
|
||||
repo_id: uuid.UUID,
|
||||
data: FileUpdateRequest,
|
||||
user_id: uuid.UUID = Depends(get_current_user_id),
|
||||
session: AsyncSession = Depends(get_db_session),
|
||||
) -> FileUpdateResponse:
|
||||
"""Update a file and create a commit."""
|
||||
_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_file(
|
||||
repo_path=repo.path,
|
||||
branch=data.branch,
|
||||
path=data.path,
|
||||
content=data.content,
|
||||
commit_message=data.commit_message,
|
||||
author_name=author_name,
|
||||
author_email=author_email,
|
||||
)
|
||||
return FileUpdateResponse(
|
||||
commit_hash=commit_hash,
|
||||
message=data.commit_message,
|
||||
branch=data.branch,
|
||||
)
|
||||
except RuntimeError as e:
|
||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(e))
|
||||
|
||||
@@ -0,0 +1,391 @@
|
||||
"""Git file utilities for browsing repository contents."""
|
||||
|
||||
import subprocess
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class FileTreeEntry:
|
||||
"""Represents a file or directory in the repository."""
|
||||
|
||||
name: str
|
||||
type: str # "file" or "directory"
|
||||
path: str
|
||||
size: int | None = None
|
||||
mode: str | None = None
|
||||
last_commit: dict[str, Any] | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class BranchInfo:
|
||||
"""Represents a git branch."""
|
||||
|
||||
name: str
|
||||
is_default: bool
|
||||
last_commit: dict[str, Any] | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class FileContent:
|
||||
"""Represents file content and metadata."""
|
||||
|
||||
path: str
|
||||
branch: str
|
||||
content: str
|
||||
size: int
|
||||
encoding: str
|
||||
language: str | None
|
||||
is_binary: bool
|
||||
last_commit: dict[str, Any] | None = None
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
def list_tree(repo_path: str, branch: str = "main", path: str = "") -> list[FileTreeEntry]:
|
||||
"""List files and directories in a repository path.
|
||||
|
||||
Args:
|
||||
repo_path: Path to the git repository
|
||||
branch: Branch name to list from
|
||||
path: Directory path within the repository (empty for root)
|
||||
|
||||
Returns:
|
||||
List of FileTreeEntry objects
|
||||
"""
|
||||
tree_path = f"{branch}:{path}" if path else branch
|
||||
|
||||
try:
|
||||
output = _run_git_command(repo_path, "ls-tree", "-l", tree_path)
|
||||
except RuntimeError:
|
||||
# Try with HEAD if branch doesn't exist
|
||||
tree_path = f"HEAD:{path}" if path else "HEAD"
|
||||
output = _run_git_command(repo_path, "ls-tree", "-l", tree_path)
|
||||
|
||||
entries = []
|
||||
for line in output.strip().split("\n"):
|
||||
if not line:
|
||||
continue
|
||||
|
||||
# Format: <mode> <type> <hash> <size>\t<name>
|
||||
parts = line.split("\t", 1)
|
||||
if len(parts) != 2:
|
||||
continue
|
||||
|
||||
meta, name = parts
|
||||
meta_parts = meta.split()
|
||||
if len(meta_parts) < 4:
|
||||
continue
|
||||
|
||||
mode = meta_parts[0]
|
||||
obj_type = meta_parts[1]
|
||||
_ = meta_parts[2] # object hash, not used
|
||||
size = int(meta_parts[3]) if obj_type == "blob" else None
|
||||
|
||||
entry_path = f"{path}/{name}" if path else name
|
||||
|
||||
# Get last commit info for this entry
|
||||
last_commit = _get_last_commit_for_path(repo_path, branch, entry_path)
|
||||
|
||||
entries.append(
|
||||
FileTreeEntry(
|
||||
name=name,
|
||||
type="directory" if obj_type == "tree" else "file",
|
||||
path=entry_path,
|
||||
size=size,
|
||||
mode=mode,
|
||||
last_commit=last_commit,
|
||||
)
|
||||
)
|
||||
|
||||
return entries
|
||||
|
||||
|
||||
def _get_last_commit_for_path(repo_path: str, branch: str, path: str) -> dict[str, Any] | None:
|
||||
"""Get the last commit that modified a path."""
|
||||
try:
|
||||
output = _run_git_command(
|
||||
repo_path,
|
||||
"log",
|
||||
"-1",
|
||||
"--format=%H|%s|%an|%aI",
|
||||
branch,
|
||||
"--",
|
||||
path,
|
||||
)
|
||||
if not output.strip():
|
||||
return None
|
||||
|
||||
parts = output.strip().split("|", 3)
|
||||
if len(parts) != 4:
|
||||
return None
|
||||
|
||||
return {
|
||||
"hash": parts[0],
|
||||
"message": parts[1],
|
||||
"author": parts[2],
|
||||
"date": parts[3],
|
||||
}
|
||||
except RuntimeError:
|
||||
return None
|
||||
|
||||
|
||||
def get_file_content(repo_path: str, branch: str, path: str) -> FileContent:
|
||||
"""Get the content of a file.
|
||||
|
||||
Args:
|
||||
repo_path: Path to the git repository
|
||||
branch: Branch name
|
||||
path: File path within the repository
|
||||
|
||||
Returns:
|
||||
FileContent with content and metadata
|
||||
"""
|
||||
# Check if file exists
|
||||
try:
|
||||
_run_git_command(repo_path, "cat-file", "-e", f"{branch}:{path}")
|
||||
except RuntimeError:
|
||||
raise FileNotFoundError(f"File '{path}' not found in branch '{branch}'")
|
||||
|
||||
# Get file size
|
||||
size_output = _run_git_command(repo_path, "cat-file", "-s", f"{branch}:{path}")
|
||||
size = int(size_output.strip())
|
||||
|
||||
# Check if binary
|
||||
is_binary = _is_binary_file(repo_path, branch, path)
|
||||
|
||||
# Get content (only for text files)
|
||||
content = ""
|
||||
if not is_binary:
|
||||
content = _run_git_command(repo_path, "show", f"{branch}:{path}")
|
||||
|
||||
# Detect language from extension
|
||||
language = _detect_language(path)
|
||||
|
||||
# Get last commit
|
||||
last_commit = _get_last_commit_for_path(repo_path, branch, path)
|
||||
|
||||
return FileContent(
|
||||
path=path,
|
||||
branch=branch,
|
||||
content=content,
|
||||
size=size,
|
||||
encoding="utf-8",
|
||||
language=language,
|
||||
is_binary=is_binary,
|
||||
last_commit=last_commit,
|
||||
)
|
||||
|
||||
|
||||
def _is_binary_file(repo_path: str, branch: str, path: str) -> bool:
|
||||
"""Check if a file is binary."""
|
||||
try:
|
||||
_run_git_command(
|
||||
repo_path,
|
||||
"diff",
|
||||
"--numstat",
|
||||
"--",
|
||||
path,
|
||||
)
|
||||
# Alternative: use git show and check for null bytes
|
||||
content = _run_git_command(repo_path, "show", f"{branch}:{path}")
|
||||
return b"\x00" in content.encode("utf-8", errors="replace")
|
||||
except RuntimeError:
|
||||
return True
|
||||
|
||||
|
||||
def _detect_language(path: str) -> str | None:
|
||||
"""Detect programming language from file extension."""
|
||||
ext = Path(path).suffix.lower()
|
||||
language_map = {
|
||||
".py": "python",
|
||||
".js": "javascript",
|
||||
".ts": "typescript",
|
||||
".jsx": "jsx",
|
||||
".tsx": "tsx",
|
||||
".html": "html",
|
||||
".css": "css",
|
||||
".scss": "scss",
|
||||
".json": "json",
|
||||
".md": "markdown",
|
||||
".yaml": "yaml",
|
||||
".yml": "yaml",
|
||||
".sh": "bash",
|
||||
".rs": "rust",
|
||||
".go": "go",
|
||||
".java": "java",
|
||||
".c": "c",
|
||||
".cpp": "cpp",
|
||||
".h": "c",
|
||||
".php": "php",
|
||||
".rb": "ruby",
|
||||
".sql": "sql",
|
||||
".dockerfile": "dockerfile",
|
||||
".vue": "vue",
|
||||
".svelte": "svelte",
|
||||
}
|
||||
return language_map.get(ext)
|
||||
|
||||
|
||||
def list_branches(repo_path: str) -> tuple[list[BranchInfo], str]:
|
||||
"""List all branches and identify the default branch.
|
||||
|
||||
Args:
|
||||
repo_path: Path to the git repository
|
||||
|
||||
Returns:
|
||||
Tuple of (list of BranchInfo, default branch name)
|
||||
"""
|
||||
# Get all branches
|
||||
output = _run_git_command(repo_path, "branch", "-a", "--format=%(refname:short)")
|
||||
|
||||
branches: list[BranchInfo] = []
|
||||
default_branch = "main"
|
||||
|
||||
for line in output.strip().split("\n"):
|
||||
if not line:
|
||||
continue
|
||||
|
||||
branch_name = line.strip()
|
||||
# Skip remote tracking branches (they start with remotes/)
|
||||
if branch_name.startswith("remotes/"):
|
||||
# Extract just the branch name part
|
||||
parts = branch_name.split("/", 2)
|
||||
if len(parts) >= 3:
|
||||
branch_name = parts[2]
|
||||
else:
|
||||
continue
|
||||
|
||||
# Skip duplicates
|
||||
if any(b.name == branch_name for b in branches):
|
||||
continue
|
||||
|
||||
# Check if this is the default branch (HEAD points to it)
|
||||
try:
|
||||
head_output = _run_git_command(
|
||||
repo_path,
|
||||
"symbolic-ref",
|
||||
"HEAD",
|
||||
)
|
||||
if head_output.strip() == f"refs/heads/{branch_name}":
|
||||
default_branch = branch_name
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
# Get last commit for branch
|
||||
last_commit = _get_last_commit_for_path(repo_path, branch_name, ".")
|
||||
|
||||
branches.append(
|
||||
BranchInfo(
|
||||
name=branch_name,
|
||||
is_default=(branch_name == default_branch),
|
||||
last_commit=last_commit,
|
||||
)
|
||||
)
|
||||
|
||||
# If no branches found, try to get HEAD
|
||||
if not branches:
|
||||
try:
|
||||
output = _run_git_command(repo_path, "rev-parse", "--abbrev-ref", "HEAD")
|
||||
branch_name = output.strip()
|
||||
if branch_name and branch_name != "HEAD":
|
||||
last_commit = _get_last_commit_for_path(repo_path, branch_name, ".")
|
||||
branches.append(
|
||||
BranchInfo(
|
||||
name=branch_name,
|
||||
is_default=True,
|
||||
last_commit=last_commit,
|
||||
)
|
||||
)
|
||||
default_branch = branch_name
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
return branches, default_branch
|
||||
|
||||
|
||||
def commit_file(
|
||||
repo_path: str,
|
||||
branch: str,
|
||||
path: str,
|
||||
content: str,
|
||||
commit_message: str,
|
||||
author_name: str,
|
||||
author_email: str,
|
||||
) -> str:
|
||||
"""Commit a file change.
|
||||
|
||||
Args:
|
||||
repo_path: Path to the git repository
|
||||
branch: Branch to commit to
|
||||
path: File path within the repository
|
||||
content: New file content
|
||||
commit_message: Commit message
|
||||
author_name: Author name
|
||||
author_email: Author email
|
||||
|
||||
Returns:
|
||||
Commit hash
|
||||
"""
|
||||
# For bare repositories, we need to use git commands differently
|
||||
# We'll create a temporary worktree, make changes, and commit
|
||||
|
||||
import tempfile
|
||||
import os
|
||||
|
||||
# Create a temporary worktree
|
||||
with tempfile.TemporaryDirectory() as worktree_path:
|
||||
# Add worktree
|
||||
_run_git_command(
|
||||
repo_path,
|
||||
"worktree",
|
||||
"add",
|
||||
"--detach",
|
||||
worktree_path,
|
||||
branch,
|
||||
)
|
||||
|
||||
try:
|
||||
# Write file content
|
||||
file_path = os.path.join(worktree_path, path)
|
||||
os.makedirs(os.path.dirname(file_path), exist_ok=True)
|
||||
with open(file_path, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
|
||||
# Configure git author
|
||||
_run_git_command(worktree_path, "config", "user.name", author_name)
|
||||
_run_git_command(worktree_path, "config", "user.email", author_email)
|
||||
|
||||
# Stage and commit
|
||||
_run_git_command(worktree_path, "add", path)
|
||||
_run_git_command(
|
||||
worktree_path,
|
||||
"commit",
|
||||
"-m",
|
||||
commit_message,
|
||||
)
|
||||
|
||||
# Get commit hash
|
||||
commit_hash = _run_git_command(
|
||||
worktree_path,
|
||||
"rev-parse",
|
||||
"HEAD",
|
||||
).strip()
|
||||
|
||||
return commit_hash
|
||||
|
||||
finally:
|
||||
# Remove worktree
|
||||
_run_git_command(repo_path, "worktree", "remove", worktree_path)
|
||||
Reference in New Issue
Block a user