docfast/sdk/python/docfast/client.py

"""DocFast API clients (sync and async)."""

from __future__ import annotations

from typing import Any, Dict, List, Optional
from urllib.parse import quote

import httpx


class DocFastError(Exception):
    """Error returned by the DocFast API."""

    def __init__(self, message: str, status: int, code: Optional[str] = None):
        super().__init__(message)
        self.status = status
        self.code = code


_KEY_MAP = {"print_background": "printBackground"}


def _build_body(key: str, value: str, options: Optional[Dict[str, Any]]) -> Dict[str, Any]:
    body: Dict[str, Any] = {key: value}
    if options:
        body["options"] = {_KEY_MAP.get(k, k): v for k, v in options.items()}
    return body


def _handle_error(response: httpx.Response) -> None:
    if response.is_success:
        return
    message = f"HTTP {response.status_code}"
    code = None
    try:
        data = response.json()
        if "error" in data:
            message = data["error"]
        code = data.get("code")
    except Exception:
        pass
    raise DocFastError(message, response.status_code, code)


class DocFast:
    """Synchronous DocFast client."""

    def __init__(self, api_key: str, *, base_url: Optional[str] = None):
        if not api_key:
            raise ValueError("API key is required")
        self._base_url = (base_url or "https://docfast.dev").rstrip("/")
        self._client = httpx.Client(
            base_url=self._base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=120.0,
        )

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

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

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

    def _convert(self, path: str, body: Dict[str, Any]) -> bytes:
        r = self._client.post(path, json=body)
        _handle_error(r)
        return r.content

    def html(self, html: str, **options: Any) -> bytes:
        """Convert HTML to PDF."""
        return self._convert("/v1/convert/html", _build_body("html", html, options or None))

    def markdown(self, markdown: str, **options: Any) -> bytes:
        """Convert Markdown to PDF."""
        return self._convert("/v1/convert/markdown", _build_body("markdown", markdown, options or None))

    def url(self, url: str, **options: Any) -> bytes:
        """Convert a URL to PDF."""
        return self._convert("/v1/convert/url", _build_body("url", url, options or None))

    def templates(self) -> List[Dict[str, Any]]:
        """List available templates."""
        r = self._client.get("/v1/templates")
        _handle_error(r)
        return r.json()

    def render_template(self, template_id: str, data: Dict[str, Any], **options: Any) -> bytes:
        """Render a template to PDF."""
        body: Dict[str, Any] = {"data": data}
        if options:
            body["options"] = options
        return self._convert(f"/v1/templates/{quote(template_id, safe='')}/render", body)


class AsyncDocFast:
    """Asynchronous DocFast client."""

    def __init__(self, api_key: str, *, base_url: Optional[str] = None):
        if not api_key:
            raise ValueError("API key is required")
        self._base_url = (base_url or "https://docfast.dev").rstrip("/")
        self._client = httpx.AsyncClient(
            base_url=self._base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=120.0,
        )

    async def __aenter__(self) -> "AsyncDocFast":
        return self

    async def __aexit__(self, *args: Any) -> None:
        await self.close()

    async def close(self) -> None:
        await self._client.aclose()

    async def _convert(self, path: str, body: Dict[str, Any]) -> bytes:
        r = await self._client.post(path, json=body)
        _handle_error(r)
        return r.content

    async def html(self, html: str, **options: Any) -> bytes:
        """Convert HTML to PDF."""
        return await self._convert("/v1/convert/html", _build_body("html", html, options or None))

    async def markdown(self, markdown: str, **options: Any) -> bytes:
        """Convert Markdown to PDF."""
        return await self._convert("/v1/convert/markdown", _build_body("markdown", markdown, options or None))

    async def url(self, url: str, **options: Any) -> bytes:
        """Convert a URL to PDF."""
        return await self._convert("/v1/convert/url", _build_body("url", url, options or None))

    async def templates(self) -> List[Dict[str, Any]]:
        """List available templates."""
        r = await self._client.get("/v1/templates")
        _handle_error(r)
        return r.json()

    async def render_template(self, template_id: str, data: Dict[str, Any], **options: Any) -> bytes:
        """Render a template to PDF."""
        body: Dict[str, Any] = {"data": data}
        if options:
            body["options"] = options
        return await self._convert(f"/v1/templates/{quote(template_id, safe='')}/render", body)