refactor: extract git repository helpers to service

- Extract git operations helpers to services/git/operations.py - Slim git_repositories.py from 1,588 to 1,422 lines Quality gates: py_compile passes
2026-06-05 20:00:50 +00:00
parent de6a6a3b00
commit d8ab7734cb
2 changed files with 183 additions and 180 deletions
@@ -44,6 +44,13 @@ from src.utils.git_control import (
 )
 from src.utils.git_history import get_commit_detail, get_commit_history
 from src.utils.git_url_parser import parse_git_url
+from src.services.git.operations import (
+    build_provider_clone_url,
+    clone_working_repository,
+    get_repo_path,
+    init_working_repository,
+    preflight_remote_repository,
+)
 from src.services.shared.ssh_keys import _get_fernet

 router = APIRouter(prefix="/projects", tags=["git-repositories"])
@@ -51,179 +58,6 @@ router = APIRouter(prefix="/projects", tags=["git-repositories"])
 logger = logging.getLogger(__name__)


-def _get_repo_path(user_id: uuid.UUID, project_id: uuid.UUID, name: str) -> str:
-    """Generate the filesystem path for a repository.
-
-    Args:
-        user_id: UUID of the repository owner.
-        project_id: UUID of the project.
-        name: Repository name.
-
-    Returns:
-        Absolute path to the repository directory.
-    """
-    base = Settings().repo_base_path or "/data/repos"
-    return os.path.join(base, str(user_id), str(project_id), f"{name}.git")
-
-
-def _build_provider_clone_url(owner: str, repo: str) -> str:
-    """Build the SSH clone URL for the fixed git provider."""
-    return f"git@git.commumedia.org:{owner}/{repo}.git"
-
-
-def _prepare_ssh_env(ssh_key: SSHKey | None) -> dict | None:
-    """Prepare environment variables for git commands with SSH authentication.
-
-    Returns a dict of extra env vars, or None if no SSH key provided.
-    The caller is responsible for cleaning up the temporary key file.
-    """
-    if ssh_key is None:
-        return None
-
-    import tempfile
-
-    # Decrypt private key
-    fernet = _get_fernet()
-    private_key = fernet.decrypt(ssh_key.private_key_encrypted.encode()).decode()
-
-    # Write to temp file with restricted permissions
-    fd, key_path = tempfile.mkstemp(prefix="ssh_key_")
-    try:
-        os.write(fd, private_key.encode())
-    finally:
-        os.close(fd)
-    os.chmod(key_path, 0o600)
-
-    # Return env vars and the key path for cleanup
-    env = {
-        "GIT_SSH_COMMAND": f"ssh -i {key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null"
-    }
-    return env, key_path
-
-
-def _preflight_remote_repository(
-    remote_url: str, ssh_key: SSHKey | None = None
-) -> None:
-    """Verify a remote repository is reachable before cloning."""
-    env = None
-    key_path = None
-
-    if ssh_key is not None:
-        ssh_result = _prepare_ssh_env(ssh_key)
-        if ssh_result:
-            env, key_path = ssh_result
-
-    try:
-        result = subprocess.run(
-            ["git", "ls-remote", remote_url],
-            capture_output=True,
-            text=True,
-            timeout=60,
-            env={**os.environ, **env} if env else None,
-        )
-    except subprocess.TimeoutExpired:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail="remote repository check timed out",
-        )
-    except FileNotFoundError:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="git command not found",
-        )
-    finally:
-        if key_path and os.path.exists(key_path):
-            os.unlink(key_path)
-
-    if result.returncode != 0:
-        logger.error(
-            "Preflight check failed for %s: stderr=%s", remote_url, result.stderr
-        )
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"repository not found or inaccessible: {result.stderr}",
-        )
-
-
-def _clone_working_repository(
-    remote_url: str, repo_path: str, ssh_key: SSHKey | None = None
-) -> None:
-    env = None
-    key_path = None
-
-    if ssh_key is not None:
-        ssh_result = _prepare_ssh_env(ssh_key)
-        if ssh_result:
-            env, key_path = ssh_result
-
-    try:
-        result = subprocess.run(
-            ["git", "clone", remote_url, repo_path],
-            capture_output=True,
-            text=True,
-            timeout=300,
-            env={**os.environ, **env} if env else None,
-        )
-    except subprocess.TimeoutExpired:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST, detail="clone operation timed out"
-        )
-    except FileNotFoundError:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="git command not found",
-        )
-    finally:
-        if key_path and os.path.exists(key_path):
-            os.unlink(key_path)
-
-    if result.returncode != 0:
-        logger.error("Clone failed for %s: stderr=%s", remote_url, result.stderr)
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"failed to clone repository: {result.stderr}",
-        )
-
-
-def _init_working_repository(repo_path: str) -> None:
-    try:
-        result = subprocess.run(
-            ["git", "init", "-b", "main", repo_path],
-            capture_output=True,
-            text=True,
-        )
-    except FileNotFoundError:
-        raise HTTPException(
-            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-            detail="git command not found",
-        )
-
-    if result.returncode == 0:
-        return
-
-    fallback = subprocess.run(
-        ["git", "init", repo_path],
-        capture_output=True,
-        text=True,
-    )
-    if fallback.returncode != 0:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"failed to initialize repository: {fallback.stderr}",
-        )
-
-    ref_result = subprocess.run(
-        ["git", "-C", repo_path, "symbolic-ref", "HEAD", "refs/heads/main"],
-        capture_output=True,
-        text=True,
-    )
-    if ref_result.returncode != 0:
-        raise HTTPException(
-            status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"failed to set initial branch: {ref_result.stderr}",
-        )
-
-
@router.get(
    "/repositories",
    response_model=list[GitRepositoryResponse],
@@ -350,7 +184,7 @@ async def create_external_repository(
            )

    if remote_url:
-        _preflight_remote_repository(remote_url, ssh_key)
+        preflight_remote_repository(remote_url, ssh_key)

    # Create external repo with no project
    repo = GitRepository(
@@ -370,7 +204,7 @@ async def create_external_repository(

    if remote_url:
        try:
-            _clone_working_repository(remote_url, repo_path, ssh_key)
+            clone_working_repository(remote_url, repo_path, ssh_key)
            repo.is_mirror = False
        except Exception as exc:
            await session.rollback()
@@ -541,17 +375,17 @@ async def create_repository(
            )

    if remote_url:
-        _preflight_remote_repository(remote_url, ssh_key)
+        preflight_remote_repository(remote_url, ssh_key)

-    repo_path = _get_repo_path(user_id, project_id, data.name)
+    repo_path = get_repo_path(user_id, project_id, data.name)

    # Ensure parent directory exists
    os.makedirs(os.path.dirname(repo_path), exist_ok=True)

    if remote_url:
-        _clone_working_repository(remote_url, repo_path, ssh_key)
+        clone_working_repository(remote_url, repo_path, ssh_key)
    else:
-        _init_working_repository(repo_path)
+        init_working_repository(repo_path)

    repo = GitRepository(
        name=data.name,
@@ -969,7 +803,7 @@ async def get_repository_branches(
        if repo.ssh_key_id:
            ssh_key = await session.get(SSHKey, repo.ssh_key_id)

-        ssh_result = _prepare_ssh_env(ssh_key)
+        ssh_result = prepare_ssh_env(ssh_key)
        env = None
        key_path = None
        if ssh_result:
@@ -0,0 +1,169 @@
+"""Git repository operations service."""
+
+import logging
+import os
+import subprocess
+import uuid
+
+from fastapi import HTTPException, status
+
+from src.config import Settings
+from src.models import SSHKey
+from src.services.shared.ssh_keys import _get_fernet
+
+logger = logging.getLogger(__name__)
+
+
+def get_repo_path(user_id: uuid.UUID, project_id: uuid.UUID, name: str) -> str:
+    """Generate the filesystem path for a repository."""
+    base = Settings().repo_base_path or "/data/repos"
+    return os.path.join(base, str(user_id), str(project_id), f"{name}.git")
+
+
+def build_provider_clone_url(owner: str, repo: str) -> str:
+    """Build the SSH clone URL for the fixed git provider."""
+    return f"git@git.commumedia.org:{owner}/{repo}.git"
+
+
+def prepare_ssh_env(ssh_key: SSHKey | None) -> tuple[dict, str] | None:
+    """Prepare environment variables for git commands with SSH authentication."""
+    if ssh_key is None:
+        return None
+
+    import tempfile
+
+    fernet = _get_fernet()
+    private_key = fernet.decrypt(ssh_key.private_key_encrypted.encode()).decode()
+
+    fd, key_path = tempfile.mkstemp(prefix="ssh_key_")
+    try:
+        os.write(fd, private_key.encode())
+    finally:
+        os.close(fd)
+    os.chmod(key_path, 0o600)
+
+    env = {
+        "GIT_SSH_COMMAND": f"ssh -i {key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null"
+    }
+    return env, key_path
+
+
+def preflight_remote_repository(remote_url: str, ssh_key: SSHKey | None = None) -> None:
+    """Verify a remote repository is reachable before cloning."""
+    env = None
+    key_path = None
+
+    if ssh_key is not None:
+        ssh_result = prepare_ssh_env(ssh_key)
+        if ssh_result:
+            env, key_path = ssh_result
+
+    try:
+        result = subprocess.run(
+            ["git", "ls-remote", remote_url],
+            capture_output=True,
+            text=True,
+            timeout=60,
+            env={**os.environ, **env} if env else None,
+        )
+    except subprocess.TimeoutExpired:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="remote repository check timed out",
+        )
+    except FileNotFoundError:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="git command not found",
+        )
+    finally:
+        if key_path and os.path.exists(key_path):
+            os.unlink(key_path)
+
+    if result.returncode != 0:
+        logger.error(
+            "Preflight check failed for %s: stderr=%s", remote_url, result.stderr
+        )
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"repository not found or inaccessible: {result.stderr}",
+        )
+
+
+def clone_working_repository(remote_url: str, repo_path: str, ssh_key: SSHKey | None = None) -> None:
+    """Clone a remote repository to a local path."""
+    env = None
+    key_path = None
+
+    if ssh_key is not None:
+        ssh_result = prepare_ssh_env(ssh_key)
+        if ssh_result:
+            env, key_path = ssh_result
+
+    try:
+        result = subprocess.run(
+            ["git", "clone", remote_url, repo_path],
+            capture_output=True,
+            text=True,
+            timeout=300,
+            env={**os.environ, **env} if env else None,
+        )
+    except subprocess.TimeoutExpired:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST, detail="clone operation timed out"
+        )
+    except FileNotFoundError:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="git command not found",
+        )
+    finally:
+        if key_path and os.path.exists(key_path):
+            os.unlink(key_path)
+
+    if result.returncode != 0:
+        logger.error("Clone failed for %s: stderr=%s", remote_url, result.stderr)
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"failed to clone repository: {result.stderr}",
+        )
+
+
+def init_working_repository(repo_path: str) -> None:
+    """Initialize a new git repository at the given path."""
+    try:
+        result = subprocess.run(
+            ["git", "init", "-b", "main", repo_path],
+            capture_output=True,
+            text=True,
+        )
+    except FileNotFoundError:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail="git command not found",
+        )
+
+    if result.returncode == 0:
+        return
+
+    fallback = subprocess.run(
+        ["git", "init", repo_path],
+        capture_output=True,
+        text=True,
+    )
+    if fallback.returncode != 0:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"failed to initialize repository: {fallback.stderr}",
+        )
+
+    ref_result = subprocess.run(
+        ["git", "-C", repo_path, "symbolic-ref", "HEAD", "refs/heads/main"],
+        capture_output=True,
+        text=True,
+    )
+    if ref_result.returncode != 0:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"failed to set initial branch: {ref_result.stderr}",
+        )