← eval_submission

fastapi_14463

resolved RESOLVED PASS · 13 tool calls · 39 s · fastapi/fastapi

Task input

🐛 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 calls (14)

{
 "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 refs in anyOf: {len(any_of)}\")\nEOF\npython3 /tmp/repro.py"
}

{"status": "ok", "stdout": "Number of refs in anyOf: 6\n", "exit_code": 0}

{
 "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, {})[\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

{
 "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}

{
 "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}

{
 "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, 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}

{
 "end_line": 410,
 "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                    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(\"schema\", {})\n                        )\n                        deep_dict_update(additional_schema, additional_field_schema)\n                    status_text: Optional[str] = status_code_ranges.get(\n                        str(additional_status_code).upper()\n                    ) or http.client.responses.get(int(additional_status_code))\n                    description = (\n", "start_line": 380, "end_line": 410, "total_lines": 567, "is_truncated": false}

{
 "filepath": "fastapi/openapi/utils.py",
 "new_string": "                    import copy\\n                    process_response = copy.deepcopy(additional_response)\\n                    process_response.pop(\\\"model\\\", None)",
 "old_string": "                    process_response = additional_response.copy()\\n                    process_response.pop(\\\"model\\\", None)"
}

{"status": "error", "error_type": "FileEditError", "error_message": "old_string not found in file (tried exact, flexible and regex matching)"}

{
 "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                    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}

{
 "filepath": "fastapi/openapi/utils.py",
 "new_string": "                    import copy\n                    process_response = copy.deepcopy(additional_response)\n                    process_response.pop(\"model\", None)",
 "old_string": "                    process_response = additional_response.copy()\n                    process_response.pop(\"model\", None)"
}

{"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,8 @@\n                     additional_status_code,\n                     additional_response,\n                 ) in route.responses.items():\n-                    process_response = additional_response.copy()\n+                    import 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}

{
 "command": "python3 /tmp/repro.py"
}

{"status": "ok", "stdout": "Number of refs in anyOf: 2\n", "exit_code": 0}

{
 "end_line": 50,
 "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\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    get_flat_params,\n    get_validation_alias,\n)\nfrom fastapi.encoders import jsonable_encoder\nfrom fastapi.exceptions import FastAPIDeprecationWarning\nfrom fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX\nfrom fastapi.openapi.models import OpenAPI\nfrom fastapi.params import Body, ParamTypes\nfrom fastapi.responses import Response\nfrom fastapi.types import ModelNameMap\nfrom fastapi.utils import (\n    deep_dict_update,\n    generate_operation_id_for_path,\n    is_body_allowed_for_status_code,\n)\nfrom pydantic import BaseModel\nfrom starlette.responses import JSONResponse\nfrom starlette.routing import BaseRoute\nfrom typing_extensions import Literal\n\nvalidation_error_definition = {\n    \"title\": \"ValidationError\",\n    \"type\": \"object\",\n    \"properties\": {\n        \"loc\": {\n            \"title\": \"Location\",\n            \"type\": \"array\",\n            \"items\": {\"anyOf\": [{\"type\": \"string\"}, {\"type\": \"integer\"}]},\n        },\n        \"msg\": {\"title\": \"Message\", \"type\": \"string\"},\n", "start_line": 1, "end_line": 50, "total_lines": 568, "is_truncated": false}

{
 "filepath": "fastapi/openapi/utils.py",
 "new_string": "import copy\nimport http.client",
 "old_string": "import http.client"
}

{"status": "ok", "filepath": "fastapi/openapi/utils.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/openapi/utils.py\n+++ b/fastapi/openapi/utils.py\n@@ -1,3 +1,4 @@\n+import copy\n import http.client\n import inspect\n import warnings\n", "is_truncated": false}

{
 "filepath": "fastapi/openapi/utils.py",
 "new_string": "                    process_response = copy.deepcopy(additional_response)\n                    process_response.pop(\"model\", None)",
 "old_string": "                    import copy\n                    process_response = copy.deepcopy(additional_response)\n                    process_response.pop(\"model\", None)"
}

{"status": "ok", "filepath": "fastapi/openapi/utils.py", "occurrences": 1, "strategy": "exact", "diff": "--- a/fastapi/openapi/utils.py\n+++ b/fastapi/openapi/utils.py\n@@ -378,7 +378,6 @@\n                     additional_status_code,\n                     additional_response,\n                 ) in route.responses.items():\n-                    import copy\n                     process_response = copy.deepcopy(additional_response)\n                     process_response.pop(\"model\", None)\n                     status_code_key = str(additional_status_code).upper()\n", "is_truncated": false}

[]

{"status": "ok", "patch_size": 745, "files_changed": 1}

Patch

diff --git a/fastapi/openapi/utils.py b/fastapi/openapi/utils.py
index 75ff2610..d56027b5 100644
--- a/fastapi/openapi/utils.py
+++ b/fastapi/openapi/utils.py
@@ -1,3 +1,4 @@
+import copy
 import http.client
 import inspect
 import warnings
@@ -377,7 +378,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":

Test output

.                                                                        [100%]
=============================== 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
1 passed, 1 warning in 0.84s