failed NO_PATCH NO_PATCH UNSUBMITTED empty_patch(budget) · 6 tool calls · 125 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)}\")\n# Expected: 2. If bug exists: 6 (2 * 3 routes)\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 | read_file | {
"end_line": 390,
"filepath": "fastapi/openapi/utils.py",
"start_line": 380
} | {"status": "ok", "filepath": "fastapi/openapi/utils.py", "content": " process_response = additional_response.copy()\n process_response.pop(\"model\", None)\n stfull{"status": "ok", "filepath": "fastapi/openapi/utils.py", "content": " 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", "start_line": 380, "end_line": 390, "total_lines": 567, "is_truncated": false} |