forked from Archives/langchain
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
111 lines
3.7 KiB
Python
111 lines
3.7 KiB
Python
"""Quick and dirty representation for OpenAPI specs."""
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Dict, List, Tuple, Union
|
|
|
|
|
|
def dereference_refs(spec_obj: dict, full_spec: dict) -> Union[dict, list]:
|
|
"""Try to substitute $refs.
|
|
|
|
The goal is to get the complete docs for each endpoint in context for now.
|
|
|
|
In the few OpenAPI specs I studied, $refs referenced models
|
|
(or in OpenAPI terms, components) and could be nested. This code most
|
|
likely misses lots of cases.
|
|
"""
|
|
|
|
def _retrieve_ref_path(path: str, full_spec: dict) -> dict:
|
|
components = path.split("/")
|
|
if components[0] != "#":
|
|
raise RuntimeError(
|
|
"All $refs I've seen so far are uri fragments (start with hash)."
|
|
)
|
|
out = full_spec
|
|
for component in components[1:]:
|
|
out = out[component]
|
|
return out
|
|
|
|
def _dereference_refs(
|
|
obj: Union[dict, list], stop: bool = False
|
|
) -> Union[dict, list]:
|
|
if stop:
|
|
return obj
|
|
obj_out: Dict[str, Any] = {}
|
|
if isinstance(obj, dict):
|
|
for k, v in obj.items():
|
|
if k == "$ref":
|
|
# stop=True => don't dereference recursively.
|
|
return _dereference_refs(
|
|
_retrieve_ref_path(v, full_spec), stop=True
|
|
)
|
|
elif isinstance(v, list):
|
|
obj_out[k] = [_dereference_refs(el) for el in v]
|
|
elif isinstance(v, dict):
|
|
obj_out[k] = _dereference_refs(v)
|
|
else:
|
|
obj_out[k] = v
|
|
return obj_out
|
|
elif isinstance(obj, list):
|
|
return [_dereference_refs(el) for el in obj]
|
|
else:
|
|
return obj
|
|
|
|
return _dereference_refs(spec_obj)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReducedOpenAPISpec:
|
|
servers: List[dict]
|
|
description: str
|
|
endpoints: List[Tuple[str, str, dict]]
|
|
|
|
|
|
def reduce_openapi_spec(spec: dict, dereference: bool = True) -> ReducedOpenAPISpec:
|
|
"""Simplify/distill/minify a spec somehow.
|
|
|
|
I want a smaller target for retrieval and (more importantly)
|
|
I want smaller results from retrieval.
|
|
I was hoping https://openapi.tools/ would have some useful bits
|
|
to this end, but doesn't seem so.
|
|
"""
|
|
# 1. Consider only get, post endpoints.
|
|
endpoints = [
|
|
(f"{operation_name.upper()} {route}", docs.get("description"), docs)
|
|
for route, operation in spec["paths"].items()
|
|
for operation_name, docs in operation.items()
|
|
if operation_name in ["get", "post"]
|
|
]
|
|
|
|
# 2. Replace any refs so that complete docs are retrieved.
|
|
# Note: probably want to do this post-retrieval, it blows up the size of the spec.
|
|
if dereference:
|
|
endpoints = [
|
|
(name, description, dereference_refs(docs, spec))
|
|
for name, description, docs in endpoints
|
|
]
|
|
|
|
# 3. Strip docs down to required request args + happy path response.
|
|
def reduce_endpoint_docs(docs: dict) -> dict:
|
|
out = {}
|
|
if docs.get("description"):
|
|
out["description"] = docs.get("description")
|
|
if docs.get("parameters"):
|
|
out["parameters"] = [
|
|
parameter
|
|
for parameter in docs.get("parameters", [])
|
|
if parameter.get("required")
|
|
]
|
|
if "200" in docs["responses"]:
|
|
out["responses"] = docs["responses"]["200"]
|
|
return out
|
|
|
|
endpoints = [
|
|
(name, description, reduce_endpoint_docs(docs))
|
|
for name, description, docs in endpoints
|
|
]
|
|
return ReducedOpenAPISpec(
|
|
servers=spec["servers"],
|
|
description=spec["info"].get("description", ""),
|
|
endpoints=endpoints,
|
|
)
|