openenv/coding_env

1#!/usr/bin/env python32"""3Git Server Client for connecting to external Gitea instance.4 5This module provides a lightweight client for interacting with a shared6Gitea service, optimized for task-based isolation where multiple environment7instances share the same Gitea server but have isolated workspaces.8"""9 10import json11import os12import shutil13import subprocess14import time15from dataclasses import dataclass16from pathlib import Path17from urllib.parse import urlparse18 19 20@dataclass21class RepoInfo:22    """Information about a repository."""23 24    name: str25    url: str26    commit: str27    clone_url: str28 29 30class GitServerClient:31    """32    Client for connecting to an external Gitea server.33 34    This client is optimized for task-based isolation where:35    - Multiple tasks share the same Gitea instance36    - Each task has its own isolated workspace37    - Fast reset() via git operations (no server restart)38    - Repos are pre-migrated to Gitea once39 40    Args:41        gitea_url: URL of the Gitea server (e.g., "http://gitea:3000")42        username: Gitea username for authentication43        password: Gitea password for authentication44        workspace_dir: Local workspace directory for cloning repos45 46    Example:47        >>> # Connect to shared Gitea (credentials from environment)48        >>> import os49        >>> client = GitServerClient(50        ...     gitea_url=os.getenv("GITEA_URL"),51        ...     username=os.getenv("GITEA_USERNAME"),52        ...     password=os.getenv("GITEA_PASSWORD")53        ... )54        >>> client.wait_for_ready()55        >>> # Clone repo to workspace56        >>> path = client.clone_to_workspace("my-repo", commit="abc123")57        >>> # Fast reset to base state58        >>> client.reset_workspace("my-repo", commit="abc123")59    """60 61    def __init__(62        self,63        gitea_url: str,64        username: str,65        password: str,66        workspace_dir: str = "/workspace",67    ):68        """Initialize Git Server Client."""69        self.gitea_url = gitea_url.rstrip("/")70        self.username = username71        self.password = password72        self.workspace_dir = Path(workspace_dir)73        self.is_ready = False74 75        # Parse Gitea URL76        parsed = urlparse(self.gitea_url)77        self.domain = parsed.hostname or "localhost"78        self.port = parsed.port or 300079 80        # Ensure workspace exists81        os.makedirs(self.workspace_dir, exist_ok=True)82 83        # Configure git credentials84        self._configure_git()85 86    def _configure_git(self):87        """Configure git credentials for automatic authentication."""88        home_dir = Path.home()89 90        # Git config91        git_config = f"""[user]92    name = {self.username}93    email = {self.username}@local.env94[init]95    defaultBranch = main96[credential]97    helper = store98"""99        gitconfig_path = home_dir / ".gitconfig"100        gitconfig_path.write_text(git_config)101 102        # Git credentials103        git_credentials = (104            f"http://{self.username}:{self.password}@{self.domain}:{self.port}\n"105        )106        gitcreds_path = home_dir / ".git-credentials"107        gitcreds_path.write_text(git_credentials)108        gitcreds_path.chmod(0o600)109 110    def wait_for_ready(self, timeout: int = 30) -> bool:111        """112        Wait for Gitea server to be ready.113 114        Args:115            timeout: Maximum seconds to wait116 117        Returns:118            True if server is ready, False otherwise119        """120        start_time = time.time()121        while time.time() - start_time < timeout:122            try:123                result = subprocess.run(124                    ["curl", "-sf", f"{self.gitea_url}/"],125                    capture_output=True,126                    timeout=5,127                )128                if result.returncode == 0:129                    self.is_ready = True130                    return True131            except subprocess.TimeoutExpired:132                pass133            except Exception:134                pass135 136            time.sleep(1)137 138        return False139 140    def list_repositories(self) -> list[dict[str, str]]:141        """142        List all repositories in Gitea.143 144        Returns:145            List of repository information dictionaries146        """147        if not self.is_ready:148            raise RuntimeError("Gitea server is not ready")149 150        result = subprocess.run(151            [152                "curl",153                "-s",154                f"{self.gitea_url}/api/v1/user/repos",155                "-u",156                f"{self.username}:{self.password}",157            ],158            capture_output=True,159            text=True,160        )161 162        if result.returncode != 0:163            return []164 165        try:166            repos = json.loads(result.stdout)167            return [168                {169                    "name": repo["name"],170                    "full_name": repo["full_name"],171                    "clone_url": repo["clone_url"],172                    "description": repo.get("description", ""),173                }174                for repo in repos175            ]176        except (json.JSONDecodeError, KeyError):177            return []178 179    def clone_to_workspace(180        self, repo_name: str, target_dir: str | None = None, commit: str = "main"181    ) -> str:182        """183        Clone a repository to the workspace at a specific commit.184 185        This creates a fresh clone optimized for task isolation.186 187        Args:188            repo_name: Name of repository to clone189            target_dir: Target directory name (defaults to repo_name)190            commit: Commit hash or branch to check out191 192        Returns:193            Path to cloned repository194 195        Raises:196            RuntimeError: If clone fails197        """198        if not self.is_ready:199            raise RuntimeError("Gitea server is not ready")200 201        target_dir = target_dir or repo_name202        target_path = self.workspace_dir / target_dir203 204        # Remove existing directory if present205        if target_path.exists():206            shutil.rmtree(target_path)207 208        clone_url = f"{self.gitea_url}/{self.username}/{repo_name}.git"209 210        # Clone repository211        result = subprocess.run(212            ["git", "clone", clone_url, str(target_path)],213            capture_output=True,214            text=True,215        )216 217        if result.returncode != 0:218            raise RuntimeError(f"Clone failed: {result.stderr}")219 220        # Checkout specific commit221        if commit != "main":222            result = subprocess.run(223                ["git", "checkout", commit],224                cwd=str(target_path),225                capture_output=True,226                text=True,227            )228 229            if result.returncode != 0:230                raise RuntimeError(f"Checkout failed: {result.stderr}")231 232        return str(target_path)233 234    def reset_workspace(self, repo_name: str, commit: str = "main") -> bool:235        """236        Fast reset of workspace to base state (optimized for task resets).237 238        This is much faster than re-cloning. It:239        1. Checks out the target commit240        2. Resets to that commit (hard)241        3. Cleans untracked files242 243        Args:244            repo_name: Name of repository (directory in workspace)245            commit: Commit hash or branch to reset to246 247        Returns:248            True if reset successful249 250        Raises:251            RuntimeError: If reset fails252        """253        repo_path = self.workspace_dir / repo_name254 255        if not repo_path.exists():256            raise RuntimeError(f"Repository not found in workspace: {repo_name}")257 258        # Fetch latest (in case commit is new)259        subprocess.run(260            ["git", "fetch", "--all"],261            cwd=str(repo_path),262            capture_output=True,263        )264 265        # Checkout and hard reset to commit266        result = subprocess.run(267            ["git", "checkout", commit],268            cwd=str(repo_path),269            capture_output=True,270            text=True,271        )272 273        if result.returncode != 0:274            raise RuntimeError(f"Checkout failed: {result.stderr}")275 276        result = subprocess.run(277            [278                "git",279                "reset",280                "--hard",281                f"origin/{commit}" if commit != "main" else commit,282            ],283            cwd=str(repo_path),284            capture_output=True,285            text=True,286        )287 288        if result.returncode != 0:289            # Try without origin/ prefix290            result = subprocess.run(291                ["git", "reset", "--hard", commit],292                cwd=str(repo_path),293                capture_output=True,294                text=True,295            )296            if result.returncode != 0:297                raise RuntimeError(f"Reset failed: {result.stderr}")298 299        # Clean untracked files and directories300        subprocess.run(301            ["git", "clean", "-fdx"],302            cwd=str(repo_path),303            capture_output=True,304        )305 306        return True307 308    def execute_git_command(309        self, command: str, working_dir: str = ""310    ) -> tuple[int, str, str]:311        """312        Execute a git command in the workspace.313 314        Args:315            command: Git command to execute (without 'git' prefix)316            working_dir: Working directory relative to workspace317 318        Returns:319            Tuple of (exit_code, stdout, stderr)320        """321        work_path = (322            self.workspace_dir / working_dir if working_dir else self.workspace_dir323        )324 325        if not work_path.exists():326            return (1, "", f"Working directory does not exist: {work_path}")327 328        # Split command safely329        cmd_parts = ["git"] + command.split()330 331        result = subprocess.run(332            cmd_parts,333            cwd=str(work_path),334            capture_output=True,335            text=True,336        )337 338        return (result.returncode, result.stdout, result.stderr)339 340    def get_current_commit(self, repo_name: str) -> str:341        """342        Get current commit hash of a workspace repository.343 344        Args:345            repo_name: Name of repository in workspace346 347        Returns:348            Commit hash349        """350        repo_path = self.workspace_dir / repo_name351 352        if not repo_path.exists():353            raise RuntimeError(f"Repository not found: {repo_name}")354 355        result = subprocess.run(356            ["git", "rev-parse", "HEAD"],357            cwd=str(repo_path),358            capture_output=True,359            text=True,360        )361 362        if result.returncode != 0:363            raise RuntimeError(f"Failed to get commit: {result.stderr}")364 365        return result.stdout.strip()366 367    def workspace_exists(self, repo_name: str) -> bool:368        """Check if a repository exists in workspace."""369        return (self.workspace_dir / repo_name).exists()370