""" OpenAPI-first HTTP client for contract-driven services. --- ## Summary 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`. Notes: **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 **Responsibilities:** - Parses an OpenAPI 3.x specification - Dynamically creates one callable per operationId - Enforces presence of servers, paths, and operationId - Formats path parameters safely - Handles JSON request bodies explicitly - Returns raw `httpx.Response` objects **Constraints:** - This module intentionally does NOT: Generate client code, validate request/response schemas, deserialize responses, retry requests, implement authentication helpers, or assume non-2xx responses are failures. """ 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). Notes: **Responsibilities:** - This client derives all callable methods directly from an OpenAPI 3.x specification. Each operationId becomes a method on the client instance. **Guarantees:** - 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 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()) ``` """ def __init__( self, spec: dict[str, Any], base_url: Optional[str] = None, client: Optional[httpx.Client] = None, ) -> None: """ Initialize the OpenAPI client. Args: spec (dict[str, Any]): Parsed OpenAPI 3.x specification. base_url (str, optional): Base URL of the target service. If omitted, the first entry in the OpenAPI `servers` list is used. client (httpx.Client, optional): Optional preconfigured httpx client instance. Raises: OpenAPIClientError: If no servers are defined, spec has no paths, operationIds are missing/duplicate, or required parameters are missing. """ 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)