freepik/freepik_cli/api/client.py

"""Freepik HTTP client with authentication, error handling, and download support."""

from __future__ import annotations

from typing import Any, Optional

import httpx

from freepik_cli import __version__

BASE_URL = "https://api.freepik.com"
DEFAULT_TIMEOUT = 60.0


class FreepikAPIError(Exception):
    """Raised when the Freepik API returns an error response."""

    def __init__(self, message: str, status_code: Optional[int] = None, raw: Optional[dict] = None):
        super().__init__(message)
        self.status_code = status_code
        self.raw = raw or {}

    @classmethod
    def from_response(cls, response: httpx.Response) -> "FreepikAPIError":
        try:
            body = response.json()
        except Exception:
            body = {}

        message = (
            body.get("message")
            or body.get("error", {}).get("message")
            or body.get("errors", [{}])[0].get("message")
            or f"HTTP {response.status_code}"
        )

        # Append individual field validation errors when present
        invalid = body.get("invalid_params", [])
        if invalid:
            details = "\n".join(
                f"  • {p.get('field', '?')}: {p.get('reason', '')}" for p in invalid
            )
            message = f"{message}\n\n{details}"

        hints = {
            401: "Check your API key — set FREEPIK_API_KEY or use --api-key.",
            403: "Your plan may not support this feature. Check your Freepik subscription.",
            429: "Rate limit exceeded. Please wait before retrying.",
        }
        hint = hints.get(response.status_code)
        if hint:
            message = f"{message}\n\nHint: {hint}"

        return cls(message, status_code=response.status_code, raw=body)


class FreepikClient:
    """Thin synchronous HTTP wrapper around the Freepik API."""

    def __init__(
        self,
        api_key: str,
        base_url: str = BASE_URL,
        timeout: float = DEFAULT_TIMEOUT,
    ) -> None:
        self._client = httpx.Client(
            base_url=base_url,
            headers={
                "x-freepik-api-key": api_key,
                "Content-Type": "application/json",
                "Accept": "application/json",
                "User-Agent": f"freepik-cli/{__version__}",
            },
            timeout=httpx.Timeout(timeout),
        )

    def post(self, path: str, json: dict[str, Any]) -> dict[str, Any]:
        try:
            response = self._client.post(path, json=json)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as exc:
            raise FreepikAPIError.from_response(exc.response) from exc
        except httpx.RequestError as exc:
            raise FreepikAPIError(f"Network error: {exc}") from exc

    def post_multipart(self, path: str, data: dict[str, Any], files: dict[str, Any]) -> dict[str, Any]:
        """POST with multipart/form-data (for file uploads)."""
        headers = {k: v for k, v in self._client.headers.items() if k.lower() != "content-type"}
        try:
            response = self._client.post(path, data=data, files=files, headers=headers)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as exc:
            raise FreepikAPIError.from_response(exc.response) from exc
        except httpx.RequestError as exc:
            raise FreepikAPIError(f"Network error: {exc}") from exc

    def get(self, path: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
        try:
            response = self._client.get(path, params=params)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as exc:
            raise FreepikAPIError.from_response(exc.response) from exc
        except httpx.RequestError as exc:
            raise FreepikAPIError(f"Network error: {exc}") from exc

    def __enter__(self) -> "FreepikClient":
        return self

    def __exit__(self, *args: Any) -> None:
        self._client.close()

    def close(self) -> None:
        self._client.close()