feat: OpenAPI assembly from endpoint records
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,56 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from auto_reverse.models import EndpointRecord
|
||||
|
||||
_PARAM = re.compile(r"\{([^}]+)\}")
|
||||
|
||||
|
||||
def _path_params(template: str) -> list[dict[str, Any]]:
|
||||
return [
|
||||
{"name": name, "in": "path", "required": True, "schema": {"type": "string"}}
|
||||
for name in _PARAM.findall(template)
|
||||
]
|
||||
|
||||
|
||||
def _operation(rec: EndpointRecord) -> dict[str, Any]:
|
||||
op: dict[str, Any] = {}
|
||||
if rec.summary:
|
||||
op["summary"] = rec.summary
|
||||
if rec.description:
|
||||
op["description"] = rec.description
|
||||
if rec.tag:
|
||||
op["tags"] = [rec.tag]
|
||||
params = _path_params(rec.signature.path_template)
|
||||
params += [
|
||||
{"name": q, "in": "query", "required": False, "schema": {"type": "string"}}
|
||||
for q in sorted(rec.query_params)
|
||||
]
|
||||
if params:
|
||||
op["parameters"] = params
|
||||
if rec.request_schema is not None:
|
||||
op["requestBody"] = {
|
||||
"content": {"application/json": {"schema": rec.request_schema}}
|
||||
}
|
||||
status = rec.signature.status_class.replace("x", "X")
|
||||
response: dict[str, Any] = {"description": rec.summary or "Response"}
|
||||
if rec.response_schema is not None:
|
||||
response["content"] = {"application/json": {"schema": rec.response_schema}}
|
||||
op["responses"] = {status[0] + "XX": response}
|
||||
return op
|
||||
|
||||
|
||||
def build_openapi(records: list[EndpointRecord], title: str) -> dict[str, Any]:
|
||||
paths: dict[str, dict[str, Any]] = {}
|
||||
for rec in records:
|
||||
template = rec.signature.path_template
|
||||
method = rec.signature.method.lower()
|
||||
paths.setdefault(template, {})[method] = _operation(rec)
|
||||
return {
|
||||
"openapi": "3.1.0",
|
||||
"info": {"title": title, "version": "0.0.0"},
|
||||
"paths": paths,
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
from auto_reverse.doc.openapi import build_openapi
|
||||
from auto_reverse.models import EndpointRecord, Signature
|
||||
|
||||
|
||||
def _record(method: str, template: str, **kw) -> EndpointRecord:
|
||||
rec = EndpointRecord(signature=Signature(method, "ex.com", template, "2xx"))
|
||||
for k, v in kw.items():
|
||||
setattr(rec, k, v)
|
||||
return rec
|
||||
|
||||
|
||||
def test_builds_paths_and_methods():
|
||||
records = [
|
||||
_record("GET", "/api/users", summary="List users",
|
||||
response_schema={"type": "array"}),
|
||||
_record("POST", "/api/users", summary="Create user",
|
||||
request_schema={"type": "object"}),
|
||||
]
|
||||
spec = build_openapi(records, title="ex.com API")
|
||||
assert spec["openapi"].startswith("3.")
|
||||
assert spec["info"]["title"] == "ex.com API"
|
||||
assert set(spec["paths"]["/api/users"]) == {"get", "post"}
|
||||
assert spec["paths"]["/api/users"]["get"]["summary"] == "List users"
|
||||
|
||||
|
||||
def test_path_param_declared_for_template():
|
||||
rec = _record("GET", "/api/users/{id}", summary="Get user")
|
||||
spec = build_openapi([rec], title="x")
|
||||
params = spec["paths"]["/api/users/{id}"]["get"]["parameters"]
|
||||
assert any(p["name"] == "id" and p["in"] == "path" for p in params)
|
||||
|
||||
|
||||
def test_request_body_included_when_schema_present():
|
||||
rec = _record("POST", "/api/x", request_schema={"type": "object"})
|
||||
op = build_openapi([rec], title="x")["paths"]["/api/x"]["post"]
|
||||
assert op["requestBody"]["content"]["application/json"]["schema"] == {"type": "object"}
|
||||
Reference in New Issue
Block a user