openapi-first/openapi_first/client.py

from typing import Any, Callable, Dict, Optional
from urllib.parse import urljoin

import httpx

from .errors import OpenAPIFirstError


class OpenAPIClientError(OpenAPIFirstError):
    """Raised when an OpenAPI client operation fails."""


class OpenAPIClient:
    """
    OpenAPI-first HTTP client (httpx-based).

    - One callable per operationId
    - Explicit parameters (path, query, headers, body)
    - No implicit schema inference or mutation
    """

    def __init__(
        self,
        spec: dict[str, Any],
        base_url: Optional[str] = None,
        client: Optional[httpx.Client] = None,
    ) -> None:
        self.spec = spec
        self.base_url = base_url or self._resolve_base_url(spec)
        self.client = client or httpx.Client(base_url=self.base_url)

        self._operations: Dict[str, Callable[..., httpx.Response]] = {}
        self._build_operations()

    # ------------------------------------------------------------------ #
    # Public API
    # ------------------------------------------------------------------ #

    def __getattr__(self, name: str) -> Callable[..., httpx.Response]:
        try:
            return self._operations[name]
        except KeyError:
            raise AttributeError(f"No such operationId: {name}") from None

    def operations(self) -> Dict[str, Callable[..., httpx.Response]]:
        return dict(self._operations)

    # ------------------------------------------------------------------ #
    # Internal mechanics
    # ------------------------------------------------------------------ #

    def _resolve_base_url(self, spec: dict[str, Any]) -> str:
        servers = spec.get("servers")
        if not servers:
            raise OpenAPIClientError("No servers defined in OpenAPI spec")

        url = servers[0].get("url")
        if not url:
            raise OpenAPIClientError("Server entry missing 'url'")

        return url.rstrip("/") + "/"

    def _build_operations(self) -> None:
        paths = self.spec.get("paths", {})
        if not paths:
            raise OpenAPIClientError("OpenAPI spec contains no paths")

        for path, path_item in paths.items():
            for method, operation in path_item.items():
                if method.lower() not in {
                    "get", "post", "put", "patch", "delete", "head", "options"
                }:
                    continue

                operation_id = operation.get("operationId")
                if not operation_id:
                    raise OpenAPIClientError(
                        f"Missing operationId for {method.upper()} {path}"
                    )

                if operation_id in self._operations:
                    raise OpenAPIClientError(
                        f"Duplicate operationId detected: {operation_id}"
                    )

                self._operations[operation_id] = self._make_operation(
                    method=method.upper(),
                    path=path,
                    operation=operation,
                )

    def _make_operation(
        self,
        *,
        method: str,
        path: str,
        operation: dict[str, Any],
    ) -> Callable[..., httpx.Response]:
        request_body = operation.get("requestBody")

        def call(
            *,
            path_params: Optional[dict[str, Any]] = None,
            query: Optional[dict[str, Any]] = None,
            headers: Optional[dict[str, str]] = None,
            body: Optional[Any] = None,
            timeout: Optional[float] = None,
        ) -> httpx.Response:
            url = self._build_url(path, path_params or {})

            req_headers = headers.copy() if headers else {}

            json_data = None
            content = None

            if request_body is not None:
                if body is None:
                    raise OpenAPIClientError(
                        f"Request body required for operation {operation['operationId']}"
                    )

                media_types = request_body.get("content", {})
                if "application/json" in media_types:
                    json_data = body
                    req_headers.setdefault("Content-Type", "application/json")
                else:
                    content = body

            response = self.client.request(
                method=method,
                url=url,
                params=query,
                headers=req_headers,
                json=json_data,
                content=content,
                timeout=timeout,
            )

            return response

        call.__name__ = operation["operationId"]
        call.__doc__ = self._build_docstring(method, path, operation)

        return call

    def _build_url(self, path: str, path_params: dict[str, Any]) -> str:
        try:
            formatted_path = path.format(**path_params)
        except KeyError as exc:
            raise OpenAPIClientError(
                f"Missing path parameter: {exc.args[0]}"
            ) from exc

        return urljoin(self.base_url, formatted_path.lstrip("/"))

    def _build_docstring(
        self,
        method: str,
        path: str,
        operation: dict[str, Any],
    ) -> str:
        lines = [
            f"{method} {path}",
            "",
            operation.get("summary", ""),
            "",
            "Parameters:",
            "  path_params: dict | None",
            "  query: dict | None",
            "  headers: dict | None",
            "  body: Any | None",
            "",
            "Returns:",
            "  httpx.Response",
        ]
        return "\n".join(lines)