gitea-mcp-remote/src/gitea_mcp/client.py

"""HTTP client for Gitea API."""

import httpx
from typing import Any, Optional
from .auth import AuthConfig


class GiteaClientError(Exception):
    """Base exception for Gitea client errors."""

    pass


class GiteaAuthError(GiteaClientError):
    """Authentication error."""

    pass


class GiteaNotFoundError(GiteaClientError):
    """Resource not found error."""

    pass


class GiteaServerError(GiteaClientError):
    """Server error."""

    pass


class GiteaClient:
    """Async HTTP client for Gitea API."""

    def __init__(self, config: AuthConfig, timeout: float = 30.0):
        """Initialize Gitea API client.

        Args:
            config: Authentication configuration.
            timeout: Request timeout in seconds (default: 30.0).
        """
        self.config = config
        self.timeout = timeout
        self._client: Optional[httpx.AsyncClient] = None

    async def __aenter__(self):
        """Async context manager entry."""
        self._client = httpx.AsyncClient(
            base_url=self.config.api_url,
            headers=self.config.get_auth_headers(),
            timeout=self.timeout,
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """Async context manager exit."""
        if self._client:
            await self._client.aclose()

    def _handle_error(self, response: httpx.Response) -> None:
        """Handle HTTP error responses.

        Args:
            response: HTTP response object.

        Raises:
            GiteaAuthError: For 401/403 errors.
            GiteaNotFoundError: For 404 errors.
            GiteaServerError: For 500+ errors.
            GiteaClientError: For other errors.
        """
        status = response.status_code

        if status == 401:
            raise GiteaAuthError(
                "Authentication failed. Please check your GITEA_API_TOKEN."
            )
        elif status == 403:
            raise GiteaAuthError(
                "Access forbidden. Your API token may not have required permissions."
            )
        elif status == 404:
            raise GiteaNotFoundError(
                f"Resource not found: {response.request.url}"
            )
        elif status >= 500:
            raise GiteaServerError(
                f"Gitea server error (HTTP {status}): {response.text}"
            )
        else:
            raise GiteaClientError(
                f"API request failed (HTTP {status}): {response.text}"
            )

    async def get(self, path: str, **kwargs) -> dict[str, Any]:
        """Make GET request to Gitea API.

        Args:
            path: API endpoint path (e.g., "/api/v1/repos/owner/repo").
            **kwargs: Additional arguments for httpx request.

        Returns:
            dict: JSON response data.

        Raises:
            GiteaClientError: If request fails.
        """
        if not self._client:
            raise GiteaClientError(
                "Client not initialized. Use 'async with' context manager."
            )

        try:
            response = await self._client.get(path, **kwargs)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError:
            self._handle_error(response)
        except httpx.RequestError as e:
            raise GiteaClientError(
                f"Request failed: {e}"
            ) from e

    async def post(self, path: str, json: Optional[dict[str, Any]] = None, **kwargs) -> dict[str, Any]:
        """Make POST request to Gitea API.

        Args:
            path: API endpoint path.
            json: JSON data to send in request body.
            **kwargs: Additional arguments for httpx request.

        Returns:
            dict: JSON response data.

        Raises:
            GiteaClientError: If request fails.
        """
        if not self._client:
            raise GiteaClientError(
                "Client not initialized. Use 'async with' context manager."
            )

        try:
            response = await self._client.post(path, json=json, **kwargs)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError:
            self._handle_error(response)
        except httpx.RequestError as e:
            raise GiteaClientError(
                f"Request failed: {e}"
            ) from e

    async def patch(self, path: str, json: Optional[dict[str, Any]] = None, **kwargs) -> dict[str, Any]:
        """Make PATCH request to Gitea API.

        Args:
            path: API endpoint path.
            json: JSON data to send in request body.
            **kwargs: Additional arguments for httpx request.

        Returns:
            dict: JSON response data.

        Raises:
            GiteaClientError: If request fails.
        """
        if not self._client:
            raise GiteaClientError(
                "Client not initialized. Use 'async with' context manager."
            )

        try:
            response = await self._client.patch(path, json=json, **kwargs)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError:
            self._handle_error(response)
        except httpx.RequestError as e:
            raise GiteaClientError(
                f"Request failed: {e}"
            ) from e

    async def delete(self, path: str, **kwargs) -> Optional[dict[str, Any]]:
        """Make DELETE request to Gitea API.

        Args:
            path: API endpoint path.
            **kwargs: Additional arguments for httpx request.

        Returns:
            dict or None: JSON response data if available, None for 204 responses.

        Raises:
            GiteaClientError: If request fails.
        """
        if not self._client:
            raise GiteaClientError(
                "Client not initialized. Use 'async with' context manager."
            )

        try:
            response = await self._client.delete(path, **kwargs)
            response.raise_for_status()

            # DELETE requests may return 204 No Content
            if response.status_code == 204:
                return None

            return response.json()
        except httpx.HTTPStatusError:
            self._handle_error(response)
        except httpx.RequestError as e:
            raise GiteaClientError(
                f"Request failed: {e}"
            ) from e