""" openapi_first.client ==================== OpenAPI-first HTTP client for contract-driven services. This module provides `OpenAPIClient`, a thin, strict HTTP client that derives all callable operations directly from an OpenAPI 3.x specification. It is the client counterpart to `OpenAPIFirstApp`. Core principles --------------- - The OpenAPI specification is the single source of truth. - Each operationId becomes a callable Python method. - No implicit schema mutation or inference. - No code generation step. - Minimal abstraction over httpx. What this module does --------------------- - Parses an OpenAPI 3.x specification. - Dynamically creates one callable per operationId. - Enforces presence of: - servers - paths - operationId - Formats path parameters safely. - Handles JSON request bodies explicitly. - Returns raw `httpx.Response` objects. What this module does NOT do ---------------------------- - It does not generate client code. - It does not validate request/response schemas. - It does not deserialize responses. - It does not retry requests. - It does not implement authentication helpers. - It does not assume non-2xx responses are failures. Intended usage -------------- This client is designed for: - Service-to-service communication - Integration testing - Contract-driven internal SDK usage - Systems that want OpenAPI-first symmetry with `OpenAPIFirstApp` Design constraints ------------------ - Only the first server in the OpenAPI `servers` list is used if `base_url` is not explicitly provided. - Only explicitly declared request bodies are allowed. - `application/json` is handled natively; other media types are sent as raw content. - All responses are returned as-is. """ 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). This client derives all callable methods directly from an OpenAPI 3.x specification. Each operationId becomes a method on the client instance. Design principles ----------------- - One callable per operationId - Explicit parameters (path, query, headers, body) - No implicit schema inference or mutation - Returns raw httpx.Response objects - No response validation or deserialization Parameters ---------- spec : dict Parsed OpenAPI 3.x specification. base_url : str | None Base URL of the target service. If omitted, the first entry in the OpenAPI `servers` list is used. client : httpx.Client | None Optional preconfigured httpx client instance. Raises ------ OpenAPIClientError If: - No servers are defined and base_url is not provided - OpenAPI spec has no paths - An operation is missing operationId - Duplicate operationIds are detected - Required path parameters are missing - Required request body is missing Example ------- ```python from openapi_first import loader, client spec = loader.load_openapi("openapi.yaml") api = client.OpenAPIClient( spec=spec, base_url="http://localhost:8000", ) # Call operationId: getUser response = api.getUser( path_params={"user_id": 123} ) print(response.status_code) print(response.json()) # Call operationId: createUser response = api.createUser( body={"name": "Bob"} ) print(response.status_code) ``` """ 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)