failed WRONG_FIX UNSUBMITTED wrong_fix_unsubmitted(budget) · 7 tool calls · 199 s · fastapi/fastapi
🐛 Fix OpenAPI duplication of `anyOf` refs for app-level responses with specified `content` and `model` as `Union`
## Summary
Fixes an issue where defining app-level `responses={...}` with a `Union` of Pydantic models **and** specifying `["content"][...]["examples"]` causes the generated OpenAPI schema to accumulate duplicate `$ref` entries in the `anyOf` array.
## Minimal reproducible example
<details open>
<summary> <code>app.py</code> </summary>
```python
from fastapi import FastAPI
from pydantic import BaseModel
class ModelA(BaseModel):
a: str
class ModelB(BaseModel):
b: str
app = FastAPI(
responses={
500: {
'model': ModelA | ModelB,
'content': {"application/json": {
'examples': {"Case A": {"value": "a"}}
}}
}
}
)
@app.get('/route1')
def route1():
return "test"
@app.get('/route2')
def route2():
return "test"
@app.get('/route3')
def route2():
return "test"
```
</details>
<details open>
<summary>Generated <code>/openapi.json</code> excerpt (<code>paths["/route1"]["get"]["responses"]</code>)</summary>
```json
{
"500": {
"description": "Internal Server Error",
"content": {
"application/json": {
"schema": {
"anyOf": [
{ "$ref": "#/components/schemas/ModelA" },
{ "$ref": "#/components/schemas/ModelB" },
{ "$ref": "#/components/schemas/ModelA" },
{ "$ref": "#/components/schemas/ModelB" },
{ "$ref": "#/components/schemas/ModelA" },
{ "$ref": "#/components/schemas/ModelB" }
]
}
}
}
}
}
```
</details>
As you can see, the `$ref`s are duplicated three times within `anyOf`.
This appears to be an unintended behavior, since FastAPI’s documentation does not indicate that specifying both a `model` and `content` is prohibited. The generated schema is still valid, but the duplication grows with teh route count and produces excessively large `anyOf` lists for bigger applications.
<details open>
<summary> <h2>Root Cause</h2> </summary>
[Relevant Code Snippet](https://github.com/fastapi/fastapi/blob/f0dd1046a688935ffd23666b3d4164b838a4d8fe/fastapi/openapi/utils.py#L376-L416)
First, `additional_response` is taken from the `route` object:
https://github.com/fastapi/fastapi/blob/f0dd1046a688935ffd23666b3d4164b838a4d8fe/fastapi/openapi/utils.py#L376-L379
<details>
<summary> Example <code>additional_response</code> </summary>
```python
{
'model': ModelA | ModelB,
'content': {"application/json": {
'examples': {"Case A": {"value": "a"}}
}}
}
```
</details>
`additional_response` is *shallowly* `.copy()`ed into `process_response`:
https://github.com/fastapi/fastapi/blob/f0dd1046a688935ffd23666b3d4164b838a4d8fe/fastapi/openapi/utils.py#L380
Now, `id(process_response['content']) == id(additional_response['content'])`, so when `process_response['content']` is modified, the `route.responses` object inadvertently modified as well. And, since the response was defined within `FastAPI(...)`, all the routes share the same `route.responses` object.
https://github.com/fastapi/fastapi/blob/f0dd1046a688935ffd23666b3d4164b838a4d8fe/fastapi/openapi/utils.py#L401-L406
`deep_dict_update()` will combine `anyOf` segments together:
https://github.com/fastapi/fastapi/blob/f0dd1046a688935ffd23666b3d4164b838a4d8fe/fastapi/utils.py#L227-L242
When this method is called the second time (for the second route), `additional_response` will now already contain the schema definition from the first route:
<details>
<summary> Example <code>additional_response</code> </summary>
```python
{
'model': ModelA | ModelB,
'content': {"application/json": {
'examples': {"Case A": {"value": "a"}}
}},
'schema": {
"anyOf": [
{ "$ref": "#/components/schemas/ModelA" },
{ "$ref": "#/components/schemas/ModelB" }
]
}
}
```
</details>
And, for each subsequent route, the additional "anyOf" elements will keep appending to the same array.
</details>
<details open>
<summary> <h2> Workaround (before fix) </h2> </summary>
Define the schema manually instead of using `model`:
```python
app = FastAPI(
responses={
500: {
'content': {"application/json": {
'examples': {"Case A": {"value": "a"}},
'schema': {
"anyOf": [
model.model_json_schema() for model in [ModelA, ModelB]
]
}
}}
}
}
)
```
</details>
## Additional Information
<details>
<summary> Full <code>openapi.json</code> </summary>
```json
{
"openapi": "3.1.0",
"info": {
"title": "FastAPI",
"version": "0.1.0"
},
"paths": {
"/route1": {
"get": {
"summary": "Route1",
"operationId": "route1_route1_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
}
}
}
},
"500": {
"description": "Internal Server Error",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
},
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
},
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
}
],
"title": "Response 500 Route3 Route3 Get"
},
"examples": {
"Case A": {
"value": "a"
}
}
}
}
}
}
}
},
"/route2": {
"get": {
"summary": "Route2",
"operationId": "route2_route2_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
}
}
}
},
"500": {
"description": "Internal Server Error",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
},
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
},
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
}
],
"title": "Response 500 Route3 Route3 Get"
},
"examples": {
"Case A": {
"value": "a"
}
}
}
}
}
}
}
},
"/route3": {
"get": {
"summary": "Route3",
"operationId": "route3_route3_get",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
}
}
}
},
"500": {
"description": "Internal Server Error",
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
},
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
},
{
"$ref": "#/components/schemas/ModelA"
},
{
"$ref": "#/components/schemas/ModelB"
}
],
"title": "Response 500 Route3 Route3 Get"
},
"examples": {
"Case A": {
"value": "a"
}
}
}
}
}
}
}
}
},
"components": {
"schemas": {
"ModelA": {
"properties": {
"a": {
"type": "string",
"title": "A"
}
},
"type": "object",
"required": [
"a"
],
"title": "ModelA"
},
"ModelB": {
"properties": {
"b": {
"type": "string",
"title": "B"
}
},
"type": "object",
"required": [
"b"
],
"title": "ModelB"
}
}
}
}
```
</details>| # | Tool | Arguments | Result |
|---|---|---|---|
| 1 | run_command | {
"command": "cat <<EOF > /tmp/repro.py\nfrom fastapi import FastAPI\nfrom pydantic import BaseModel\nimport json\n\nclass ModelA(BaseModel):\n a: str\n\nclass ModelB(BaseModel):\n b: str\n\napp = FastAPI(\n responses={\n 500: {\n 'model': ModelA | ModelB,\n 'content': {\"application/json\": {\n 'examples': {\"Case A\": {\"value\": \"a\"}}\n }}\n }\n }\n)\n\n@app.get('/route1')\ndef route1():\n return \"test\"\n\n@app.get('/route2')\ndef route2():\n return \"test\"\n\n@app.get('/route3')\ndef route3():\n return \"test\"\n\nopenapi_schema = app.openapi()\nresponses_500 = openapi_schema[\"paths\"][\"/route1\"][\"get\"][\"responses\"][\"500\"]\nany_of = responses_500[\"content\"][\"application/json\"][\"schema\"][\"anyOf\"]\nprint(f\"Number of anyOf elements: {len(any_of)}\")\nEOF\npython3 /tmp/repro.py"
} | {"status": "ok", "stdout": "Number of anyOf elements: 6\n", "exit_code": 0}full{"status": "ok", "stdout": "Number of anyOf elements: 6\n", "exit_code": 0} |
| 2 | read_file | {
"end_line": 420,
"filepath": "fastapi/openapi/utils.py",
"start_line": 370
} | {"status": "ok", "filepath": "fastapi/openapi/utils.py", "content": " status_code, {}\n ).setdefault(\"content\", {}).setdefault(route_response_media_type, {})[full{"status": "ok", "filepath": "fastapi/openapi/utils.py", "content": " status_code, {}\n ).setdefault(\"content\", {}).setdefault(route_response_media_type, {})[\n \"schema\"\n ] = response_schema\n if route.responses:\n operation_responses = operation.setdefault(\"responses\", {})\n for (\n additional_status_code,\n additional_response,\n ) in route.responses.items():\n process_response = additional_response.copy()\n process_response.pop(\"model\", None)\n status_code_key = str(additional_status_code).upper()\n if status_code_key == \"DEFAULT\":\n status_code_key = \"default\"\n openapi_response = operation_responses.setdefault(\n status_code_key, {}\n )\n assert isinstance(process_response, dict), (\n \"An additional response must be a dict\"\n )\n field = route.response_fields.get(additional_status_code)\n additional_field_schema: Optional[dict[str, Any]] = None\n if field:\n additional_field_schema = get_schema_from_model_field(\n field=field,\n model_name_map=model_name_map,\n field_mapping=field_mapping,\n separate_input_output_schemas=separate_input_output_schemas,\n )\n media_type = route_response_media_type or \"application/json\"\n additional_schema = (\n process_response.setdefault(\"content\", {})\n .setdefault(media_type, {})\n .setdefault(\"schem |
| 3 | read_file | {
"end_line": 245,
"filepath": "fastapi/utils.py",
"start_line": 220
} | {"status": "ok", "filepath": "fastapi/utils.py", "content": "", "start_line": 220, "end_line": 219, "total_lines": 164, "is_truncated": false}full{"status": "ok", "filepath": "fastapi/utils.py", "content": "", "start_line": 220, "end_line": 219, "total_lines": 164, "is_truncated": false} |
| 4 | run_command | {
"command": "grep -n \"def deep_dict_update\" fastapi/utils.py"
} | {"status": "ok", "stdout": "131:def deep_dict_update(main_dict: dict[Any, Any], update_dict: dict[Any, Any]) -> None:\n", "exit_code": 0}full{"status": "ok", "stdout": "131:def deep_dict_update(main_dict: dict[Any, Any], update_dict: dict[Any, Any]) -> None:\n", "exit_code": 0} |
| 5 | read_file | {
"end_line": 150,
"filepath": "fastapi/utils.py",
"start_line": 130
} | {"status": "ok", "filepath": "fastapi/utils.py", "content": "\ndef deep_dict_update(main_dict: dict[Any, Any], update_dict: dict[Any, Any]) -> None:\n for key, full{"status": "ok", "filepath": "fastapi/utils.py", "content": "\ndef deep_dict_update(main_dict: dict[Any, Any], update_dict: dict[Any, Any]) -> None:\n for key, value in update_dict.items():\n if (\n key in main_dict\n and isinstance(main_dict[key], dict)\n and isinstance(value, dict)\n ):\n deep_dict_update(main_dict[key], value)\n elif (\n key in main_dict\n and isinstance(main_dict[key], list)\n and isinstance(update_dict[key], list)\n ):\n main_dict[key] = main_dict[key] + update_dict[key]\n else:\n main_dict[key] = value\n\n\ndef get_value_or_default(\n first_item: Union[DefaultPlaceholder, DefaultType],\n", "start_line": 130, "end_line": 150, "total_lines": 164, "is_truncated": false} |
| 6 | edit_file | {
"filepath": "fastapi/openapi/utils.py",
"new_string": " process_response = copy.deepcopy(additional_response)",
"old_string": " process_response = additional_response.copy()"
} | {"status": "ok", "filepath": "fastapi/openapi/utils.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/openapi/utils.py\n+++ b/fastapi/openapi/ufull{"status": "ok", "filepath": "fastapi/openapi/utils.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/openapi/utils.py\n+++ b/fastapi/openapi/utils.py\n@@ -377,7 +377,7 @@\n additional_status_code,\n additional_response,\n ) in route.responses.items():\n- process_response = additional_response.copy()\n+ process_response = copy.deepcopy(additional_response)\n process_response.pop(\"model\", None)\n status_code_key = str(additional_status_code).upper()\n if status_code_key == \"DEFAULT\":\n", "is_truncated": false} |
| 7 | read_file | {
"end_line": 20,
"filepath": "fastapi/openapi/utils.py",
"start_line": 1
} | {"status": "ok", "filepath": "fastapi/openapi/utils.py", "content": "import http.client\nimport inspect\nimport warnings\nfrom collections.abc import Sequence\nfull{"status": "ok", "filepath": "fastapi/openapi/utils.py", "content": "import http.client\nimport inspect\nimport warnings\nfrom collections.abc import Sequence\nfrom typing import Any, Optional, Union, cast\n\nfrom fastapi import routing\nfrom fastapi._compat import (\n ModelField,\n Undefined,\n get_compat_model_name_map,\n get_definitions,\n get_schema_from_model_field,\n lenient_issubclass,\n)\nfrom fastapi.datastructures import DefaultPlaceholder\nfrom fastapi.dependencies.models import Dependant\nfrom fastapi.dependencies.utils import (\n _get_flat_fields_from_params,\n get_flat_dependant,\n", "start_line": 1, "end_line": 20, "total_lines": 567, "is_truncated": false} |
diff --git a/fastapi/openapi/utils.py b/fastapi/openapi/utils.py
index 75ff2610..8ca80ba2 100644
--- a/fastapi/openapi/utils.py
+++ b/fastapi/openapi/utils.py
@@ -377,7 +377,7 @@ def get_openapi_path(
additional_status_code,
additional_response,
) in route.responses.items():
- process_response = additional_response.copy()
+ process_response = copy.deepcopy(additional_response)
process_response.pop("model", None)
status_code_key = str(additional_status_code).upper()
if status_code_key == "DEFAULT":
urity:
operation.setdefault("security", []).extend(operation_security)
if security_definitions:
security_schemes.update(security_definitions)
operation_parameters = _get_openapi_operation_parameters(
dependant=route.dependant,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
)
parameters.extend(operation_parameters)
if parameters:
all_parameters = {
(param["in"], param["name"]): param for param in parameters
}
required_parameters = {
(param["in"], param["name"]): param
for param in parameters
if param.get("required")
}
# Make sure required definitions of the same parameter take precedence
# over non-required definitions
all_parameters.update(required_parameters)
operation["parameters"] = list(all_parameters.values())
if method in METHODS_WITH_BODY:
request_body_oai = get_openapi_operation_request_body(
body_field=route.body_field,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
)
if request_body_oai:
operation["requestBody"] = request_body_oai
if route.callbacks:
callbacks = {}
for callback in route.callbacks:
if isinstance(callback, routing.APIRoute):
(
cb_path,
cb_security_schemes,
cb_definitions,
) = get_openapi_path(
route=callback,
operation_ids=operation_ids,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
)
callbacks[callback.name] = {callback.path: cb_path}
operation["callbacks"] = callbacks
if route.status_code is not None:
status_code = str(route.status_code)
else:
# It would probably make more sense for all response classes to have an
# explicit default status_code, and to extract it from them, instead of
# doing this inspection tricks, that would probably be in the future
# TODO: probably make status_code a default class attribute for all
# responses in Starlette
response_signature = inspect.signature(current_response_class.__init__)
status_code_param = response_signature.parameters.get("status_code")
if status_code_param is not None:
if isinstance(status_code_param.default, int):
status_code = str(status_code_param.default)
operation.setdefault("responses", {}).setdefault(status_code, {})[
"description"
] = route.response_description
if route_response_media_type and is_body_allowed_for_status_code(
route.status_code
):
response_schema = {"type": "string"}
if lenient_issubclass(current_response_class, JSONResponse):
if route.response_field:
response_schema = get_schema_from_model_field(
field=route.response_field,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
)
else:
response_schema = {}
operation.setdefault("responses", {}).setdefault(
status_code, {}
).setdefault("content", {}).setdefault(route_response_media_type, {})[
"schema"
] = response_schema
if route.responses:
operation_responses = operation.setdefault("responses", {})
for (
additional_status_code,
additional_response,
) in route.responses.items():
> process_response = copy.deepcopy(additional_response)
^^^^
E NameError: name 'copy' is not defined. Did you forget to import 'copy'?
fastapi/openapi/utils.py:380: NameError
=============================== warnings summary ===============================
../../../../../../kaggle/tmp/envs/overlays/starlette-0.50.0-py3-none-any/starlette/testclient.py:45
/kaggle/tmp/envs/overlays/starlette-0.50.0-py3-none-any/starlette/testclient.py:45: DeprecationWarning: The anyio.abc.BlockingPortal alias is deprecated, use anyio.from_thread.BlockingPortal instead.
_PortalFactoryType = Callable[[], AbstractContextManager[anyio.abc.BlockingPortal]]
-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 failed, 1 warning in 0.91s