fix(proxy): document request body and response schemas for the Responses API in OpenAPI (#42802)

* fix(proxy): document responses API request and response schemas in openapi

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(proxy): namespace colliding openapi defs instead of overwriting existing components

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* chore(proxy): regenerate lazy openapi snapshot and dashboard schema types

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(proxy): require model and input in responses schema, document event stream, fix def collision refs

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(proxy): mark responses request fields readonly required

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(proxy): reuse existing OpenAPI components when a $defs entry has the same shape

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: kerry <kerry@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-24 01:28:34 +00:00 • committed by GitHub
parent 61996e1837
commit 8e74bb0d23
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 11654 additions and 83 deletions

File diff suppressed because it is too large Load diff

View file

@ -1,5 +1,8 @@
from collections.abc import Mapping, Sequence
from typing import Final, TypeAlias, Union
from types import MappingProxyType
from typing import Final, TypeAlias, Union, cast
from pydantic import TypeAdapter
from litellm._logging import verbose_proxy_logger
@ -28,7 +31,7 @@ class CustomOpenAPISpec:
"/openai/deployments/{model}/embeddings",
]
RESPONSES_API_PATHS = ["/v1/responses", "/responses"]
RESPONSES_API_PATHS = ["/v1/responses", "/responses", "/openai/v1/responses"]
@staticmethod
def _as_object(node: JsonValue) -> JsonObject:
@ -44,26 +47,18 @@ class CustomOpenAPISpec:
return CustomOpenAPISpec._as_object(components.setdefault("schemas", {}))
@staticmethod
def get_pydantic_schema(model_class) -> JsonObject | None:
def get_pydantic_schema(model_class: type) -> JsonObject | None:
"""
Get JSON schema from a Pydantic model, handling both v1 and v2 APIs.
Get JSON schema for a request or response model class, including TypedDicts.
Args:
model_class: Pydantic model class
model_class: Pydantic model class or TypedDict
Returns:
JSON schema dict or None if failed
"""
try:
# Try Pydantic v2 method first
return model_class.model_json_schema()
except AttributeError:
try:
# Fallback to Pydantic v1 method
return model_class.schema()
except AttributeError:
# If both methods fail, return None
return None
return cast(JsonObject, TypeAdapter(model_class).json_schema()) # cast-ok: pydantic returns dict[str, Any]
except Exception as e:
# FastAPI 0.120+ may fail schema generation for certain types (e.g., openai.Timeout)
# Log the error and return None to skip schema generation for this model
@ -83,13 +78,18 @@ class CustomOpenAPISpec:
# Ensure components/schemas structure exists
_ = CustomOpenAPISpec._components_schemas(openapi_schema)
# Add the schema
CustomOpenAPISpec._move_defs_to_components(openapi_schema, {schema_name: schema_def})
defs: Final[Mapping[str, JsonValue]] = (
CustomOpenAPISpec._as_object(schema_def["$defs"]) if "$defs" in schema_def else MappingProxyType({})
)
renames: Final = CustomOpenAPISpec._move_defs_to_components(openapi_schema, defs, schema_name)
schemas: Final = CustomOpenAPISpec._components_schemas(openapi_schema)
schemas[schema_name] = CustomOpenAPISpec._rewrite_defs_refs(schema_def, renames)
@staticmethod
def _expanded_request_field(field_name: str, field_def: JsonValue) -> JsonValue:
expanded: Final = CustomOpenAPISpec._rewrite_defs_refs(
CustomOpenAPISpec._expand_field_definition(CustomOpenAPISpec._as_object(field_def))
CustomOpenAPISpec._expand_field_definition(CustomOpenAPISpec._as_object(field_def)),
MappingProxyType({}),
)
if field_name != "messages":
return expanded
@ -127,13 +127,6 @@ class CustomOpenAPISpec:
schema_properties = CustomOpenAPISpec._as_object(actual_schema.get("properties"))
required_fields = actual_schema.get("required", [])
# Extract $defs and add them to components/schemas
# This fixes Pydantic v2 $defs not being resolvable in Swagger/OpenAPI
if "$defs" in actual_schema:
CustomOpenAPISpec._move_defs_to_components(
openapi_schema, CustomOpenAPISpec._as_object(actual_schema["$defs"])
)
# Create an expanded inline schema instead of just a $ref
# This makes Swagger UI show all individual fields in the request body editor
expanded_schema: JsonObject = {
@ -161,7 +154,9 @@ class CustomOpenAPISpec:
]
@staticmethod
def _move_defs_to_components(openapi_schema: JsonObject, defs: Mapping[str, JsonValue]) -> None:
def _move_defs_to_components(
openapi_schema: JsonObject, defs: Mapping[str, JsonValue], namespace: str
) -> Mapping[str, str]:
"""
Move $defs from Pydantic v2 schema to OpenAPI components/schemas.
This makes the definitions resolvable in Swagger/OpenAPI viewers.
@ -169,36 +164,68 @@ class CustomOpenAPISpec:
Args:
openapi_schema: The OpenAPI schema dict to modify
defs: The $defs dictionary from Pydantic schema
namespace: Prefix used to rename defs that would overwrite an existing component
Returns:
Map of original def names to renamed component names for collision cases
"""
if not defs:
return
# Ensure components/schemas exists
schemas: Final = CustomOpenAPISpec._components_schemas(openapi_schema)
# Add each definition to components/schemas
renames: Final = CustomOpenAPISpec._fixed_renames(schemas, defs, namespace, MappingProxyType({}))
for def_name, def_schema in defs.items():
# Recursively rewrite any nested $defs references within this definition
schemas[def_name] = CustomOpenAPISpec._rewrite_defs_refs(def_schema)
# If this definition also has $defs, process them recursively
def_object = CustomOpenAPISpec._as_object(def_schema)
if "$defs" in def_object:
CustomOpenAPISpec._move_defs_to_components(
openapi_schema, CustomOpenAPISpec._as_object(def_object["$defs"])
)
if def_name in schemas and def_name not in renames:
continue
schemas[renames.get(def_name, def_name)] = CustomOpenAPISpec._rewrite_defs_refs(def_schema, renames)
return renames
@staticmethod
def _rewritten_defs_entry(key: str, value: JsonValue) -> JsonValue:
def _def_collisions(
schemas: JsonObject, defs: Mapping[str, JsonValue], namespace: str, renames: Mapping[str, str]
) -> Mapping[str, str]:
return MappingProxyType(
{
name: f"{namespace}_{name}"
for name, d in defs.items()
if name in schemas
and not CustomOpenAPISpec._same_shape(schemas[name], CustomOpenAPISpec._rewrite_defs_refs(d, renames))
}
)
@staticmethod
def _same_shape(existing: JsonValue, incoming: JsonValue) -> bool:
if existing == incoming:
return True
existing_obj: Final = CustomOpenAPISpec._as_object(existing)
incoming_obj: Final = CustomOpenAPISpec._as_object(incoming)
existing_props: Final = CustomOpenAPISpec._as_object(existing_obj.get("properties"))
incoming_props: Final = CustomOpenAPISpec._as_object(incoming_obj.get("properties"))
if not existing_props or not incoming_props:
return False
return existing_props.keys() == incoming_props.keys() and frozenset(
x for x in CustomOpenAPISpec._as_array(existing_obj.get("required")) if isinstance(x, str)
) == frozenset(x for x in CustomOpenAPISpec._as_array(incoming_obj.get("required")) if isinstance(x, str))
@staticmethod
def _fixed_renames(
schemas: JsonObject, defs: Mapping[str, JsonValue], namespace: str, renames: Mapping[str, str]
) -> Mapping[str, str]:
next_renames: Final = MappingProxyType(
{**renames, **CustomOpenAPISpec._def_collisions(schemas, defs, namespace, renames)}
)
if next_renames == renames:
return renames
return CustomOpenAPISpec._fixed_renames(schemas, defs, namespace, next_renames)
@staticmethod
def _rewritten_defs_entry(key: str, value: JsonValue, renames: Mapping[str, str]) -> JsonValue:
if key == "$ref" and isinstance(value, str) and value.startswith("#/$defs/"):
# Rewrite the reference to use components/schemas
def_name: Final = value.replace("#/$defs/", "")
return f"#/components/schemas/{def_name}"
return f"#/components/schemas/{renames.get(def_name, def_name)}"
# Recursively process nested structures
return CustomOpenAPISpec._rewrite_defs_refs(value)
return CustomOpenAPISpec._rewrite_defs_refs(value, renames)
@staticmethod
def _rewrite_defs_refs(schema: JsonValue) -> JsonValue:
def _rewrite_defs_refs(schema: JsonValue, renames: Mapping[str, str]) -> JsonValue:
"""
Recursively rewrite $ref values from #/$defs/... to #/components/schemas/...
This converts Pydantic v2 references to OpenAPI-compatible references.
@ -211,12 +238,12 @@ class CustomOpenAPISpec:
"""
if isinstance(schema, dict):
return {
key: CustomOpenAPISpec._rewritten_defs_entry(key, value)
key: CustomOpenAPISpec._rewritten_defs_entry(key, value, renames)
for key, value in schema.items()
if key != "$defs"
}
if isinstance(schema, list):
return [CustomOpenAPISpec._rewrite_defs_refs(item) for item in schema]
return [CustomOpenAPISpec._rewrite_defs_refs(item, renames) for item in schema]
return schema
@staticmethod

View file

@ -12,6 +12,7 @@ from uuid import uuid4
import fastapi
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from fastapi.responses import JSONResponse
from openai.types.responses import ResponseItemList
from openai.types.responses.response_create_params import ResponseInputParam
from pydantic import BaseModel, ConfigDict, ValidationError
from starlette.websockets import WebSocket, WebSocketDisconnect
@ -48,6 +49,20 @@ if TYPE_CHECKING:
router: Final = APIRouter()
_ResponseDocSchemas = dict[int | str, dict[str, Any]] # pyright: ignore[reportExplicitAny] # fastapi's responses kwarg
RESPONSES_API_RESPONSE_SCHEMAS: Final[_ResponseDocSchemas] = {200: {"model": ResponsesAPIResponse}}
RESPONSES_API_CREATE_RESPONSE_SCHEMAS: Final[_ResponseDocSchemas] = {
200: {
"model": ResponsesAPIResponse,
"content": {
"text/event-stream": {"schema": {"type": "string", "description": "Server sent events when stream=true"}}
},
}
}
DELETE_RESPONSE_SCHEMAS: Final[_ResponseDocSchemas] = {200: {"model": DeleteResponseResult}}
RESPONSE_ITEM_LIST_SCHEMAS: Final[_ResponseDocSchemas] = {200: {"model": ResponseItemList}}
_user_api_key_auth_dep: Final = Depends(user_api_key_auth)
_RESPONSES_TAGS: Final[list[str | Enum]] = ["responses"] # mutable-ok: fastapi's route signature requires list tags
@ -181,16 +196,19 @@ async def _resolve_cursor_model_variant_before_auth(request: Request) -> None:
"/v1/responses",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSES_API_CREATE_RESPONSE_SCHEMAS,
)
@router.post(
"/responses",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSES_API_CREATE_RESPONSE_SCHEMAS,
)
@router.post(
"/openai/v1/responses",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSES_API_CREATE_RESPONSE_SCHEMAS,
)
async def responses_api(
request: Request,
@ -664,16 +682,19 @@ async def cursor_chat_completions(
"/v1/responses/{response_id}",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSES_API_RESPONSE_SCHEMAS,
)
@router.get(
"/responses/{response_id}",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSES_API_RESPONSE_SCHEMAS,
)
@router.get(
"/openai/v1/responses/{response_id}",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSES_API_RESPONSE_SCHEMAS,
)
async def get_response(
response_id: str,
@ -777,16 +798,19 @@ async def get_response(
"/v1/responses/{response_id}",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=DELETE_RESPONSE_SCHEMAS,
)
@router.delete(
"/responses/{response_id}",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=DELETE_RESPONSE_SCHEMAS,
)
@router.delete(
"/openai/v1/responses/{response_id}",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=DELETE_RESPONSE_SCHEMAS,
)
async def delete_response(
response_id: str,
@ -883,16 +907,19 @@ async def delete_response(
"/v1/responses/{response_id}/input_items",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSE_ITEM_LIST_SCHEMAS,
)
@router.get(
"/responses/{response_id}/input_items",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSE_ITEM_LIST_SCHEMAS,
)
@router.get(
"/openai/v1/responses/{response_id}/input_items",
dependencies=[Depends(user_api_key_auth)],
tags=["responses"],
responses=RESPONSE_ITEM_LIST_SCHEMAS,
)
async def get_response_input_items(
response_id: str,

View file

@ -1301,8 +1301,8 @@ class ResponsesAPIOptionalRequestParams(TypedDict, total=False):
class ResponsesAPIRequestParams(ResponsesAPIOptionalRequestParams, total=False):
"""TypedDict for request parameters supported by the responses API."""
input: str | ResponseInputParam
model: str
input: Required[ReadOnly[str | ResponseInputParam]]
model: Required[ReadOnly[str]]
class OutputTokensDetails(BaseLiteLLMOpenAIResponseObject):

View file

@ -1,21 +1,43 @@
import pytest
from integration._support.client import Gateway, object_value
from typing import Final
from integration._support.client import Gateway, object_value, string_value
from pydantic import JsonValue
def _assert_responses_post_is_documented(openapi: dict[str, JsonValue]) -> None:
post: dict[str, JsonValue] = object_value(object_value(object_value(openapi["paths"])["/v1/responses"])["post"])
body: dict[str, JsonValue] = object_value(post["requestBody"])
schema: dict[str, JsonValue] = object_value(
object_value(object_value(body["content"])["application/json"])["schema"]
)
properties: dict[str, JsonValue] = object_value(schema.get("properties"))
assert "model" in properties and "input" in properties, schema
ok: dict[str, JsonValue] = object_value(object_value(object_value(post)["responses"])["200"])
assert "schema" in object_value(object_value(ok["content"])["application/json"]), ok
def _operation(openapi: dict[str, JsonValue], path: str, method: str) -> dict[str, JsonValue]:
return object_value(object_value(object_value(openapi["paths"])[path])[method])
def _ok_schema_properties(openapi: dict[str, JsonValue], operation: dict[str, JsonValue]) -> dict[str, JsonValue]:
ok: Final = object_value(object_value(operation["responses"])["200"])
schema: Final = object_value(object_value(object_value(ok["content"])["application/json"])["schema"])
if "$ref" in schema:
name: Final = string_value(schema["$ref"]).rsplit("/", 1)[-1]
return object_value(object_value(object_value(object_value(openapi["components"])["schemas"])[name])["properties"])
assert "properties" in schema, ok
return object_value(schema["properties"])
def test_v1_responses_post_declares_a_request_body_and_response_schema(gateway: Gateway) -> None:
pytest.skip("BUG: POST /v1/responses takes a raw Request, so /openapi.json documents no body or response schema")
openapi: dict[str, JsonValue] = gateway.get("/openapi.json")
_assert_responses_post_is_documented(openapi)
post: Final = _operation(openapi, "/v1/responses", "post")
body: Final = object_value(post["requestBody"])
schema: Final = object_value(object_value(object_value(body["content"])["application/json"])["schema"])
properties: Final = object_value(schema.get("properties"))
assert {"model", "input", "instructions", "tools", "previous_response_id", "background", "stream"} <= set(
properties
), sorted(properties)
assert {"id", "object", "output", "usage"} <= set(_ok_schema_properties(openapi, post)), post["responses"]
assert "tool_calls" in object_value(
object_value(object_value(object_value(openapi["components"])["schemas"])["Message"])["properties"]
)
def test_v1_responses_by_id_routes_declare_response_schemas(gateway: Gateway) -> None:
openapi: dict[str, JsonValue] = gateway.get("/openapi.json")
get: Final = _operation(openapi, "/v1/responses/{response_id}", "get")
assert {"id", "object", "output"} <= set(_ok_schema_properties(openapi, get)), get["responses"]
delete: Final = _operation(openapi, "/v1/responses/{response_id}", "delete")
assert {"id", "object", "deleted"} <= set(_ok_schema_properties(openapi, delete)), delete["responses"]
items: Final = _operation(openapi, "/v1/responses/{response_id}/input_items", "get")
assert {"data", "object", "has_more"} <= set(_ok_schema_properties(openapi, items)), items["responses"]

View file

@ -153,7 +153,7 @@ def test_move_defs_to_components():
},
}
CustomOpenAPISpec._move_defs_to_components(openapi_schema=openapi_schema, defs=defs)
CustomOpenAPISpec._move_defs_to_components(openapi_schema=openapi_schema, defs=defs, namespace="Req")
assert "components" in openapi_schema
assert "schemas" in openapi_schema["components"]
@ -185,7 +185,7 @@ def test_rewrite_defs_refs():
},
}
rewritten = CustomOpenAPISpec._rewrite_defs_refs(schema=schema)
rewritten = CustomOpenAPISpec._rewrite_defs_refs(schema=schema, renames={})
assert "$defs" not in rewritten
assert (
@ -196,3 +196,197 @@ def test_rewrite_defs_refs():
rewritten["properties"]["messages"]["items"]["anyOf"][1]["$ref"]
== "#/components/schemas/AssistantMessage"
)
def test_get_pydantic_schema_generates_schema_for_responses_request_typed_dict():
from litellm.types.llms.openai import ResponsesAPIRequestParams
schema = CustomOpenAPISpec.get_pydantic_schema(ResponsesAPIRequestParams)
assert schema is not None
properties = schema["properties"]
assert isinstance(properties, dict)
for field in (
"model",
"input",
"instructions",
"tools",
"previous_response_id",
"background",
"stream",
):
assert field in properties
def test_responses_api_paths_covers_all_three_routes():
assert CustomOpenAPISpec.RESPONSES_API_PATHS == [
"/v1/responses",
"/responses",
"/openai/v1/responses",
]
def test_add_schema_to_components_renames_colliding_def_instead_of_overwriting():
openapi = {
"components": {
"schemas": {
"Message": {"type": "object", "properties": {"content": {"type": "string"}}},
}
}
}
CustomOpenAPISpec.add_schema_to_components(
openapi,
"Req",
{
"type": "object",
"properties": {"m": {"$ref": "#/$defs/Message"}, "n": {"$ref": "#/$defs/Other"}},
"$defs": {
"Message": {"type": "object", "properties": {"role": {"type": "string"}}},
"Other": {"type": "integer"},
},
},
)
schemas = openapi["components"]["schemas"]
assert schemas["Message"] == {"type": "object", "properties": {"content": {"type": "string"}}}
assert schemas["Req_Message"] == {"type": "object", "properties": {"role": {"type": "string"}}}
assert schemas["Other"] == {"type": "integer"}
assert schemas["Req"]["properties"]["m"]["$ref"] == "#/components/schemas/Req_Message"
assert schemas["Req"]["properties"]["n"]["$ref"] == "#/components/schemas/Other"
assert "$defs" not in schemas["Req"]
def test_add_schema_to_components_keeps_name_for_identical_existing_def():
openapi = {"components": {"schemas": {"Same": {"type": "integer"}}}}
CustomOpenAPISpec.add_schema_to_components(
openapi,
"Req",
{
"type": "object",
"properties": {"s": {"$ref": "#/$defs/Same"}},
"$defs": {"Same": {"type": "integer"}},
},
)
schemas = openapi["components"]["schemas"]
assert "Req_Same" not in schemas
assert schemas["Req"]["properties"]["s"]["$ref"] == "#/components/schemas/Same"
def test_responses_request_params_schema_requires_model_and_input():
from litellm.types.llms.openai import ResponsesAPIRequestParams
schema = CustomOpenAPISpec.get_pydantic_schema(ResponsesAPIRequestParams)
assert schema is not None
assert set(schema["required"]) == {"model", "input"}
def test_move_defs_to_components_renames_defs_whose_refs_point_at_renamed_defs():
openapi = {
"components": {
"schemas": {
"Inner": {"type": "string"},
"Wrapper": {"$ref": "#/components/schemas/Inner"},
}
}
}
renames = CustomOpenAPISpec._move_defs_to_components(
openapi,
{
"Inner": {"type": "integer"},
"Wrapper": {"$ref": "#/$defs/Inner"},
},
"NS",
)
schemas = openapi["components"]["schemas"]
assert renames == {"Inner": "NS_Inner", "Wrapper": "NS_Wrapper"}
assert schemas["Inner"] == {"type": "string"}
assert schemas["NS_Inner"] == {"type": "integer"}
assert schemas["Wrapper"] == {"$ref": "#/components/schemas/Inner"}
assert schemas["NS_Wrapper"] == {"$ref": "#/components/schemas/NS_Inner"}
def test_add_schema_to_components_keeps_name_for_same_shape_existing_def():
openapi = {
"components": {
"schemas": {
"Block": {
"type": "object",
"properties": {"type": {"type": "string"}, "x": {"type": "string"}},
"required": ["type", "x"],
"additionalProperties": True,
},
}
}
}
CustomOpenAPISpec.add_schema_to_components(
openapi,
"Req",
{
"type": "object",
"properties": {"b": {"$ref": "#/$defs/Block"}},
"$defs": {
"Block": {
"type": "object",
"properties": {"type": {"type": "string"}, "x": {"type": "string"}},
"required": ["type", "x"],
},
},
},
)
schemas = openapi["components"]["schemas"]
assert "Req_Block" not in schemas
assert schemas["Block"]["additionalProperties"] is True
assert schemas["Req"]["properties"]["b"]["$ref"] == "#/components/schemas/Block"
def test_add_schema_to_components_renames_def_with_different_required_set():
openapi = {
"components": {
"schemas": {
"Block": {
"type": "object",
"properties": {
"keys": {"type": "array"},
"type": {"type": "string"},
"x": {"type": "string"},
"y": {"type": "string"},
},
"required": ["type", "x", "y"],
},
}
}
}
CustomOpenAPISpec.add_schema_to_components(
openapi,
"Req",
{
"type": "object",
"properties": {"b": {"$ref": "#/$defs/Block"}},
"$defs": {
"Block": {
"type": "object",
"properties": {
"keys": {"type": "array"},
"type": {"type": "string"},
"x": {"type": "string"},
"y": {"type": "string"},
},
"required": ["keys", "type", "x", "y"],
},
},
},
)
schemas = openapi["components"]["schemas"]
assert schemas["Block"]["required"] == ["type", "x", "y"]
assert schemas["Req_Block"]["required"] == ["keys", "type", "x", "y"]
assert schemas["Req"]["properties"]["b"]["$ref"] == "#/components/schemas/Req_Block"

View file

@ -2427,3 +2427,38 @@ class TestResponsesInputTokens:
assert response.status_code == 429, response.text
assert response.json()["error"]["message"] == "rate limited"
def test_responses_routes_document_response_models_in_openapi_schema():
from typing import cast
from fastapi import FastAPI
from litellm.proxy.response_api_endpoints.endpoints import router
def as_object(value: object) -> dict[str, object]:
assert isinstance(value, dict)
return cast(dict[str, object], value)
openapi_app = FastAPI()
openapi_app.include_router(router)
openapi: Final = cast(dict[str, object], openapi_app.openapi())
def ok_200_properties(path: str, method: str) -> dict[str, object]:
operation: Final = as_object(as_object(as_object(openapi)["paths"])[path])[method]
schema: Final = as_object(
as_object(
as_object(as_object(as_object(as_object(operation)["responses"])["200"])["content"])["application/json"]
)["schema"]
)
ref: Final = schema["$ref"]
assert isinstance(ref, str)
component: Final = ref.rsplit("/", 1)[-1]
return as_object(
as_object(as_object(as_object(as_object(openapi)["components"])["schemas"])[component])["properties"]
)
assert "output" in ok_200_properties("/v1/responses", "post")
assert "output" in ok_200_properties("/v1/responses/{response_id}", "get")
assert "deleted" in ok_200_properties("/v1/responses/{response_id}", "delete")
assert "data" in ok_200_properties("/v1/responses/{response_id}/input_items", "get")

File diff suppressed because it is too large Load diff