258 lines
8.0 KiB
Python
258 lines
8.0 KiB
Python
"""
|
|
# Summary
|
|
|
|
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`.
|
|
|
|
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)
|