We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
OpenAPI: HTTP_422 response does not use custom media_type
**Describe the bug**
FastAPI automatically adds an HTTP_422 response to all paths in the OpenAPI specification that have parameters or request body. This response does not use the media_type of response_class if any custom defined. Furthermore, it overwrites any error object format with the default one.

**To Reproduce**
Create a path with parameters and add custom response_class to decorator. Add custom exception handlers that reformat the default error responses as per your liking. Then observe generated openapi.json

```python
from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from starlette import status
from starlette.responses import JSONResponse
from . import schemas

app = FastAPI()

class JsonApiResponse(JSONResponse):
    media_type = 'application/vnd+json.api'

@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc: HTTPException) -> JsonApiResponse:
    headers = getattr(exc, "headers", None)
    content = schemas.ErrorResponse(errors=[dict(title="Bad request", detail=exc.detail, status=exc.status_code)]).dict()
    status_code = exc.status_code
    if headers:
        return JsonApiResponse(content=content, status_code=status_code, headers=headers)
    else:
        return JsonApiResponse(content=content, status_code=status_code)

@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(request, exc: RequestValidationError) -> JsonApiResponse:
    http422 = status.HTTP_422_UNPROCESSABLE_ENTITY
    return JsonApiResponse(
        content=schemas.ErrorResponse(errors=[
            dict(title=err['type'], detail=err['msg'], source='/'.join(err['loc']), status=http422)
            for err in exc.errors()
        ]).dict(),
        status_code=http422,
    )

@app.post('/customers',
         status_code=status.HTTP_201_CREATED,
         response_model=schemas.CustomerDetailsResponse,
         response_class=JsonApiResponse,
         )
def customer_create(data: schemas.Customer = Body(..., media_type='application/vnd+json.api', embed=True)):
    created_customer = {**data.dict(), **{'id': '1'}}
    return {'data': created_customer}
``` 

The openapi.json will include the unwanted 422 response with the FastAPI default error object definitions:

```yaml
  # ...
  '422':
    description: Validation Error
    content:
      application/json:
        schema:
          "$ref": "#/components/schemas/HTTPValidationError"
```

**Expected behavior**
At least, the media_type of the response_class should be respected. But the best would be if the 422 would not be added to the specification unless requested via the path decorator. Or if the 422 definitions of mine were respected.

```python
@app.post('/customers',
         status_code=status.HTTP_201_CREATED,
         response_model=schemas.CustomerDetailsResponse,
         response_class=JsonApiResponse,
         responses={
             422: {
                 'model': schemas.ErrorResponse
             },
         })
data: schemas.Customer = Body(..., media_type='application/vnd+json.api', embed=True)):
    pass
```

**Environment:**
 - OS: masOS 10.14.6
 - Python: 3.6.5
 - FastAPI: 0.35.0

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `fastapi/openapi/utils.py`
Content:
```
1 import http.client
2 from typing import Any, Dict, List, Optional, Sequence, Tuple, Type, cast
3 
4 from fastapi import routing
5 from fastapi.dependencies.models import Dependant
6 from fastapi.dependencies.utils import get_flat_dependant
7 from fastapi.encoders import jsonable_encoder
8 from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX
9 from fastapi.openapi.models import OpenAPI
10 from fastapi.params import Body, Param
11 from fastapi.utils import (
12     generate_operation_id_for_path,
13     get_flat_models_from_routes,
14     get_model_definitions,
15 )
16 from pydantic.fields import Field
17 from pydantic.schema import field_schema, get_model_name_map
18 from pydantic.utils import lenient_issubclass
19 from starlette.responses import JSONResponse
20 from starlette.routing import BaseRoute
21 from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
22 
23 validation_error_definition = {
24     "title": "ValidationError",
25     "type": "object",
26     "properties": {
27         "loc": {"title": "Location", "type": "array", "items": {"type": "string"}},
28         "msg": {"title": "Message", "type": "string"},
29         "type": {"title": "Error Type", "type": "string"},
30     },
31     "required": ["loc", "msg", "type"],
32 }
33 
34 validation_error_response_definition = {
35     "title": "HTTPValidationError",
36     "type": "object",
37     "properties": {
38         "detail": {
39             "title": "Detail",
40             "type": "array",
41             "items": {"$ref": REF_PREFIX + "ValidationError"},
42         }
43     },
44 }
45 
46 
47 def get_openapi_params(dependant: Dependant) -> List[Field]:
48     flat_dependant = get_flat_dependant(dependant)
49     return (
50         flat_dependant.path_params
51         + flat_dependant.query_params
52         + flat_dependant.header_params
53         + flat_dependant.cookie_params
54     )
55 
56 
57 def get_openapi_security_definitions(flat_dependant: Dependant) -> Tuple[Dict, List]:
58     security_definitions = {}
59     operation_security = []
60     for security_requirement in flat_dependant.security_requirements:
61         security_definition = jsonable_encoder(
62             security_requirement.security_scheme.model,
63             by_alias=True,
64             include_none=False,
65         )
66         security_name = security_requirement.security_scheme.scheme_name
67         security_definitions[security_name] = security_definition
68         operation_security.append({security_name: security_requirement.scopes})
69     return security_definitions, operation_security
70 
71 
72 def get_openapi_operation_parameters(
73     all_route_params: Sequence[Field]
74 ) -> Tuple[Dict[str, Dict], List[Dict[str, Any]]]:
75     definitions: Dict[str, Dict] = {}
76     parameters = []
77     for param in all_route_params:
78         schema = param.schema
79         schema = cast(Param, schema)
80         if "ValidationError" not in definitions:
81             definitions["ValidationError"] = validation_error_definition
82             definitions["HTTPValidationError"] = validation_error_response_definition
83         parameter = {
84             "name": param.alias,
85             "in": schema.in_.value,
86             "required": param.required,
87             "schema": field_schema(param, model_name_map={})[0],
88         }
89         if schema.description:
90             parameter["description"] = schema.description
91         if schema.deprecated:
92             parameter["deprecated"] = schema.deprecated
93         parameters.append(parameter)
94     return definitions, parameters
95 
96 
97 def get_openapi_operation_request_body(
98     *, body_field: Optional[Field], model_name_map: Dict[Type, str]
99 ) -> Optional[Dict]:
100     if not body_field:
101         return None
102     assert isinstance(body_field, Field)
103     body_schema, _, _ = field_schema(
104         body_field, model_name_map=model_name_map, ref_prefix=REF_PREFIX
105     )
106     body_field.schema = cast(Body, body_field.schema)
107     request_media_type = body_field.schema.media_type
108     required = body_field.required
109     request_body_oai: Dict[str, Any] = {}
110     if required:
111         request_body_oai["required"] = required
112     request_body_oai["content"] = {request_media_type: {"schema": body_schema}}
113     return request_body_oai
114 
115 
116 def generate_operation_id(*, route: routing.APIRoute, method: str) -> str:
117     if route.operation_id:
118         return route.operation_id
119     path: str = route.path_format
120     return generate_operation_id_for_path(name=route.name, path=path, method=method)
121 
122 
123 def generate_operation_summary(*, route: routing.APIRoute, method: str) -> str:
124     if route.summary:
125         return route.summary
126     return route.name.replace("_", " ").title()
127 
128 
129 def get_openapi_operation_metadata(*, route: routing.APIRoute, method: str) -> Dict:
130     operation: Dict[str, Any] = {}
131     if route.tags:
132         operation["tags"] = route.tags
133     operation["summary"] = generate_operation_summary(route=route, method=method)
134     if route.description:
135         operation["description"] = route.description
136     operation["operationId"] = generate_operation_id(route=route, method=method)
137     if route.deprecated:
138         operation["deprecated"] = route.deprecated
139     return operation
140 
141 
142 def get_openapi_path(
143     *, route: routing.APIRoute, model_name_map: Dict[Type, str]
144 ) -> Tuple[Dict, Dict, Dict]:
145     path = {}
146     security_schemes: Dict[str, Any] = {}
147     definitions: Dict[str, Any] = {}
148     assert route.methods is not None, "Methods must be a list"
149     if route.include_in_schema:
150         for method in route.methods:
151             operation = get_openapi_operation_metadata(route=route, method=method)
152             parameters: List[Dict] = []
153             flat_dependant = get_flat_dependant(route.dependant)
154             security_definitions, operation_security = get_openapi_security_definitions(
155                 flat_dependant=flat_dependant
156             )
157             if operation_security:
158                 operation.setdefault("security", []).extend(operation_security)
159             if security_definitions:
160                 security_schemes.update(security_definitions)
161             all_route_params = get_openapi_params(route.dependant)
162             validation_definitions, operation_parameters = get_openapi_operation_parameters(
163                 all_route_params=all_route_params
164             )
165             definitions.update(validation_definitions)
166             parameters.extend(operation_parameters)
167             if parameters:
168                 operation["parameters"] = parameters
169             if method in METHODS_WITH_BODY:
170                 request_body_oai = get_openapi_operation_request_body(
171                     body_field=route.body_field, model_name_map=model_name_map
172                 )
173                 if request_body_oai:
174                     operation["requestBody"] = request_body_oai
175                     if "ValidationError" not in definitions:
176                         definitions["ValidationError"] = validation_error_definition
177                         definitions[
178                             "HTTPValidationError"
179                         ] = validation_error_response_definition
180             if route.responses:
181                 for (additional_status_code, response) in route.responses.items():
182                     assert isinstance(
183                         response, dict
184                     ), "An additional response must be a dict"
185                     field = route.response_fields.get(additional_status_code)
186                     if field:
187                         response_schema, _, _ = field_schema(
188                             field, model_name_map=model_name_map, ref_prefix=REF_PREFIX
189                         )
190                         response.setdefault("content", {}).setdefault(
191                             "application/json", {}
192                         )["schema"] = response_schema
193                     status_text = http.client.responses.get(int(additional_status_code))
194                     response.setdefault(
195                         "description", status_text or "Additional Response"
196                     )
197                     operation.setdefault("responses", {})[
198                         str(additional_status_code)
199                     ] = response
200             status_code = str(route.status_code)
201             response_schema = {"type": "string"}
202             if lenient_issubclass(route.response_class, JSONResponse):
203                 if route.response_field:
204                     response_schema, _, _ = field_schema(
205                         route.response_field,
206                         model_name_map=model_name_map,
207                         ref_prefix=REF_PREFIX,
208                     )
209                 else:
210                     response_schema = {}
211             operation.setdefault("responses", {}).setdefault(status_code, {})[
212                 "description"
213             ] = route.response_description
214             operation.setdefault("responses", {}).setdefault(
215                 status_code, {}
216             ).setdefault("content", {}).setdefault(route.response_class.media_type, {})[
217                 "schema"
218             ] = response_schema
219             if all_route_params or route.body_field:
220                 operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {
221                     "description": "Validation Error",
222                     "content": {
223                         "application/json": {
224                             "schema": {"$ref": REF_PREFIX + "HTTPValidationError"}
225                         }
226                     },
227                 }
228             path[method.lower()] = operation
229     return path, security_schemes, definitions
230 
231 
232 def get_openapi(
233     *,
234     title: str,
235     version: str,
236     openapi_version: str = "3.0.2",
237     description: str = None,
238     routes: Sequence[BaseRoute],
239     openapi_prefix: str = ""
240 ) -> Dict:
241     info = {"title": title, "version": version}
242     if description:
243         info["description"] = description
244     output = {"openapi": openapi_version, "info": info}
245     components: Dict[str, Dict] = {}
246     paths: Dict[str, Dict] = {}
247     flat_models = get_flat_models_from_routes(routes)
248     model_name_map = get_model_name_map(flat_models)
249     definitions = get_model_definitions(
250         flat_models=flat_models, model_name_map=model_name_map
251     )
252     for route in routes:
253         if isinstance(route, routing.APIRoute):
254             result = get_openapi_path(route=route, model_name_map=model_name_map)
255             if result:
256                 path, security_schemes, path_definitions = result
257                 if path:
258                     paths.setdefault(openapi_prefix + route.path_format, {}).update(
259                         path
260                     )
261                 if security_schemes:
262                     components.setdefault("securitySchemes", {}).update(
263                         security_schemes
264                     )
265                 if path_definitions:
266                     definitions.update(path_definitions)
267     if definitions:
268         components.setdefault("schemas", {}).update(definitions)
269     if components:
270         output["components"] = components
271     output["paths"] = paths
272     return jsonable_encoder(OpenAPI(**output), by_alias=True, include_none=False)
273
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/fastapi/openapi/utils.py b/fastapi/openapi/utils.py
--- a/fastapi/openapi/utils.py
+++ b/fastapi/openapi/utils.py
@@ -71,15 +71,11 @@
 
 def get_openapi_operation_parameters(
     all_route_params: Sequence[Field]
-) -> Tuple[Dict[str, Dict], List[Dict[str, Any]]]:
-    definitions: Dict[str, Dict] = {}
+) -> List[Dict[str, Any]]:
     parameters = []
     for param in all_route_params:
         schema = param.schema
         schema = cast(Param, schema)
-        if "ValidationError" not in definitions:
-            definitions["ValidationError"] = validation_error_definition
-            definitions["HTTPValidationError"] = validation_error_response_definition
         parameter = {
             "name": param.alias,
             "in": schema.in_.value,
@@ -91,7 +87,7 @@
         if schema.deprecated:
             parameter["deprecated"] = schema.deprecated
         parameters.append(parameter)
-    return definitions, parameters
+    return parameters
 
 
 def get_openapi_operation_request_body(
@@ -159,10 +155,7 @@
             if security_definitions:
                 security_schemes.update(security_definitions)
             all_route_params = get_openapi_params(route.dependant)
-            validation_definitions, operation_parameters = get_openapi_operation_parameters(
-                all_route_params=all_route_params
-            )
-            definitions.update(validation_definitions)
+            operation_parameters = get_openapi_operation_parameters(all_route_params)
             parameters.extend(operation_parameters)
             if parameters:
                 operation["parameters"] = parameters
@@ -172,11 +165,6 @@
                 )
                 if request_body_oai:
                     operation["requestBody"] = request_body_oai
-                    if "ValidationError" not in definitions:
-                        definitions["ValidationError"] = validation_error_definition
-                        definitions[
-                            "HTTPValidationError"
-                        ] = validation_error_response_definition
             if route.responses:
                 for (additional_status_code, response) in route.responses.items():
                     assert isinstance(
@@ -188,7 +176,7 @@
                             field, model_name_map=model_name_map, ref_prefix=REF_PREFIX
                         )
                         response.setdefault("content", {}).setdefault(
-                            "application/json", {}
+                            route.response_class.media_type, {}
                         )["schema"] = response_schema
                     status_text = http.client.responses.get(int(additional_status_code))
                     response.setdefault(
@@ -216,8 +204,15 @@
             ).setdefault("content", {}).setdefault(route.response_class.media_type, {})[
                 "schema"
             ] = response_schema
-            if all_route_params or route.body_field:
-                operation["responses"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {
+
+            http422 = str(HTTP_422_UNPROCESSABLE_ENTITY)
+            if (all_route_params or route.body_field) and not any(
+                [
+                    status in operation["responses"]
+                    for status in [http422, "4xx", "default"]
+                ]
+            ):
+                operation["responses"][http422] = {
                     "description": "Validation Error",
                     "content": {
                         "application/json": {
@@ -225,6 +220,13 @@
                         }
                     },
                 }
+                if "ValidationError" not in definitions:
+                    definitions.update(
+                        {
+                            "ValidationError": validation_error_definition,
+                            "HTTPValidationError": validation_error_response_definition,
+                        }
+                    )
             path[method.lower()] = operation
     return path, security_schemes, definitions

{"golden_diff": "diff --git a/fastapi/openapi/utils.py b/fastapi/openapi/utils.py\n--- a/fastapi/openapi/utils.py\n+++ b/fastapi/openapi/utils.py\n@@ -71,15 +71,11 @@\n \n def get_openapi_operation_parameters(\n     all_route_params: Sequence[Field]\n-) -> Tuple[Dict[str, Dict], List[Dict[str, Any]]]:\n-    definitions: Dict[str, Dict] = {}\n+) -> List[Dict[str, Any]]:\n     parameters = []\n     for param in all_route_params:\n         schema = param.schema\n         schema = cast(Param, schema)\n-        if \"ValidationError\" not in definitions:\n-            definitions[\"ValidationError\"] = validation_error_definition\n-            definitions[\"HTTPValidationError\"] = validation_error_response_definition\n         parameter = {\n             \"name\": param.alias,\n             \"in\": schema.in_.value,\n@@ -91,7 +87,7 @@\n         if schema.deprecated:\n             parameter[\"deprecated\"] = schema.deprecated\n         parameters.append(parameter)\n-    return definitions, parameters\n+    return parameters\n \n \n def get_openapi_operation_request_body(\n@@ -159,10 +155,7 @@\n             if security_definitions:\n                 security_schemes.update(security_definitions)\n             all_route_params = get_openapi_params(route.dependant)\n-            validation_definitions, operation_parameters = get_openapi_operation_parameters(\n-                all_route_params=all_route_params\n-            )\n-            definitions.update(validation_definitions)\n+            operation_parameters = get_openapi_operation_parameters(all_route_params)\n             parameters.extend(operation_parameters)\n             if parameters:\n                 operation[\"parameters\"] = parameters\n@@ -172,11 +165,6 @@\n                 )\n                 if request_body_oai:\n                     operation[\"requestBody\"] = request_body_oai\n-                    if \"ValidationError\" not in definitions:\n-                        definitions[\"ValidationError\"] = validation_error_definition\n-                        definitions[\n-                            \"HTTPValidationError\"\n-                        ] = validation_error_response_definition\n             if route.responses:\n                 for (additional_status_code, response) in route.responses.items():\n                     assert isinstance(\n@@ -188,7 +176,7 @@\n                             field, model_name_map=model_name_map, ref_prefix=REF_PREFIX\n                         )\n                         response.setdefault(\"content\", {}).setdefault(\n-                            \"application/json\", {}\n+                            route.response_class.media_type, {}\n                         )[\"schema\"] = response_schema\n                     status_text = http.client.responses.get(int(additional_status_code))\n                     response.setdefault(\n@@ -216,8 +204,15 @@\n             ).setdefault(\"content\", {}).setdefault(route.response_class.media_type, {})[\n                 \"schema\"\n             ] = response_schema\n-            if all_route_params or route.body_field:\n-                operation[\"responses\"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {\n+\n+            http422 = str(HTTP_422_UNPROCESSABLE_ENTITY)\n+            if (all_route_params or route.body_field) and not any(\n+                [\n+                    status in operation[\"responses\"]\n+                    for status in [http422, \"4xx\", \"default\"]\n+                ]\n+            ):\n+                operation[\"responses\"][http422] = {\n                     \"description\": \"Validation Error\",\n                     \"content\": {\n                         \"application/json\": {\n@@ -225,6 +220,13 @@\n                         }\n                     },\n                 }\n+                if \"ValidationError\" not in definitions:\n+                    definitions.update(\n+                        {\n+                            \"ValidationError\": validation_error_definition,\n+                            \"HTTPValidationError\": validation_error_response_definition,\n+                        }\n+                    )\n             path[method.lower()] = operation\n     return path, security_schemes, definitions\n", "issue": "OpenAPI: HTTP_422 response does not use custom media_type\n**Describe the bug**\r\nFastAPI automatically adds an HTTP_422 response to all paths in the OpenAPI specification that have parameters or request body. This response does not use the media_type of response_class if any custom defined. Furthermore, it overwrites any error object format with the default one.\r\n\r\n**To Reproduce**\r\nCreate a path with parameters and add custom response_class to decorator. Add custom exception handlers that reformat the default error responses as per your liking. Then observe generated openapi.json\r\n\r\n```python\r\nfrom fastapi import FastAPI, HTTPException\r\nfrom fastapi.exceptions import RequestValidationError\r\nfrom starlette import status\r\nfrom starlette.responses import JSONResponse\r\nfrom . import schemas\r\n\r\napp = FastAPI()\r\n\r\nclass JsonApiResponse(JSONResponse):\r\n    media_type = 'application/vnd+json.api'\r\n\r\n@app.exception_handler(HTTPException)\r\nasync def http_exception_handler(request, exc: HTTPException) -> JsonApiResponse:\r\n    headers = getattr(exc, \"headers\", None)\r\n    content = schemas.ErrorResponse(errors=[dict(title=\"Bad request\", detail=exc.detail, status=exc.status_code)]).dict()\r\n    status_code = exc.status_code\r\n    if headers:\r\n        return JsonApiResponse(content=content, status_code=status_code, headers=headers)\r\n    else:\r\n        return JsonApiResponse(content=content, status_code=status_code)\r\n\r\n@app.exception_handler(RequestValidationError)\r\nasync def request_validation_exception_handler(request, exc: RequestValidationError) -> JsonApiResponse:\r\n    http422 = status.HTTP_422_UNPROCESSABLE_ENTITY\r\n    return JsonApiResponse(\r\n        content=schemas.ErrorResponse(errors=[\r\n            dict(title=err['type'], detail=err['msg'], source='/'.join(err['loc']), status=http422)\r\n            for err in exc.errors()\r\n        ]).dict(),\r\n        status_code=http422,\r\n    )\r\n\r\n@app.post('/customers',\r\n         status_code=status.HTTP_201_CREATED,\r\n         response_model=schemas.CustomerDetailsResponse,\r\n         response_class=JsonApiResponse,\r\n         )\r\ndef customer_create(data: schemas.Customer = Body(..., media_type='application/vnd+json.api', embed=True)):\r\n    created_customer = {**data.dict(), **{'id': '1'}}\r\n    return {'data': created_customer}\r\n``` \r\n\r\nThe openapi.json will include the unwanted 422 response with the FastAPI default error object definitions:\r\n\r\n```yaml\r\n  # ...\r\n  '422':\r\n    description: Validation Error\r\n    content:\r\n      application/json:\r\n        schema:\r\n          \"$ref\": \"#/components/schemas/HTTPValidationError\"\r\n```\r\n\r\n**Expected behavior**\r\nAt least, the media_type of the response_class should be respected. But the best would be if the 422 would not be added to the specification unless requested via the path decorator. Or if the 422 definitions of mine were respected.\r\n\r\n```python\r\n@app.post('/customers',\r\n         status_code=status.HTTP_201_CREATED,\r\n         response_model=schemas.CustomerDetailsResponse,\r\n         response_class=JsonApiResponse,\r\n         responses={\r\n             422: {\r\n                 'model': schemas.ErrorResponse\r\n             },\r\n         })\r\ndata: schemas.Customer = Body(..., media_type='application/vnd+json.api', embed=True)):\r\n    pass\r\n```\r\n\r\n**Environment:**\r\n - OS: masOS 10.14.6\r\n - Python: 3.6.5\r\n - FastAPI: 0.35.0\n", "before_files": [{"content": "import http.client\nfrom typing import Any, Dict, List, Optional, Sequence, Tuple, Type, cast\n\nfrom fastapi import routing\nfrom fastapi.dependencies.models import Dependant\nfrom fastapi.dependencies.utils import get_flat_dependant\nfrom fastapi.encoders import jsonable_encoder\nfrom fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX\nfrom fastapi.openapi.models import OpenAPI\nfrom fastapi.params import Body, Param\nfrom fastapi.utils import (\n    generate_operation_id_for_path,\n    get_flat_models_from_routes,\n    get_model_definitions,\n)\nfrom pydantic.fields import Field\nfrom pydantic.schema import field_schema, get_model_name_map\nfrom pydantic.utils import lenient_issubclass\nfrom starlette.responses import JSONResponse\nfrom starlette.routing import BaseRoute\nfrom starlette.status import HTTP_422_UNPROCESSABLE_ENTITY\n\nvalidation_error_definition = {\n    \"title\": \"ValidationError\",\n    \"type\": \"object\",\n    \"properties\": {\n        \"loc\": {\"title\": \"Location\", \"type\": \"array\", \"items\": {\"type\": \"string\"}},\n        \"msg\": {\"title\": \"Message\", \"type\": \"string\"},\n        \"type\": {\"title\": \"Error Type\", \"type\": \"string\"},\n    },\n    \"required\": [\"loc\", \"msg\", \"type\"],\n}\n\nvalidation_error_response_definition = {\n    \"title\": \"HTTPValidationError\",\n    \"type\": \"object\",\n    \"properties\": {\n        \"detail\": {\n            \"title\": \"Detail\",\n            \"type\": \"array\",\n            \"items\": {\"$ref\": REF_PREFIX + \"ValidationError\"},\n        }\n    },\n}\n\n\ndef get_openapi_params(dependant: Dependant) -> List[Field]:\n    flat_dependant = get_flat_dependant(dependant)\n    return (\n        flat_dependant.path_params\n        + flat_dependant.query_params\n        + flat_dependant.header_params\n        + flat_dependant.cookie_params\n    )\n\n\ndef get_openapi_security_definitions(flat_dependant: Dependant) -> Tuple[Dict, List]:\n    security_definitions = {}\n    operation_security = []\n    for security_requirement in flat_dependant.security_requirements:\n        security_definition = jsonable_encoder(\n            security_requirement.security_scheme.model,\n            by_alias=True,\n            include_none=False,\n        )\n        security_name = security_requirement.security_scheme.scheme_name\n        security_definitions[security_name] = security_definition\n        operation_security.append({security_name: security_requirement.scopes})\n    return security_definitions, operation_security\n\n\ndef get_openapi_operation_parameters(\n    all_route_params: Sequence[Field]\n) -> Tuple[Dict[str, Dict], List[Dict[str, Any]]]:\n    definitions: Dict[str, Dict] = {}\n    parameters = []\n    for param in all_route_params:\n        schema = param.schema\n        schema = cast(Param, schema)\n        if \"ValidationError\" not in definitions:\n            definitions[\"ValidationError\"] = validation_error_definition\n            definitions[\"HTTPValidationError\"] = validation_error_response_definition\n        parameter = {\n            \"name\": param.alias,\n            \"in\": schema.in_.value,\n            \"required\": param.required,\n            \"schema\": field_schema(param, model_name_map={})[0],\n        }\n        if schema.description:\n            parameter[\"description\"] = schema.description\n        if schema.deprecated:\n            parameter[\"deprecated\"] = schema.deprecated\n        parameters.append(parameter)\n    return definitions, parameters\n\n\ndef get_openapi_operation_request_body(\n    *, body_field: Optional[Field], model_name_map: Dict[Type, str]\n) -> Optional[Dict]:\n    if not body_field:\n        return None\n    assert isinstance(body_field, Field)\n    body_schema, _, _ = field_schema(\n        body_field, model_name_map=model_name_map, ref_prefix=REF_PREFIX\n    )\n    body_field.schema = cast(Body, body_field.schema)\n    request_media_type = body_field.schema.media_type\n    required = body_field.required\n    request_body_oai: Dict[str, Any] = {}\n    if required:\n        request_body_oai[\"required\"] = required\n    request_body_oai[\"content\"] = {request_media_type: {\"schema\": body_schema}}\n    return request_body_oai\n\n\ndef generate_operation_id(*, route: routing.APIRoute, method: str) -> str:\n    if route.operation_id:\n        return route.operation_id\n    path: str = route.path_format\n    return generate_operation_id_for_path(name=route.name, path=path, method=method)\n\n\ndef generate_operation_summary(*, route: routing.APIRoute, method: str) -> str:\n    if route.summary:\n        return route.summary\n    return route.name.replace(\"_\", \" \").title()\n\n\ndef get_openapi_operation_metadata(*, route: routing.APIRoute, method: str) -> Dict:\n    operation: Dict[str, Any] = {}\n    if route.tags:\n        operation[\"tags\"] = route.tags\n    operation[\"summary\"] = generate_operation_summary(route=route, method=method)\n    if route.description:\n        operation[\"description\"] = route.description\n    operation[\"operationId\"] = generate_operation_id(route=route, method=method)\n    if route.deprecated:\n        operation[\"deprecated\"] = route.deprecated\n    return operation\n\n\ndef get_openapi_path(\n    *, route: routing.APIRoute, model_name_map: Dict[Type, str]\n) -> Tuple[Dict, Dict, Dict]:\n    path = {}\n    security_schemes: Dict[str, Any] = {}\n    definitions: Dict[str, Any] = {}\n    assert route.methods is not None, \"Methods must be a list\"\n    if route.include_in_schema:\n        for method in route.methods:\n            operation = get_openapi_operation_metadata(route=route, method=method)\n            parameters: List[Dict] = []\n            flat_dependant = get_flat_dependant(route.dependant)\n            security_definitions, operation_security = get_openapi_security_definitions(\n                flat_dependant=flat_dependant\n            )\n            if operation_security:\n                operation.setdefault(\"security\", []).extend(operation_security)\n            if security_definitions:\n                security_schemes.update(security_definitions)\n            all_route_params = get_openapi_params(route.dependant)\n            validation_definitions, operation_parameters = get_openapi_operation_parameters(\n                all_route_params=all_route_params\n            )\n            definitions.update(validation_definitions)\n            parameters.extend(operation_parameters)\n            if parameters:\n                operation[\"parameters\"] = parameters\n            if method in METHODS_WITH_BODY:\n                request_body_oai = get_openapi_operation_request_body(\n                    body_field=route.body_field, model_name_map=model_name_map\n                )\n                if request_body_oai:\n                    operation[\"requestBody\"] = request_body_oai\n                    if \"ValidationError\" not in definitions:\n                        definitions[\"ValidationError\"] = validation_error_definition\n                        definitions[\n                            \"HTTPValidationError\"\n                        ] = validation_error_response_definition\n            if route.responses:\n                for (additional_status_code, response) in route.responses.items():\n                    assert isinstance(\n                        response, dict\n                    ), \"An additional response must be a dict\"\n                    field = route.response_fields.get(additional_status_code)\n                    if field:\n                        response_schema, _, _ = field_schema(\n                            field, model_name_map=model_name_map, ref_prefix=REF_PREFIX\n                        )\n                        response.setdefault(\"content\", {}).setdefault(\n                            \"application/json\", {}\n                        )[\"schema\"] = response_schema\n                    status_text = http.client.responses.get(int(additional_status_code))\n                    response.setdefault(\n                        \"description\", status_text or \"Additional Response\"\n                    )\n                    operation.setdefault(\"responses\", {})[\n                        str(additional_status_code)\n                    ] = response\n            status_code = str(route.status_code)\n            response_schema = {\"type\": \"string\"}\n            if lenient_issubclass(route.response_class, JSONResponse):\n                if route.response_field:\n                    response_schema, _, _ = field_schema(\n                        route.response_field,\n                        model_name_map=model_name_map,\n                        ref_prefix=REF_PREFIX,\n                    )\n                else:\n                    response_schema = {}\n            operation.setdefault(\"responses\", {}).setdefault(status_code, {})[\n                \"description\"\n            ] = route.response_description\n            operation.setdefault(\"responses\", {}).setdefault(\n                status_code, {}\n            ).setdefault(\"content\", {}).setdefault(route.response_class.media_type, {})[\n                \"schema\"\n            ] = response_schema\n            if all_route_params or route.body_field:\n                operation[\"responses\"][str(HTTP_422_UNPROCESSABLE_ENTITY)] = {\n                    \"description\": \"Validation Error\",\n                    \"content\": {\n                        \"application/json\": {\n                            \"schema\": {\"$ref\": REF_PREFIX + \"HTTPValidationError\"}\n                        }\n                    },\n                }\n            path[method.lower()] = operation\n    return path, security_schemes, definitions\n\n\ndef get_openapi(\n    *,\n    title: str,\n    version: str,\n    openapi_version: str = \"3.0.2\",\n    description: str = None,\n    routes: Sequence[BaseRoute],\n    openapi_prefix: str = \"\"\n) -> Dict:\n    info = {\"title\": title, \"version\": version}\n    if description:\n        info[\"description\"] = description\n    output = {\"openapi\": openapi_version, \"info\": info}\n    components: Dict[str, Dict] = {}\n    paths: Dict[str, Dict] = {}\n    flat_models = get_flat_models_from_routes(routes)\n    model_name_map = get_model_name_map(flat_models)\n    definitions = get_model_definitions(\n        flat_models=flat_models, model_name_map=model_name_map\n    )\n    for route in routes:\n        if isinstance(route, routing.APIRoute):\n            result = get_openapi_path(route=route, model_name_map=model_name_map)\n            if result:\n                path, security_schemes, path_definitions = result\n                if path:\n                    paths.setdefault(openapi_prefix + route.path_format, {}).update(\n                        path\n                    )\n                if security_schemes:\n                    components.setdefault(\"securitySchemes\", {}).update(\n                        security_schemes\n                    )\n                if path_definitions:\n                    definitions.update(path_definitions)\n    if definitions:\n        components.setdefault(\"schemas\", {}).update(definitions)\n    if components:\n        output[\"components\"] = components\n    output[\"paths\"] = paths\n    return jsonable_encoder(OpenAPI(**output), by_alias=True, include_none=False)\n", "path": "fastapi/openapi/utils.py"}], "after_files": [{"content": "import http.client\nfrom typing import Any, Dict, List, Optional, Sequence, Tuple, Type, cast\n\nfrom fastapi import routing\nfrom fastapi.dependencies.models import Dependant\nfrom fastapi.dependencies.utils import get_flat_dependant\nfrom fastapi.encoders import jsonable_encoder\nfrom fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX\nfrom fastapi.openapi.models import OpenAPI\nfrom fastapi.params import Body, Param\nfrom fastapi.utils import (\n    generate_operation_id_for_path,\n    get_flat_models_from_routes,\n    get_model_definitions,\n)\nfrom pydantic.fields import Field\nfrom pydantic.schema import field_schema, get_model_name_map\nfrom pydantic.utils import lenient_issubclass\nfrom starlette.responses import JSONResponse\nfrom starlette.routing import BaseRoute\nfrom starlette.status import HTTP_422_UNPROCESSABLE_ENTITY\n\nvalidation_error_definition = {\n    \"title\": \"ValidationError\",\n    \"type\": \"object\",\n    \"properties\": {\n        \"loc\": {\"title\": \"Location\", \"type\": \"array\", \"items\": {\"type\": \"string\"}},\n        \"msg\": {\"title\": \"Message\", \"type\": \"string\"},\n        \"type\": {\"title\": \"Error Type\", \"type\": \"string\"},\n    },\n    \"required\": [\"loc\", \"msg\", \"type\"],\n}\n\nvalidation_error_response_definition = {\n    \"title\": \"HTTPValidationError\",\n    \"type\": \"object\",\n    \"properties\": {\n        \"detail\": {\n            \"title\": \"Detail\",\n            \"type\": \"array\",\n            \"items\": {\"$ref\": REF_PREFIX + \"ValidationError\"},\n        }\n    },\n}\n\n\ndef get_openapi_params(dependant: Dependant) -> List[Field]:\n    flat_dependant = get_flat_dependant(dependant)\n    return (\n        flat_dependant.path_params\n        + flat_dependant.query_params\n        + flat_dependant.header_params\n        + flat_dependant.cookie_params\n    )\n\n\ndef get_openapi_security_definitions(flat_dependant: Dependant) -> Tuple[Dict, List]:\n    security_definitions = {}\n    operation_security = []\n    for security_requirement in flat_dependant.security_requirements:\n        security_definition = jsonable_encoder(\n            security_requirement.security_scheme.model,\n            by_alias=True,\n            include_none=False,\n        )\n        security_name = security_requirement.security_scheme.scheme_name\n        security_definitions[security_name] = security_definition\n        operation_security.append({security_name: security_requirement.scopes})\n    return security_definitions, operation_security\n\n\ndef get_openapi_operation_parameters(\n    all_route_params: Sequence[Field]\n) -> List[Dict[str, Any]]:\n    parameters = []\n    for param in all_route_params:\n        schema = param.schema\n        schema = cast(Param, schema)\n        parameter = {\n            \"name\": param.alias,\n            \"in\": schema.in_.value,\n            \"required\": param.required,\n            \"schema\": field_schema(param, model_name_map={})[0],\n        }\n        if schema.description:\n            parameter[\"description\"] = schema.description\n        if schema.deprecated:\n            parameter[\"deprecated\"] = schema.deprecated\n        parameters.append(parameter)\n    return parameters\n\n\ndef get_openapi_operation_request_body(\n    *, body_field: Optional[Field], model_name_map: Dict[Type, str]\n) -> Optional[Dict]:\n    if not body_field:\n        return None\n    assert isinstance(body_field, Field)\n    body_schema, _, _ = field_schema(\n        body_field, model_name_map=model_name_map, ref_prefix=REF_PREFIX\n    )\n    body_field.schema = cast(Body, body_field.schema)\n    request_media_type = body_field.schema.media_type\n    required = body_field.required\n    request_body_oai: Dict[str, Any] = {}\n    if required:\n        request_body_oai[\"required\"] = required\n    request_body_oai[\"content\"] = {request_media_type: {\"schema\": body_schema}}\n    return request_body_oai\n\n\ndef generate_operation_id(*, route: routing.APIRoute, method: str) -> str:\n    if route.operation_id:\n        return route.operation_id\n    path: str = route.path_format\n    return generate_operation_id_for_path(name=route.name, path=path, method=method)\n\n\ndef generate_operation_summary(*, route: routing.APIRoute, method: str) -> str:\n    if route.summary:\n        return route.summary\n    return route.name.replace(\"_\", \" \").title()\n\n\ndef get_openapi_operation_metadata(*, route: routing.APIRoute, method: str) -> Dict:\n    operation: Dict[str, Any] = {}\n    if route.tags:\n        operation[\"tags\"] = route.tags\n    operation[\"summary\"] = generate_operation_summary(route=route, method=method)\n    if route.description:\n        operation[\"description\"] = route.description\n    operation[\"operationId\"] = generate_operation_id(route=route, method=method)\n    if route.deprecated:\n        operation[\"deprecated\"] = route.deprecated\n    return operation\n\n\ndef get_openapi_path(\n    *, route: routing.APIRoute, model_name_map: Dict[Type, str]\n) -> Tuple[Dict, Dict, Dict]:\n    path = {}\n    security_schemes: Dict[str, Any] = {}\n    definitions: Dict[str, Any] = {}\n    assert route.methods is not None, \"Methods must be a list\"\n    if route.include_in_schema:\n        for method in route.methods:\n            operation = get_openapi_operation_metadata(route=route, method=method)\n            parameters: List[Dict] = []\n            flat_dependant = get_flat_dependant(route.dependant)\n            security_definitions, operation_security = get_openapi_security_definitions(\n                flat_dependant=flat_dependant\n            )\n            if operation_security:\n                operation.setdefault(\"security\", []).extend(operation_security)\n            if security_definitions:\n                security_schemes.update(security_definitions)\n            all_route_params = get_openapi_params(route.dependant)\n            operation_parameters = get_openapi_operation_parameters(all_route_params)\n            parameters.extend(operation_parameters)\n            if parameters:\n                operation[\"parameters\"] = parameters\n            if method in METHODS_WITH_BODY:\n                request_body_oai = get_openapi_operation_request_body(\n                    body_field=route.body_field, model_name_map=model_name_map\n                )\n                if request_body_oai:\n                    operation[\"requestBody\"] = request_body_oai\n            if route.responses:\n                for (additional_status_code, response) in route.responses.items():\n                    assert isinstance(\n                        response, dict\n                    ), \"An additional response must be a dict\"\n                    field = route.response_fields.get(additional_status_code)\n                    if field:\n                        response_schema, _, _ = field_schema(\n                            field, model_name_map=model_name_map, ref_prefix=REF_PREFIX\n                        )\n                        response.setdefault(\"content\", {}).setdefault(\n                            route.response_class.media_type, {}\n                        )[\"schema\"] = response_schema\n                    status_text = http.client.responses.get(int(additional_status_code))\n                    response.setdefault(\n                        \"description\", status_text or \"Additional Response\"\n                    )\n                    operation.setdefault(\"responses\", {})[\n                        str(additional_status_code)\n                    ] = response\n            status_code = str(route.status_code)\n            response_schema = {\"type\": \"string\"}\n            if lenient_issubclass(route.response_class, JSONResponse):\n                if route.response_field:\n                    response_schema, _, _ = field_schema(\n                        route.response_field,\n                        model_name_map=model_name_map,\n                        ref_prefix=REF_PREFIX,\n                    )\n                else:\n                    response_schema = {}\n            operation.setdefault(\"responses\", {}).setdefault(status_code, {})[\n                \"description\"\n            ] = route.response_description\n            operation.setdefault(\"responses\", {}).setdefault(\n                status_code, {}\n            ).setdefault(\"content\", {}).setdefault(route.response_class.media_type, {})[\n                \"schema\"\n            ] = response_schema\n\n            http422 = str(HTTP_422_UNPROCESSABLE_ENTITY)\n            if (all_route_params or route.body_field) and not any(\n                [\n                    status in operation[\"responses\"]\n                    for status in [http422, \"4xx\", \"default\"]\n                ]\n            ):\n                operation[\"responses\"][http422] = {\n                    \"description\": \"Validation Error\",\n                    \"content\": {\n                        \"application/json\": {\n                            \"schema\": {\"$ref\": REF_PREFIX + \"HTTPValidationError\"}\n                        }\n                    },\n                }\n                if \"ValidationError\" not in definitions:\n                    definitions.update(\n                        {\n                            \"ValidationError\": validation_error_definition,\n                            \"HTTPValidationError\": validation_error_response_definition,\n                        }\n                    )\n            path[method.lower()] = operation\n    return path, security_schemes, definitions\n\n\ndef get_openapi(\n    *,\n    title: str,\n    version: str,\n    openapi_version: str = \"3.0.2\",\n    description: str = None,\n    routes: Sequence[BaseRoute],\n    openapi_prefix: str = \"\"\n) -> Dict:\n    info = {\"title\": title, \"version\": version}\n    if description:\n        info[\"description\"] = description\n    output = {\"openapi\": openapi_version, \"info\": info}\n    components: Dict[str, Dict] = {}\n    paths: Dict[str, Dict] = {}\n    flat_models = get_flat_models_from_routes(routes)\n    model_name_map = get_model_name_map(flat_models)\n    definitions = get_model_definitions(\n        flat_models=flat_models, model_name_map=model_name_map\n    )\n    for route in routes:\n        if isinstance(route, routing.APIRoute):\n            result = get_openapi_path(route=route, model_name_map=model_name_map)\n            if result:\n                path, security_schemes, path_definitions = result\n                if path:\n                    paths.setdefault(openapi_prefix + route.path_format, {}).update(\n                        path\n                    )\n                if security_schemes:\n                    components.setdefault(\"securitySchemes\", {}).update(\n                        security_schemes\n                    )\n                if path_definitions:\n                    definitions.update(path_definitions)\n    if definitions:\n        components.setdefault(\"schemas\", {}).update(definitions)\n    if components:\n        output[\"components\"] = components\n    output[\"paths\"] = paths\n    return jsonable_encoder(OpenAPI(**output), by_alias=True, include_none=False)\n", "path": "fastapi/openapi/utils.py"}]}