alex
37ccaa4fdc
Service organization (19 files moved into 6 subpackages): - services/instance/ — event_bus, health_monitor, lifecycle_hooks - services/config/ — config_profile_resolver - services/git/ — clone, git_operations, git_service - services/build/ — docker_build, manifest_compiler - services/terminal/ — terminal_manager, terminal_session - services/shared/ — correlation, file_service, notification_service, permission_fixer, readiness_probe, ssh_keys, tunnel, workspace_manager API router organization (16 files moved into 6 subpackages): - api/tool/ — tool_instances, tool_types, tool_definitions, tool_types_validation, sessions (extracted from tool_instances) - api/config/ — config_profiles, user_config - api/workspace/ — workspaces, workspace_files, workspace_git, workspace_instances - api/user/ — users, auth, ssh_keys - api/project/ — projects, git_repositories - api/system/ — health, events, notifications, dashboard, terminal, instance_proxy Updated main.py imports and all __init__.py re-exports. Sessions router extracted from tool_instances.py into api/tool/sessions.py. Quality gates: py_compile passed, ruff passed.
98 lines
2.6 KiB
Python
98 lines
2.6 KiB
Python
"""Clone service for repository cloning and dirty state checking."""
|
|
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def clone_repository(
|
|
remote_url: str,
|
|
ssh_key_path: str | None,
|
|
instance_dir: str,
|
|
branch: str = "main",
|
|
) -> str:
|
|
"""Clone a git repository into the instance directory.
|
|
|
|
Args:
|
|
remote_url: Git remote URL (SSH or HTTPS)
|
|
ssh_key_path: Path to SSH private key for authentication (optional)
|
|
instance_dir: Path to instance directory
|
|
branch: Branch to clone (default: main)
|
|
|
|
Returns:
|
|
Path to the cloned repository
|
|
"""
|
|
clone_path = Path(instance_dir) / "repo-clone"
|
|
clone_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
env = os.environ.copy()
|
|
if ssh_key_path:
|
|
# Use SSH key for cloning
|
|
env["GIT_SSH_COMMAND"] = f"ssh -i {ssh_key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null"
|
|
|
|
cmd = [
|
|
"git",
|
|
"clone",
|
|
"--branch", branch,
|
|
"--single-branch",
|
|
remote_url,
|
|
str(clone_path),
|
|
]
|
|
|
|
logger.debug("Cloning repository %s (branch: %s) into %s", remote_url, branch, clone_path)
|
|
result = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
env=env,
|
|
timeout=300,
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
logger.error("Git clone failed: %s", result.stderr)
|
|
raise RuntimeError(f"Failed to clone repository: {result.stderr}")
|
|
|
|
logger.debug("Successfully cloned repository into %s", clone_path)
|
|
return str(clone_path)
|
|
|
|
|
|
def check_dirty_state(clone_path: str) -> tuple[bool, list[str]]:
|
|
"""Check for uncommitted changes in a cloned repository.
|
|
|
|
Args:
|
|
clone_path: Path to the cloned repository
|
|
|
|
Returns:
|
|
Tuple of (is_dirty, list_of_changed_files)
|
|
"""
|
|
result = subprocess.run(
|
|
["git", "-C", clone_path, "status", "--short"],
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
|
|
if result.returncode != 0:
|
|
logger.warning("Failed to check git status: %s", result.stderr)
|
|
return False, []
|
|
|
|
changed_files = [line.strip() for line in result.stdout.split("\n") if line.strip()]
|
|
is_dirty = len(changed_files) > 0
|
|
|
|
return is_dirty, changed_files
|
|
|
|
|
|
def remove_clone_directory(instance_dir: str) -> None:
|
|
"""Remove the cloned repository from the instance directory.
|
|
|
|
Args:
|
|
instance_dir: Path to instance directory
|
|
"""
|
|
clone_path = Path(instance_dir) / "repo-clone"
|
|
if clone_path.exists():
|
|
import shutil
|
|
shutil.rmtree(clone_path)
|
|
logger.debug("Removed clone directory: %s", clone_path)
|