openenv/echo_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    Examples:47 48        ```python49        import os50 51        # Connect to shared Gitea (credentials from environment)52        client = GitServerClient(53            gitea_url=os.getenv("GITEA_URL"),54            username=os.getenv("GITEA_USERNAME"),55            password=os.getenv("GITEA_PASSWORD")56        )57        client.wait_for_ready()58        # Clone repo to workspace59        path = client.clone_to_workspace("my-repo", commit="abc123")60        # Fast reset to base state61        client.reset_workspace("my-repo", commit="abc123")62        ```63    """64 65    def __init__(66        self,67        gitea_url: str,68        username: str,69        password: str,70        workspace_dir: str = "/workspace",71    ):72        """Initialize Git Server Client."""73        self.gitea_url = gitea_url.rstrip("/")74        self.username = username75        self.password = password76        self.workspace_dir = Path(workspace_dir)77        self.is_ready = False78 79        # Parse Gitea URL80        parsed = urlparse(self.gitea_url)81        self.domain = parsed.hostname or "localhost"82        self.port = parsed.port or 300083 84        # Ensure workspace exists85        os.makedirs(self.workspace_dir, exist_ok=True)86 87        # Configure git credentials88        self._configure_git()89 90    def _configure_git(self):91        """Configure git credentials for automatic authentication."""92        home_dir = Path.home()93 94        # Git config95        git_config = f"""[user]96    name = {self.username}97    email = {self.username}@local.env98[init]99    defaultBranch = main100[credential]101    helper = store102"""103        gitconfig_path = home_dir / ".gitconfig"104        gitconfig_path.write_text(git_config)105 106        # Git credentials107        git_credentials = (108            f"http://{self.username}:{self.password}@{self.domain}:{self.port}\n"109        )110        gitcreds_path = home_dir / ".git-credentials"111        gitcreds_path.write_text(git_credentials)112        gitcreds_path.chmod(0o600)113 114    def wait_for_ready(self, timeout: int = 30) -> bool:115        """116        Wait for Gitea server to be ready.117 118        Args:119            timeout (`int`, *optional*, defaults to `30`):120                Maximum seconds to wait.121 122        Returns:123            `bool`: True if server is ready, False otherwise.124        """125        start_time = time.time()126        while time.time() - start_time < timeout:127            try:128                result = subprocess.run(129                    ["curl", "-sf", f"{self.gitea_url}/"],130                    capture_output=True,131                    timeout=5,132                )133                if result.returncode == 0:134                    self.is_ready = True135                    return True136            except subprocess.TimeoutExpired:137                pass138            except Exception:139                pass140 141            time.sleep(1)142 143        return False144 145    def list_repositories(self) -> list[dict[str, str]]:146        """147        List all repositories in Gitea.148 149        Returns:150            `list` of repository information dictionaries.151        """152        if not self.is_ready:153            raise RuntimeError("Gitea server is not ready")154 155        result = subprocess.run(156            [157                "curl",158                "-s",159                f"{self.gitea_url}/api/v1/user/repos",160                "-u",161                f"{self.username}:{self.password}",162            ],163            capture_output=True,164            text=True,165        )166 167        if result.returncode != 0:168            return []169 170        try:171            repos = json.loads(result.stdout)172            return [173                {174                    "name": repo["name"],175                    "full_name": repo["full_name"],176                    "clone_url": repo["clone_url"],177                    "description": repo.get("description", ""),178                }179                for repo in repos180            ]181        except (json.JSONDecodeError, KeyError):182            return []183 184    def clone_to_workspace(185        self, repo_name: str, target_dir: str | None = None, commit: str = "main"186    ) -> str:187        """188        Clone a repository to the workspace at a specific commit.189 190        This creates a fresh clone optimized for task isolation.191 192        Args:193            repo_name (`str`):194                Name of repository to clone.195            target_dir (`str`, *optional*):196                Target directory name. Defaults to `repo_name`.197            commit (`str`, *optional*, defaults to `"main"`):198                Commit hash or branch to check out.199 200        Returns:201            `str`: Path to cloned repository.202 203        Raises:204            RuntimeError: If clone fails.205        """206        if not self.is_ready:207            raise RuntimeError("Gitea server is not ready")208 209        target_dir = target_dir or repo_name210        target_path = self.workspace_dir / target_dir211 212        # Remove existing directory if present213        if target_path.exists():214            shutil.rmtree(target_path)215 216        clone_url = f"{self.gitea_url}/{self.username}/{repo_name}.git"217 218        # Clone repository219        result = subprocess.run(220            ["git", "clone", clone_url, str(target_path)],221            capture_output=True,222            text=True,223        )224 225        if result.returncode != 0:226            raise RuntimeError(f"Clone failed: {result.stderr}")227 228        # Checkout specific commit229        if commit != "main":230            result = subprocess.run(231                ["git", "checkout", commit],232                cwd=str(target_path),233                capture_output=True,234                text=True,235            )236 237            if result.returncode != 0:238                raise RuntimeError(f"Checkout failed: {result.stderr}")239 240        return str(target_path)241 242    def reset_workspace(self, repo_name: str, commit: str = "main") -> bool:243        """244        Fast reset of workspace to base state (optimized for task resets).245 246        This is much faster than re-cloning. It:247        1. Checks out the target commit248        2. Resets to that commit (hard)249        3. Cleans untracked files250 251        Args:252            repo_name (`str`):253                Name of repository (directory in workspace).254            commit (`str`, *optional*, defaults to `"main"`):255                Commit hash or branch to reset to.256 257        Returns:258            `bool`: True if reset successful.259 260        Raises:261            RuntimeError: If reset fails.262        """263        repo_path = self.workspace_dir / repo_name264 265        if not repo_path.exists():266            raise RuntimeError(f"Repository not found in workspace: {repo_name}")267 268        # Fetch latest (in case commit is new)269        subprocess.run(270            ["git", "fetch", "--all"],271            cwd=str(repo_path),272            capture_output=True,273        )274 275        # Checkout and hard reset to commit276        result = subprocess.run(277            ["git", "checkout", commit],278            cwd=str(repo_path),279            capture_output=True,280            text=True,281        )282 283        if result.returncode != 0:284            raise RuntimeError(f"Checkout failed: {result.stderr}")285 286        result = subprocess.run(287            [288                "git",289                "reset",290                "--hard",291                f"origin/{commit}" if commit != "main" else commit,292            ],293            cwd=str(repo_path),294            capture_output=True,295            text=True,296        )297 298        if result.returncode != 0:299            # Try without origin/ prefix300            result = subprocess.run(301                ["git", "reset", "--hard", commit],302                cwd=str(repo_path),303                capture_output=True,304                text=True,305            )306            if result.returncode != 0:307                raise RuntimeError(f"Reset failed: {result.stderr}")308 309        # Clean untracked files and directories310        subprocess.run(311            ["git", "clean", "-fdx"],312            cwd=str(repo_path),313            capture_output=True,314        )315 316        return True317 318    def execute_git_command(319        self, command: str, working_dir: str = ""320    ) -> tuple[int, str, str]:321        """322        Execute a git command in the workspace.323 324        Args:325            command (`str`):326                Git command to execute (without `git` prefix).327            working_dir (`str`, *optional*, defaults to `""`):328                Working directory relative to workspace.329 330        Returns:331            `tuple` of (exit_code, stdout, stderr).332        """333        work_path = (334            self.workspace_dir / working_dir if working_dir else self.workspace_dir335        )336 337        if not work_path.exists():338            return (1, "", f"Working directory does not exist: {work_path}")339 340        # Split command safely341        cmd_parts = ["git"] + command.split()342 343        result = subprocess.run(344            cmd_parts,345            cwd=str(work_path),346            capture_output=True,347            text=True,348        )349 350        return (result.returncode, result.stdout, result.stderr)351 352    def get_current_commit(self, repo_name: str) -> str:353        """354        Get current commit hash of a workspace repository.355 356        Args:357            repo_name (`str`):358                Name of repository in workspace.359 360        Returns:361            `str`: Commit hash.362        """363        repo_path = self.workspace_dir / repo_name364 365        if not repo_path.exists():366            raise RuntimeError(f"Repository not found: {repo_name}")367 368        result = subprocess.run(369            ["git", "rev-parse", "HEAD"],370            cwd=str(repo_path),371            capture_output=True,372            text=True,373        )374 375        if result.returncode != 0:376            raise RuntimeError(f"Failed to get commit: {result.stderr}")377 378        return result.stdout.strip()379 380    def workspace_exists(self, repo_name: str) -> bool:381        """Check if a repository exists in workspace."""382        return (self.workspace_dir / repo_name).exists()383