aetos/openapi-first
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Introduce an OpenAPI-first HTTP client driven by the same specification

Browse Source 
used for server-side route binding, and refactor documentation and
templates to treat the client as a first-class contract consumer.

Key changes:
- Add OpenAPI-first client module based on httpx
- Document client usage alongside server-side binder usage
- Update mkdocs navigation to include client documentation
- Refactor CRUD and model app templates to call APIs via operationId
  instead of hardcoded paths
- Align package documentation and public API surface with client support
- Clarify server/client dependency split (fastapi vs httpx)

This establishes strict symmetry between OpenAPI-driven server binding
and OpenAPI-driven client invocation, reinforcing OpenAPI as the single
source of truth on both sides of the contract.
This commit is contained in:
Vishesh 'ironeagle' Bangotra
2026-01-11 18:41:27 +05:30
parent 31bf1b1b6b
commit a74e3d0d01
7 changed files with 295 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
docs/openapi_first/client.md Normal file
View File

@@ -0,0 +1,3 @@
# Client
::: openapi_first.client

1
mkdocs.yml
View File

@@ -42,6 +42,7 @@ nav:
- OpenAPI-First App: openapi_first/app.md
- Route Binder: openapi_first/binder.md
- Spec Loaders: openapi_first/loader.md
- Client: openapi_first/client.md
- CLI:
- Home: openapi_first/cli.md

67
openapi_first/__init__.py
View File

@@ -17,11 +17,12 @@ convenience facades.
Architecture Overview
----------------------------------------------------------------------
The library is structured around three core responsibilities:
The library is structured around four core responsibilities:
- loader: load and validate OpenAPI 3.x specifications (JSON/YAML)
- binder: bind OpenAPI operations to FastAPI routes via operationId
- app: OpenAPI-first FastAPI application bootstrap
- client: OpenAPI-first HTTP client driven by the same specification
- errors: explicit error hierarchy for contract violations
The package root acts as a **namespace**, not a facade. Consumers are
@@ -40,7 +41,8 @@ Or with Poetry:
poetry add openapi-first
Runtime dependencies are intentionally minimal:
- fastapi
- fastapi (server-side)
- httpx (client-side)
- openapi-spec-validator
- pyyaml (optional, for YAML specs)
@@ -48,7 +50,7 @@ The ASGI server (e.g., uvicorn) is an application-level dependency and is
not bundled with this library.
----------------------------------------------------------------------
Basic Usage
Server-Side Usage (OpenAPI → FastAPI)
----------------------------------------------------------------------
Minimal OpenAPI-first FastAPI application:
@@ -57,7 +59,7 @@ Minimal OpenAPI-first FastAPI application:
import my_service.routes as routes
api = app.OpenAPIFirstApp(
openapi_path="openapi.json",
openapi_path="openapi.yaml",
routes_module=routes,
title="My Service",
version="1.0.0",
@@ -81,6 +83,56 @@ OpenAPI snippet:
"200":
description: OK
The binder guarantees:
- Every OpenAPI operationId has exactly one handler
- No undocumented routes exist
- All mismatches fail at application startup
----------------------------------------------------------------------
Client-Side Usage (OpenAPI → HTTP Client)
----------------------------------------------------------------------
The same OpenAPI specification can be used to construct a strict,
operationId-driven HTTP client.
Client construction:
from openapi_first.loader import load_openapi
from openapi_first.client import OpenAPIClient
spec = load_openapi("openapi.yaml")
client = OpenAPIClient(spec)
Calling operations (operationId is the API):
response = client.get_health()
assert response.status_code == 200
assert response.json() == {"status": "ok"}
Path parameters must match the OpenAPI specification exactly:
response = client.get_item(
path_params={"item_id": 1}
)
Request bodies are passed explicitly:
response = client.create_item(
body={"name": "Orange", "price": 0.8}
)
Client guarantees:
- One callable per OpenAPI operationId
- No hardcoded URLs or HTTP methods in user code
- Path and body parameters must match the spec exactly
- Invalid or incomplete OpenAPI specs fail at client construction time
- No schema inference or mutation is performed
The client is transport-level only and returns `httpx.Response`
objects directly. Response interpretation and validation are left to
the consumer or higher-level layers.
----------------------------------------------------------------------
Extensibility Model
----------------------------------------------------------------------
@@ -107,6 +159,7 @@ The supported public API consists of the following top-level modules:
- openapi_first.app
- openapi_first.binder
- openapi_first.loader
- openapi_first.client
- openapi_first.errors
Classes and functions should be imported explicitly from these modules.
@@ -118,8 +171,8 @@ Design Guarantees
- OpenAPI is the single source of truth
- No undocumented routes can exist
- No OpenAPI operation can exist without a handler
- All contract violations fail at application startup
- No OpenAPI operation can exist without a handler or client callable
- All contract violations fail at application startup or client creation
- No hidden FastAPI magic or implicit behavior
- Deterministic, testable application assembly
- CI-friendly failure modes
@@ -131,11 +184,13 @@ enforcement over convenience shortcuts.
from . import app
from . import binder
from . import loader
from . import client
from . import errors
__all__ = [
"app",
"binder",
"loader",
"client",
"errors",
]

176
openapi_first/client.py Normal file
View File

@@ -0,0 +1,176 @@
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)

2
openapi_first/loader.py
View File

@@ -25,8 +25,6 @@ This module intentionally does NOT:
- Perform request/response validation at runtime
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any

35
openapi_first/templates/crud_app/test_crud_app.py
View File

@@ -13,14 +13,22 @@ The tests assume:
from fastapi.testclient import TestClient
from main import app
from openapi_first.loader import load_openapi
from openapi_first.client import OpenAPIClient
client = TestClient(app)
spec = load_openapi("openapi.yaml")
client = OpenAPIClient(
spec=spec,
base_url="http://testserver",
client=client,
)
def test_list_items_initial():
"""Initial items should be present."""
response = client.get("/items")
response = client.list_items()
assert response.status_code == 200
data = response.json()
@@ -34,7 +42,9 @@ def test_list_items_initial():
def test_get_item():
"""Existing item should be retrievable by ID."""
response = client.get("/items/1")
response = client.get_item(
path_params={"item_id": 1}
)
assert response.status_code == 200
item = response.json()
@@ -50,7 +60,9 @@ def test_create_item():
"price": 0.8,
}
response = client.post("/items", json=payload)
response = client.create_item(
body=payload
)
assert response.status_code == 201
item = response.json()
@@ -59,7 +71,7 @@ def test_create_item():
assert item["price"] == payload["price"]
# Verify it appears in list
list_response = client.get("/items")
list_response = client.list_items()
ids = {i["id"] for i in list_response.json()}
assert item["id"] in ids
@@ -71,7 +83,10 @@ def test_update_item():
"price": 0.6,
}
response = client.put("/items/1", json=payload)
response = client.update_item(
path_params={"item_id": 1},
body=payload,
)
assert response.status_code == 200
item = response.json()
@@ -80,7 +95,9 @@ def test_update_item():
assert item["price"] == payload["price"]
# Verify persisted update
get_response = client.get("/items/1")
get_response = client.get_item(
path_params={"item_id": 1}
)
updated = get_response.json()
assert updated["name"] == payload["name"]
assert updated["price"] == payload["price"]
@@ -88,10 +105,12 @@ def test_update_item():
def test_delete_item():
"""Deleting an item should remove it from the store."""
response = client.delete("/items/2")
response = client.delete_item(
path_params={"item_id": 2}
)
assert response.status_code == 204
# Verify deletion
list_response = client.get("/items")
list_response = client.list_items()
ids = {item["id"] for item in list_response.json()}
assert 2 not in ids

35
openapi_first/templates/model_app/test_model_app.py
View File

@@ -14,14 +14,22 @@ The tests assume:
from fastapi.testclient import TestClient
from main import app
from openapi_first.loader import load_openapi
from openapi_first.client import OpenAPIClient
client = TestClient(app)
spec = load_openapi("openapi.yaml")
client = OpenAPIClient(
spec=spec,
base_url="http://testserver",
client=client,
)
def test_list_items_initial():
"""Initial items should be present."""
response = client.get("/items")
response = client.list_items()
assert response.status_code == 200
data = response.json()
@@ -35,7 +43,9 @@ def test_list_items_initial():
def test_get_item():
"""Existing item should be retrievable by ID."""
response = client.get("/items/1")
response = client.get_item(
path_params={"item_id": 1}
)
assert response.status_code == 200
item = response.json()
@@ -51,7 +61,9 @@ def test_create_item():
"price": 0.8,
}
response = client.post("/items", json=payload)
response = client.create_item(
body=payload
)
assert response.status_code == 201
item = response.json()
@@ -60,7 +72,7 @@ def test_create_item():
assert item["price"] == payload["price"]
# Verify it appears in list
list_response = client.get("/items")
list_response = client.list_items()
ids = {i["id"] for i in list_response.json()}
assert item["id"] in ids
@@ -72,7 +84,10 @@ def test_update_item():
"price": 0.6,
}
response = client.put("/items/1", json=payload)
response = client.update_item(
path_params={"item_id": 1},
body=payload,
)
assert response.status_code == 200
item = response.json()
@@ -81,7 +96,9 @@ def test_update_item():
assert item["price"] == payload["price"]
# Verify persisted update
get_response = client.get("/items/1")
get_response = client.get_item(
path_params={"item_id": 1}
)
updated = get_response.json()
assert updated["name"] == payload["name"]
assert updated["price"] == payload["price"]
@@ -89,10 +106,12 @@ def test_update_item():
def test_delete_item():
"""Deleting an item should remove it from the store."""
response = client.delete("/items/2")
response = client.delete_item(
path_params={"item_id": 2}
)
assert response.status_code == 204
# Verify deletion
list_response = client.get("/items")
list_response = client.list_items()
ids = {item["id"] for item in list_response.json()}
assert 2 not in ids