jira-ai-fixer/api/services/bitbucket.py

"""
Bitbucket Service - Client for Bitbucket Server API.
"""
from typing import Optional, Dict, Any, List
import httpx
import logging

logger = logging.getLogger(__name__)


class BitbucketClient:
    """Bitbucket Server REST API client."""
    
    def __init__(self, base_url: str, token: str):
        self.base_url = base_url.rstrip("/")
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json",
        }
    
    async def get_file_content(
        self,
        project: str,
        repo: str,
        file_path: str,
        ref: str = "main",
    ) -> str:
        """Get raw file content from a repository."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/rest/api/1.0/projects/{project}/repos/{repo}/raw/{file_path}",
                headers=self.headers,
                params={"at": ref},
            )
            response.raise_for_status()
            return response.text
    
    async def list_files(
        self,
        project: str,
        repo: str,
        path: str = "",
        ref: str = "main",
    ) -> List[Dict[str, Any]]:
        """List files in a directory."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/rest/api/1.0/projects/{project}/repos/{repo}/files/{path}",
                headers=self.headers,
                params={"at": ref},
            )
            response.raise_for_status()
            return response.json().get("values", [])
    
    async def create_branch(
        self,
        project: str,
        repo: str,
        branch_name: str,
        start_point: str = "main",
    ) -> Dict[str, Any]:
        """Create a new branch."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/rest/api/1.0/projects/{project}/repos/{repo}/branches",
                headers=self.headers,
                json={
                    "name": branch_name,
                    "startPoint": f"refs/heads/{start_point}",
                },
            )
            response.raise_for_status()
            return response.json()
    
    async def commit_file(
        self,
        project: str,
        repo: str,
        branch: str,
        file_path: str,
        content: str,
        message: str,
    ) -> Dict[str, Any]:
        """Commit a file change to a branch."""
        # Get current commit for the branch
        async with httpx.AsyncClient() as client:
            # First, get the latest commit on the branch
            branch_response = await client.get(
                f"{self.base_url}/rest/api/1.0/projects/{project}/repos/{repo}/branches",
                headers=self.headers,
                params={"filterText": branch},
            )
            branch_response.raise_for_status()
            branches = branch_response.json().get("values", [])
            
            if not branches:
                raise ValueError(f"Branch not found: {branch}")
            
            latest_commit = branches[0].get("latestCommit")
            
            # Use file edit API
            response = await client.put(
                f"{self.base_url}/rest/api/1.0/projects/{project}/repos/{repo}/browse/{file_path}",
                headers=self.headers,
                json={
                    "content": content,
                    "message": message,
                    "branch": branch,
                    "sourceCommitId": latest_commit,
                },
            )
            response.raise_for_status()
            return response.json()
    
    async def create_pull_request(
        self,
        project: str,
        repo: str,
        title: str,
        description: str,
        source_branch: str,
        target_branch: str = "main",
        target_project: Optional[str] = None,
        target_repo: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Create a pull request."""
        target_project = target_project or project
        target_repo = target_repo or repo
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/rest/api/1.0/projects/{project}/repos/{repo}/pull-requests",
                headers=self.headers,
                json={
                    "title": title,
                    "description": description,
                    "fromRef": {
                        "id": f"refs/heads/{source_branch}",
                        "repository": {
                            "slug": repo,
                            "project": {"key": project},
                        },
                    },
                    "toRef": {
                        "id": f"refs/heads/{target_branch}",
                        "repository": {
                            "slug": target_repo,
                            "project": {"key": target_project},
                        },
                    },
                },
            )
            response.raise_for_status()
            return response.json()
    
    async def get_repositories(self, project: str) -> List[Dict[str, Any]]:
        """List repositories in a project."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/rest/api/1.0/projects/{project}/repos",
                headers=self.headers,
            )
            response.raise_for_status()
            return response.json().get("values", [])
    
    async def search_code(
        self,
        project: str,
        repo: str,
        query: str,
        ref: str = "main",
    ) -> List[Dict[str, Any]]:
        """Search for code in a repository."""
        # Bitbucket Server code search API
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/rest/search/1.0/search",
                headers=self.headers,
                params={
                    "query": query,
                    "entities": "code",
                    "projectKey": project,
                    "repositorySlug": repo,
                },
            )
            if response.status_code == 200:
                return response.json().get("values", [])
            return []