Загрузить файлы в «venv/Lib/site-packages/fastapi/openapi»
This commit is contained in:
BIN
venv/Lib/site-packages/fastapi/openapi/__init__.py
Normal file
BIN
venv/Lib/site-packages/fastapi/openapi/__init__.py
Normal file
Binary file not shown.
3
venv/Lib/site-packages/fastapi/openapi/constants.py
Normal file
3
venv/Lib/site-packages/fastapi/openapi/constants.py
Normal file
@@ -0,0 +1,3 @@
|
|||||||
|
METHODS_WITH_BODY = {"GET", "HEAD", "POST", "PUT", "DELETE", "PATCH"}
|
||||||
|
REF_PREFIX = "#/components/schemas/"
|
||||||
|
REF_TEMPLATE = "#/components/schemas/{model}"
|
||||||
389
venv/Lib/site-packages/fastapi/openapi/docs.py
Normal file
389
venv/Lib/site-packages/fastapi/openapi/docs.py
Normal file
@@ -0,0 +1,389 @@
|
|||||||
|
import json
|
||||||
|
from typing import Annotated, Any
|
||||||
|
|
||||||
|
from annotated_doc import Doc
|
||||||
|
from fastapi.encoders import jsonable_encoder
|
||||||
|
from starlette.responses import HTMLResponse
|
||||||
|
|
||||||
|
|
||||||
|
def _html_safe_json(value: Any) -> str:
|
||||||
|
"""Serialize a value to JSON with HTML special characters escaped.
|
||||||
|
|
||||||
|
This prevents injection when the JSON is embedded inside a <script> tag.
|
||||||
|
"""
|
||||||
|
return (
|
||||||
|
json.dumps(value)
|
||||||
|
.replace("<", "\\u003c")
|
||||||
|
.replace(">", "\\u003e")
|
||||||
|
.replace("&", "\\u0026")
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
swagger_ui_default_parameters: Annotated[
|
||||||
|
dict[str, Any],
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
Default configurations for Swagger UI.
|
||||||
|
|
||||||
|
You can use it as a template to add any other configurations needed.
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = {
|
||||||
|
"dom_id": "#swagger-ui",
|
||||||
|
"layout": "BaseLayout",
|
||||||
|
"deepLinking": True,
|
||||||
|
"showExtensions": True,
|
||||||
|
"showCommonExtensions": True,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def get_swagger_ui_html(
|
||||||
|
*,
|
||||||
|
openapi_url: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The OpenAPI URL that Swagger UI should load and use.
|
||||||
|
|
||||||
|
This is normally done automatically by FastAPI using the default URL
|
||||||
|
`/openapi.json`.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Conditional OpenAPI](https://fastapi.tiangolo.com/how-to/conditional-openapi/#conditional-openapi-from-settings-and-env-vars)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
],
|
||||||
|
title: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The HTML `<title>` content, normally shown in the browser tab.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Custom Docs UI Static Assets](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
],
|
||||||
|
swagger_js_url: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The URL to use to load the Swagger UI JavaScript.
|
||||||
|
|
||||||
|
It is normally set to a CDN URL.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Custom Docs UI Static Assets](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui-bundle.js",
|
||||||
|
swagger_css_url: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The URL to use to load the Swagger UI CSS.
|
||||||
|
|
||||||
|
It is normally set to a CDN URL.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Custom Docs UI Static Assets](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist@5/swagger-ui.css",
|
||||||
|
swagger_favicon_url: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The URL of the favicon to use. It is normally shown in the browser tab.
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = "https://fastapi.tiangolo.com/img/favicon.png",
|
||||||
|
oauth2_redirect_url: Annotated[
|
||||||
|
str | None,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The OAuth2 redirect URL, it is normally automatically handled by FastAPI.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Custom Docs UI Static Assets](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = None,
|
||||||
|
init_oauth: Annotated[
|
||||||
|
dict[str, Any] | None,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
A dictionary with Swagger UI OAuth2 initialization configurations.
|
||||||
|
|
||||||
|
Read more about the available configuration options in the
|
||||||
|
[Swagger UI docs](https://swagger.io/docs/open-source-tools/swagger-ui/usage/oauth2/).
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = None,
|
||||||
|
swagger_ui_parameters: Annotated[
|
||||||
|
dict[str, Any] | None,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
Configuration parameters for Swagger UI.
|
||||||
|
|
||||||
|
It defaults to [swagger_ui_default_parameters][fastapi.openapi.docs.swagger_ui_default_parameters].
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs about how to Configure Swagger UI](https://fastapi.tiangolo.com/how-to/configure-swagger-ui/).
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = None,
|
||||||
|
) -> HTMLResponse:
|
||||||
|
"""
|
||||||
|
Generate and return the HTML that loads Swagger UI for the interactive
|
||||||
|
API docs (normally served at `/docs`).
|
||||||
|
|
||||||
|
You would only call this function yourself if you needed to override some parts,
|
||||||
|
for example the URLs to use to load Swagger UI's JavaScript and CSS.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Configure Swagger UI](https://fastapi.tiangolo.com/how-to/configure-swagger-ui/)
|
||||||
|
and the [FastAPI docs for Custom Docs UI Static Assets (Self-Hosting)](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/).
|
||||||
|
"""
|
||||||
|
current_swagger_ui_parameters = swagger_ui_default_parameters.copy()
|
||||||
|
if swagger_ui_parameters:
|
||||||
|
current_swagger_ui_parameters.update(swagger_ui_parameters)
|
||||||
|
|
||||||
|
html = f"""
|
||||||
|
<!DOCTYPE html>
|
||||||
|
<html>
|
||||||
|
<head>
|
||||||
|
<meta name="viewport" content="width=device-width, initial-scale=1.0">
|
||||||
|
<link type="text/css" rel="stylesheet" href="{swagger_css_url}">
|
||||||
|
<link rel="shortcut icon" href="{swagger_favicon_url}">
|
||||||
|
<title>{title}</title>
|
||||||
|
</head>
|
||||||
|
<body>
|
||||||
|
<div id="swagger-ui">
|
||||||
|
</div>
|
||||||
|
<script src="{swagger_js_url}"></script>
|
||||||
|
<!-- `SwaggerUIBundle` is now available on the page -->
|
||||||
|
<script>
|
||||||
|
const ui = SwaggerUIBundle({{
|
||||||
|
url: '{openapi_url}',
|
||||||
|
"""
|
||||||
|
|
||||||
|
for key, value in current_swagger_ui_parameters.items():
|
||||||
|
html += f"{_html_safe_json(key)}: {_html_safe_json(jsonable_encoder(value))},\n"
|
||||||
|
|
||||||
|
if oauth2_redirect_url:
|
||||||
|
html += f"oauth2RedirectUrl: window.location.origin + '{oauth2_redirect_url}',"
|
||||||
|
|
||||||
|
html += """
|
||||||
|
presets: [
|
||||||
|
SwaggerUIBundle.presets.apis,
|
||||||
|
SwaggerUIBundle.SwaggerUIStandalonePreset
|
||||||
|
],
|
||||||
|
})"""
|
||||||
|
|
||||||
|
if init_oauth:
|
||||||
|
html += f"""
|
||||||
|
ui.initOAuth({_html_safe_json(jsonable_encoder(init_oauth))})
|
||||||
|
"""
|
||||||
|
|
||||||
|
html += """
|
||||||
|
</script>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
|
"""
|
||||||
|
return HTMLResponse(html)
|
||||||
|
|
||||||
|
|
||||||
|
def get_redoc_html(
|
||||||
|
*,
|
||||||
|
openapi_url: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The OpenAPI URL that ReDoc should load and use.
|
||||||
|
|
||||||
|
This is normally done automatically by FastAPI using the default URL
|
||||||
|
`/openapi.json`.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Conditional OpenAPI](https://fastapi.tiangolo.com/how-to/conditional-openapi/#conditional-openapi-from-settings-and-env-vars)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
],
|
||||||
|
title: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The HTML `<title>` content, normally shown in the browser tab.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Custom Docs UI Static Assets](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
],
|
||||||
|
redoc_js_url: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The URL to use to load the ReDoc JavaScript.
|
||||||
|
|
||||||
|
It is normally set to a CDN URL.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Custom Docs UI Static Assets](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/)
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = "https://cdn.jsdelivr.net/npm/redoc@2/bundles/redoc.standalone.js",
|
||||||
|
redoc_favicon_url: Annotated[
|
||||||
|
str,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
The URL of the favicon to use. It is normally shown in the browser tab.
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = "https://fastapi.tiangolo.com/img/favicon.png",
|
||||||
|
with_google_fonts: Annotated[
|
||||||
|
bool,
|
||||||
|
Doc(
|
||||||
|
"""
|
||||||
|
Load and use Google Fonts.
|
||||||
|
"""
|
||||||
|
),
|
||||||
|
] = True,
|
||||||
|
) -> HTMLResponse:
|
||||||
|
"""
|
||||||
|
Generate and return the HTML response that loads ReDoc for the alternative
|
||||||
|
API docs (normally served at `/redoc`).
|
||||||
|
|
||||||
|
You would only call this function yourself if you needed to override some parts,
|
||||||
|
for example the URLs to use to load ReDoc's JavaScript and CSS.
|
||||||
|
|
||||||
|
Read more about it in the
|
||||||
|
[FastAPI docs for Custom Docs UI Static Assets (Self-Hosting)](https://fastapi.tiangolo.com/how-to/custom-docs-ui-assets/).
|
||||||
|
"""
|
||||||
|
html = f"""
|
||||||
|
<!DOCTYPE html>
|
||||||
|
<html>
|
||||||
|
<head>
|
||||||
|
<title>{title}</title>
|
||||||
|
<!-- needed for adaptive design -->
|
||||||
|
<meta charset="utf-8"/>
|
||||||
|
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||||
|
"""
|
||||||
|
if with_google_fonts:
|
||||||
|
html += """
|
||||||
|
<link href="https://fonts.googleapis.com/css?family=Montserrat:300,400,700|Roboto:300,400,700" rel="stylesheet">
|
||||||
|
"""
|
||||||
|
html += f"""
|
||||||
|
<link rel="shortcut icon" href="{redoc_favicon_url}">
|
||||||
|
<!--
|
||||||
|
ReDoc doesn't change outer page styles
|
||||||
|
-->
|
||||||
|
<style>
|
||||||
|
body {{
|
||||||
|
margin: 0;
|
||||||
|
padding: 0;
|
||||||
|
}}
|
||||||
|
</style>
|
||||||
|
</head>
|
||||||
|
<body>
|
||||||
|
<noscript>
|
||||||
|
ReDoc requires Javascript to function. Please enable it to browse the documentation.
|
||||||
|
</noscript>
|
||||||
|
<redoc spec-url="{openapi_url}"></redoc>
|
||||||
|
<script src="{redoc_js_url}"> </script>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
|
"""
|
||||||
|
return HTMLResponse(html)
|
||||||
|
|
||||||
|
|
||||||
|
def get_swagger_ui_oauth2_redirect_html() -> HTMLResponse:
|
||||||
|
"""
|
||||||
|
Generate the HTML response with the OAuth2 redirection for Swagger UI.
|
||||||
|
|
||||||
|
You normally don't need to use or change this.
|
||||||
|
"""
|
||||||
|
# copied from https://github.com/swagger-api/swagger-ui/blob/v4.14.0/dist/oauth2-redirect.html
|
||||||
|
html = """
|
||||||
|
<!doctype html>
|
||||||
|
<html lang="en-US">
|
||||||
|
<head>
|
||||||
|
<title>Swagger UI: OAuth2 Redirect</title>
|
||||||
|
</head>
|
||||||
|
<body>
|
||||||
|
<script>
|
||||||
|
'use strict';
|
||||||
|
function run () {
|
||||||
|
var oauth2 = window.opener.swaggerUIRedirectOauth2;
|
||||||
|
var sentState = oauth2.state;
|
||||||
|
var redirectUrl = oauth2.redirectUrl;
|
||||||
|
var isValid, qp, arr;
|
||||||
|
|
||||||
|
if (/code|token|error/.test(window.location.hash)) {
|
||||||
|
qp = window.location.hash.substring(1).replace('?', '&');
|
||||||
|
} else {
|
||||||
|
qp = location.search.substring(1);
|
||||||
|
}
|
||||||
|
|
||||||
|
arr = qp.split("&");
|
||||||
|
arr.forEach(function (v,i,_arr) { _arr[i] = '"' + v.replace('=', '":"') + '"';});
|
||||||
|
qp = qp ? JSON.parse('{' + arr.join() + '}',
|
||||||
|
function (key, value) {
|
||||||
|
return key === "" ? value : decodeURIComponent(value);
|
||||||
|
}
|
||||||
|
) : {};
|
||||||
|
|
||||||
|
isValid = qp.state === sentState;
|
||||||
|
|
||||||
|
if ((
|
||||||
|
oauth2.auth.schema.get("flow") === "accessCode" ||
|
||||||
|
oauth2.auth.schema.get("flow") === "authorizationCode" ||
|
||||||
|
oauth2.auth.schema.get("flow") === "authorization_code"
|
||||||
|
) && !oauth2.auth.code) {
|
||||||
|
if (!isValid) {
|
||||||
|
oauth2.errCb({
|
||||||
|
authId: oauth2.auth.name,
|
||||||
|
source: "auth",
|
||||||
|
level: "warning",
|
||||||
|
message: "Authorization may be unsafe, passed state was changed in server. The passed state wasn't returned from auth server."
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
if (qp.code) {
|
||||||
|
delete oauth2.state;
|
||||||
|
oauth2.auth.code = qp.code;
|
||||||
|
oauth2.callback({auth: oauth2.auth, redirectUrl: redirectUrl});
|
||||||
|
} else {
|
||||||
|
let oauthErrorMsg;
|
||||||
|
if (qp.error) {
|
||||||
|
oauthErrorMsg = "["+qp.error+"]: " +
|
||||||
|
(qp.error_description ? qp.error_description+ ". " : "no accessCode received from the server. ") +
|
||||||
|
(qp.error_uri ? "More info: "+qp.error_uri : "");
|
||||||
|
}
|
||||||
|
|
||||||
|
oauth2.errCb({
|
||||||
|
authId: oauth2.auth.name,
|
||||||
|
source: "auth",
|
||||||
|
level: "error",
|
||||||
|
message: oauthErrorMsg || "[Authorization failed]: no accessCode received from the server."
|
||||||
|
});
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
oauth2.callback({auth: oauth2.auth, token: qp, isValid: isValid, redirectUrl: redirectUrl});
|
||||||
|
}
|
||||||
|
window.close();
|
||||||
|
}
|
||||||
|
|
||||||
|
if (document.readyState !== 'loading') {
|
||||||
|
run();
|
||||||
|
} else {
|
||||||
|
document.addEventListener('DOMContentLoaded', function () {
|
||||||
|
run();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
</script>
|
||||||
|
</body>
|
||||||
|
</html>
|
||||||
|
"""
|
||||||
|
return HTMLResponse(content=html)
|
||||||
435
venv/Lib/site-packages/fastapi/openapi/models.py
Normal file
435
venv/Lib/site-packages/fastapi/openapi/models.py
Normal file
@@ -0,0 +1,435 @@
|
|||||||
|
from collections.abc import Callable, Iterable, Mapping
|
||||||
|
from enum import Enum
|
||||||
|
from typing import Annotated, Any, Literal, Optional, Union
|
||||||
|
|
||||||
|
from fastapi._compat import with_info_plain_validator_function
|
||||||
|
from fastapi.logger import logger
|
||||||
|
from pydantic import (
|
||||||
|
AnyUrl,
|
||||||
|
BaseModel,
|
||||||
|
Field,
|
||||||
|
GetJsonSchemaHandler,
|
||||||
|
)
|
||||||
|
from typing_extensions import TypedDict
|
||||||
|
from typing_extensions import deprecated as typing_deprecated
|
||||||
|
|
||||||
|
try:
|
||||||
|
import email_validator
|
||||||
|
|
||||||
|
assert email_validator # make autoflake ignore the unused import
|
||||||
|
from pydantic import EmailStr
|
||||||
|
except ImportError: # pragma: no cover
|
||||||
|
|
||||||
|
class EmailStr(str): # type: ignore[no-redef]
|
||||||
|
@classmethod
|
||||||
|
def __get_validators__(cls) -> Iterable[Callable[..., Any]]:
|
||||||
|
yield cls.validate
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def validate(cls, v: Any) -> str:
|
||||||
|
logger.warning(
|
||||||
|
"email-validator not installed, email fields will be treated as str.\n"
|
||||||
|
"To install, run: pip install email-validator"
|
||||||
|
)
|
||||||
|
return str(v)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def _validate(cls, __input_value: Any, _: Any) -> str:
|
||||||
|
logger.warning(
|
||||||
|
"email-validator not installed, email fields will be treated as str.\n"
|
||||||
|
"To install, run: pip install email-validator"
|
||||||
|
)
|
||||||
|
return str(__input_value)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def __get_pydantic_json_schema__(
|
||||||
|
cls, core_schema: Mapping[str, Any], handler: GetJsonSchemaHandler
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
return {"type": "string", "format": "email"}
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def __get_pydantic_core_schema__(
|
||||||
|
cls, source: type[Any], handler: Callable[[Any], Mapping[str, Any]]
|
||||||
|
) -> Mapping[str, Any]:
|
||||||
|
return with_info_plain_validator_function(cls._validate)
|
||||||
|
|
||||||
|
|
||||||
|
class BaseModelWithConfig(BaseModel):
|
||||||
|
model_config = {"extra": "allow"}
|
||||||
|
|
||||||
|
|
||||||
|
class Contact(BaseModelWithConfig):
|
||||||
|
name: str | None = None
|
||||||
|
url: AnyUrl | None = None
|
||||||
|
email: EmailStr | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class License(BaseModelWithConfig):
|
||||||
|
name: str
|
||||||
|
identifier: str | None = None
|
||||||
|
url: AnyUrl | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class Info(BaseModelWithConfig):
|
||||||
|
title: str
|
||||||
|
summary: str | None = None
|
||||||
|
description: str | None = None
|
||||||
|
termsOfService: str | None = None
|
||||||
|
contact: Contact | None = None
|
||||||
|
license: License | None = None
|
||||||
|
version: str
|
||||||
|
|
||||||
|
|
||||||
|
class ServerVariable(BaseModelWithConfig):
|
||||||
|
enum: Annotated[list[str] | None, Field(min_length=1)] = None
|
||||||
|
default: str
|
||||||
|
description: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class Server(BaseModelWithConfig):
|
||||||
|
url: AnyUrl | str
|
||||||
|
description: str | None = None
|
||||||
|
variables: dict[str, ServerVariable] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class Reference(BaseModel):
|
||||||
|
ref: str = Field(alias="$ref")
|
||||||
|
|
||||||
|
|
||||||
|
class Discriminator(BaseModel):
|
||||||
|
propertyName: str
|
||||||
|
mapping: dict[str, str] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class XML(BaseModelWithConfig):
|
||||||
|
name: str | None = None
|
||||||
|
namespace: str | None = None
|
||||||
|
prefix: str | None = None
|
||||||
|
attribute: bool | None = None
|
||||||
|
wrapped: bool | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class ExternalDocumentation(BaseModelWithConfig):
|
||||||
|
description: str | None = None
|
||||||
|
url: AnyUrl
|
||||||
|
|
||||||
|
|
||||||
|
# Ref JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation#name-type
|
||||||
|
SchemaType = Literal[
|
||||||
|
"array", "boolean", "integer", "null", "number", "object", "string"
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
class Schema(BaseModelWithConfig):
|
||||||
|
# Ref: JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-core.html#name-the-json-schema-core-vocabu
|
||||||
|
# Core Vocabulary
|
||||||
|
schema_: str | None = Field(default=None, alias="$schema")
|
||||||
|
vocabulary: str | None = Field(default=None, alias="$vocabulary")
|
||||||
|
id: str | None = Field(default=None, alias="$id")
|
||||||
|
anchor: str | None = Field(default=None, alias="$anchor")
|
||||||
|
dynamicAnchor: str | None = Field(default=None, alias="$dynamicAnchor")
|
||||||
|
ref: str | None = Field(default=None, alias="$ref")
|
||||||
|
dynamicRef: str | None = Field(default=None, alias="$dynamicRef")
|
||||||
|
defs: dict[str, "SchemaOrBool"] | None = Field(default=None, alias="$defs")
|
||||||
|
comment: str | None = Field(default=None, alias="$comment")
|
||||||
|
# Ref: JSON Schema 2020-12: https://json-schema.org/draft/2020-12/json-schema-core.html#name-a-vocabulary-for-applying-s
|
||||||
|
# A Vocabulary for Applying Subschemas
|
||||||
|
allOf: list["SchemaOrBool"] | None = None
|
||||||
|
anyOf: list["SchemaOrBool"] | None = None
|
||||||
|
oneOf: list["SchemaOrBool"] | None = None
|
||||||
|
not_: Optional["SchemaOrBool"] = Field(default=None, alias="not")
|
||||||
|
if_: Optional["SchemaOrBool"] = Field(default=None, alias="if")
|
||||||
|
then: Optional["SchemaOrBool"] = None
|
||||||
|
else_: Optional["SchemaOrBool"] = Field(default=None, alias="else")
|
||||||
|
dependentSchemas: dict[str, "SchemaOrBool"] | None = None
|
||||||
|
prefixItems: list["SchemaOrBool"] | None = None
|
||||||
|
items: Optional["SchemaOrBool"] = None
|
||||||
|
contains: Optional["SchemaOrBool"] = None
|
||||||
|
properties: dict[str, "SchemaOrBool"] | None = None
|
||||||
|
patternProperties: dict[str, "SchemaOrBool"] | None = None
|
||||||
|
additionalProperties: Optional["SchemaOrBool"] = None
|
||||||
|
propertyNames: Optional["SchemaOrBool"] = None
|
||||||
|
unevaluatedItems: Optional["SchemaOrBool"] = None
|
||||||
|
unevaluatedProperties: Optional["SchemaOrBool"] = None
|
||||||
|
# Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-structural
|
||||||
|
# A Vocabulary for Structural Validation
|
||||||
|
type: SchemaType | list[SchemaType] | None = None
|
||||||
|
enum: list[Any] | None = None
|
||||||
|
const: Any | None = None
|
||||||
|
multipleOf: float | None = Field(default=None, gt=0)
|
||||||
|
maximum: float | None = None
|
||||||
|
exclusiveMaximum: float | None = None
|
||||||
|
minimum: float | None = None
|
||||||
|
exclusiveMinimum: float | None = None
|
||||||
|
maxLength: int | None = Field(default=None, ge=0)
|
||||||
|
minLength: int | None = Field(default=None, ge=0)
|
||||||
|
pattern: str | None = None
|
||||||
|
maxItems: int | None = Field(default=None, ge=0)
|
||||||
|
minItems: int | None = Field(default=None, ge=0)
|
||||||
|
uniqueItems: bool | None = None
|
||||||
|
maxContains: int | None = Field(default=None, ge=0)
|
||||||
|
minContains: int | None = Field(default=None, ge=0)
|
||||||
|
maxProperties: int | None = Field(default=None, ge=0)
|
||||||
|
minProperties: int | None = Field(default=None, ge=0)
|
||||||
|
required: list[str] | None = None
|
||||||
|
dependentRequired: dict[str, set[str]] | None = None
|
||||||
|
# Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-vocabularies-for-semantic-c
|
||||||
|
# Vocabularies for Semantic Content With "format"
|
||||||
|
format: str | None = None
|
||||||
|
# Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-the-conten
|
||||||
|
# A Vocabulary for the Contents of String-Encoded Data
|
||||||
|
contentEncoding: str | None = None
|
||||||
|
contentMediaType: str | None = None
|
||||||
|
contentSchema: Optional["SchemaOrBool"] = None
|
||||||
|
# Ref: JSON Schema Validation 2020-12: https://json-schema.org/draft/2020-12/json-schema-validation.html#name-a-vocabulary-for-basic-meta
|
||||||
|
# A Vocabulary for Basic Meta-Data Annotations
|
||||||
|
title: str | None = None
|
||||||
|
description: str | None = None
|
||||||
|
default: Any | None = None
|
||||||
|
deprecated: bool | None = None
|
||||||
|
readOnly: bool | None = None
|
||||||
|
writeOnly: bool | None = None
|
||||||
|
examples: list[Any] | None = None
|
||||||
|
# Ref: OpenAPI 3.1.0: https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.1.0.md#schema-object
|
||||||
|
# Schema Object
|
||||||
|
discriminator: Discriminator | None = None
|
||||||
|
xml: XML | None = None
|
||||||
|
externalDocs: ExternalDocumentation | None = None
|
||||||
|
example: Annotated[
|
||||||
|
Any | None,
|
||||||
|
typing_deprecated(
|
||||||
|
"Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, "
|
||||||
|
"although still supported. Use examples instead."
|
||||||
|
),
|
||||||
|
] = None
|
||||||
|
|
||||||
|
|
||||||
|
# Ref: https://json-schema.org/draft/2020-12/json-schema-core.html#name-json-schema-documents
|
||||||
|
# A JSON Schema MUST be an object or a boolean.
|
||||||
|
SchemaOrBool = Schema | bool
|
||||||
|
|
||||||
|
|
||||||
|
class Example(TypedDict, total=False):
|
||||||
|
summary: str | None
|
||||||
|
description: str | None
|
||||||
|
value: Any | None
|
||||||
|
externalValue: AnyUrl | None
|
||||||
|
|
||||||
|
__pydantic_config__ = {"extra": "allow"} # type: ignore[misc] # ty: ignore[invalid-typed-dict-statement]
|
||||||
|
|
||||||
|
|
||||||
|
class ParameterInType(Enum):
|
||||||
|
query = "query"
|
||||||
|
header = "header"
|
||||||
|
path = "path"
|
||||||
|
cookie = "cookie"
|
||||||
|
|
||||||
|
|
||||||
|
class Encoding(BaseModelWithConfig):
|
||||||
|
contentType: str | None = None
|
||||||
|
headers: dict[str, Union["Header", Reference]] | None = None
|
||||||
|
style: str | None = None
|
||||||
|
explode: bool | None = None
|
||||||
|
allowReserved: bool | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class MediaType(BaseModelWithConfig):
|
||||||
|
schema_: Schema | Reference | None = Field(default=None, alias="schema")
|
||||||
|
example: Any | None = None
|
||||||
|
examples: dict[str, Example | Reference] | None = None
|
||||||
|
encoding: dict[str, Encoding] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class ParameterBase(BaseModelWithConfig):
|
||||||
|
description: str | None = None
|
||||||
|
required: bool | None = None
|
||||||
|
deprecated: bool | None = None
|
||||||
|
# Serialization rules for simple scenarios
|
||||||
|
style: str | None = None
|
||||||
|
explode: bool | None = None
|
||||||
|
allowReserved: bool | None = None
|
||||||
|
schema_: Schema | Reference | None = Field(default=None, alias="schema")
|
||||||
|
example: Any | None = None
|
||||||
|
examples: dict[str, Example | Reference] | None = None
|
||||||
|
# Serialization rules for more complex scenarios
|
||||||
|
content: dict[str, MediaType] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class Parameter(ParameterBase):
|
||||||
|
name: str
|
||||||
|
in_: ParameterInType = Field(alias="in")
|
||||||
|
|
||||||
|
|
||||||
|
class Header(ParameterBase):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
class RequestBody(BaseModelWithConfig):
|
||||||
|
description: str | None = None
|
||||||
|
content: dict[str, MediaType]
|
||||||
|
required: bool | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class Link(BaseModelWithConfig):
|
||||||
|
operationRef: str | None = None
|
||||||
|
operationId: str | None = None
|
||||||
|
parameters: dict[str, Any | str] | None = None
|
||||||
|
requestBody: Any | str | None = None
|
||||||
|
description: str | None = None
|
||||||
|
server: Server | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class Response(BaseModelWithConfig):
|
||||||
|
description: str
|
||||||
|
headers: dict[str, Header | Reference] | None = None
|
||||||
|
content: dict[str, MediaType] | None = None
|
||||||
|
links: dict[str, Link | Reference] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class Operation(BaseModelWithConfig):
|
||||||
|
tags: list[str] | None = None
|
||||||
|
summary: str | None = None
|
||||||
|
description: str | None = None
|
||||||
|
externalDocs: ExternalDocumentation | None = None
|
||||||
|
operationId: str | None = None
|
||||||
|
parameters: list[Parameter | Reference] | None = None
|
||||||
|
requestBody: RequestBody | Reference | None = None
|
||||||
|
# Using Any for Specification Extensions
|
||||||
|
responses: dict[str, Response | Any] | None = None
|
||||||
|
callbacks: dict[str, dict[str, "PathItem"] | Reference] | None = None
|
||||||
|
deprecated: bool | None = None
|
||||||
|
security: list[dict[str, list[str]]] | None = None
|
||||||
|
servers: list[Server] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class PathItem(BaseModelWithConfig):
|
||||||
|
ref: str | None = Field(default=None, alias="$ref")
|
||||||
|
summary: str | None = None
|
||||||
|
description: str | None = None
|
||||||
|
get: Operation | None = None
|
||||||
|
put: Operation | None = None
|
||||||
|
post: Operation | None = None
|
||||||
|
delete: Operation | None = None
|
||||||
|
options: Operation | None = None
|
||||||
|
head: Operation | None = None
|
||||||
|
patch: Operation | None = None
|
||||||
|
trace: Operation | None = None
|
||||||
|
servers: list[Server] | None = None
|
||||||
|
parameters: list[Parameter | Reference] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class SecuritySchemeType(Enum):
|
||||||
|
apiKey = "apiKey"
|
||||||
|
http = "http"
|
||||||
|
oauth2 = "oauth2"
|
||||||
|
openIdConnect = "openIdConnect"
|
||||||
|
|
||||||
|
|
||||||
|
class SecurityBase(BaseModelWithConfig):
|
||||||
|
type_: SecuritySchemeType = Field(alias="type")
|
||||||
|
description: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class APIKeyIn(Enum):
|
||||||
|
query = "query"
|
||||||
|
header = "header"
|
||||||
|
cookie = "cookie"
|
||||||
|
|
||||||
|
|
||||||
|
class APIKey(SecurityBase):
|
||||||
|
type_: SecuritySchemeType = Field(default=SecuritySchemeType.apiKey, alias="type")
|
||||||
|
in_: APIKeyIn = Field(alias="in")
|
||||||
|
name: str
|
||||||
|
|
||||||
|
|
||||||
|
class HTTPBase(SecurityBase):
|
||||||
|
type_: SecuritySchemeType = Field(default=SecuritySchemeType.http, alias="type")
|
||||||
|
scheme: str
|
||||||
|
|
||||||
|
|
||||||
|
class HTTPBearer(HTTPBase):
|
||||||
|
scheme: Literal["bearer"] = "bearer"
|
||||||
|
bearerFormat: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class OAuthFlow(BaseModelWithConfig):
|
||||||
|
refreshUrl: str | None = None
|
||||||
|
scopes: dict[str, str] = {}
|
||||||
|
|
||||||
|
|
||||||
|
class OAuthFlowImplicit(OAuthFlow):
|
||||||
|
authorizationUrl: str
|
||||||
|
|
||||||
|
|
||||||
|
class OAuthFlowPassword(OAuthFlow):
|
||||||
|
tokenUrl: str
|
||||||
|
|
||||||
|
|
||||||
|
class OAuthFlowClientCredentials(OAuthFlow):
|
||||||
|
tokenUrl: str
|
||||||
|
|
||||||
|
|
||||||
|
class OAuthFlowAuthorizationCode(OAuthFlow):
|
||||||
|
authorizationUrl: str
|
||||||
|
tokenUrl: str
|
||||||
|
|
||||||
|
|
||||||
|
class OAuthFlows(BaseModelWithConfig):
|
||||||
|
implicit: OAuthFlowImplicit | None = None
|
||||||
|
password: OAuthFlowPassword | None = None
|
||||||
|
clientCredentials: OAuthFlowClientCredentials | None = None
|
||||||
|
authorizationCode: OAuthFlowAuthorizationCode | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class OAuth2(SecurityBase):
|
||||||
|
type_: SecuritySchemeType = Field(default=SecuritySchemeType.oauth2, alias="type")
|
||||||
|
flows: OAuthFlows
|
||||||
|
|
||||||
|
|
||||||
|
class OpenIdConnect(SecurityBase):
|
||||||
|
type_: SecuritySchemeType = Field(
|
||||||
|
default=SecuritySchemeType.openIdConnect, alias="type"
|
||||||
|
)
|
||||||
|
openIdConnectUrl: str
|
||||||
|
|
||||||
|
|
||||||
|
SecurityScheme = APIKey | HTTPBase | OAuth2 | OpenIdConnect | HTTPBearer
|
||||||
|
|
||||||
|
|
||||||
|
class Components(BaseModelWithConfig):
|
||||||
|
schemas: dict[str, Schema | Reference] | None = None
|
||||||
|
responses: dict[str, Response | Reference] | None = None
|
||||||
|
parameters: dict[str, Parameter | Reference] | None = None
|
||||||
|
examples: dict[str, Example | Reference] | None = None
|
||||||
|
requestBodies: dict[str, RequestBody | Reference] | None = None
|
||||||
|
headers: dict[str, Header | Reference] | None = None
|
||||||
|
securitySchemes: dict[str, SecurityScheme | Reference] | None = None
|
||||||
|
links: dict[str, Link | Reference] | None = None
|
||||||
|
# Using Any for Specification Extensions
|
||||||
|
callbacks: dict[str, dict[str, PathItem] | Reference | Any] | None = None
|
||||||
|
pathItems: dict[str, PathItem | Reference] | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class Tag(BaseModelWithConfig):
|
||||||
|
name: str
|
||||||
|
description: str | None = None
|
||||||
|
externalDocs: ExternalDocumentation | None = None
|
||||||
|
|
||||||
|
|
||||||
|
class OpenAPI(BaseModelWithConfig):
|
||||||
|
openapi: str
|
||||||
|
info: Info
|
||||||
|
jsonSchemaDialect: str | None = None
|
||||||
|
servers: list[Server] | None = None
|
||||||
|
# Using Any for Specification Extensions
|
||||||
|
paths: dict[str, PathItem | Any] | None = None
|
||||||
|
webhooks: dict[str, PathItem | Reference] | None = None
|
||||||
|
components: Components | None = None
|
||||||
|
security: list[dict[str, list[str]]] | None = None
|
||||||
|
tags: list[Tag] | None = None
|
||||||
|
externalDocs: ExternalDocumentation | None = None
|
||||||
|
|
||||||
|
|
||||||
|
Schema.model_rebuild()
|
||||||
|
Operation.model_rebuild()
|
||||||
|
Encoding.model_rebuild()
|
||||||
617
venv/Lib/site-packages/fastapi/openapi/utils.py
Normal file
617
venv/Lib/site-packages/fastapi/openapi/utils.py
Normal file
@@ -0,0 +1,617 @@
|
|||||||
|
import copy
|
||||||
|
import http.client
|
||||||
|
import inspect
|
||||||
|
import warnings
|
||||||
|
from collections.abc import Sequence
|
||||||
|
from typing import Any, Literal, cast
|
||||||
|
|
||||||
|
from fastapi import routing
|
||||||
|
from fastapi._compat import (
|
||||||
|
ModelField,
|
||||||
|
get_definitions,
|
||||||
|
get_flat_models_from_fields,
|
||||||
|
get_model_name_map,
|
||||||
|
get_schema_from_model_field,
|
||||||
|
lenient_issubclass,
|
||||||
|
)
|
||||||
|
from fastapi.datastructures import DefaultPlaceholder, _Unset
|
||||||
|
from fastapi.dependencies.models import Dependant
|
||||||
|
from fastapi.dependencies.utils import (
|
||||||
|
_get_flat_fields_from_params,
|
||||||
|
get_flat_dependant,
|
||||||
|
get_flat_params,
|
||||||
|
get_validation_alias,
|
||||||
|
)
|
||||||
|
from fastapi.encoders import jsonable_encoder
|
||||||
|
from fastapi.exceptions import FastAPIDeprecationWarning
|
||||||
|
from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX
|
||||||
|
from fastapi.openapi.models import OpenAPI
|
||||||
|
from fastapi.params import Body, ParamTypes
|
||||||
|
from fastapi.responses import Response
|
||||||
|
from fastapi.sse import _SSE_EVENT_SCHEMA
|
||||||
|
from fastapi.types import ModelNameMap
|
||||||
|
from fastapi.utils import (
|
||||||
|
deep_dict_update,
|
||||||
|
generate_operation_id_for_path,
|
||||||
|
is_body_allowed_for_status_code,
|
||||||
|
)
|
||||||
|
from pydantic import BaseModel
|
||||||
|
from starlette.responses import JSONResponse
|
||||||
|
from starlette.routing import BaseRoute
|
||||||
|
|
||||||
|
validation_error_definition = {
|
||||||
|
"title": "ValidationError",
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"loc": {
|
||||||
|
"title": "Location",
|
||||||
|
"type": "array",
|
||||||
|
"items": {"anyOf": [{"type": "string"}, {"type": "integer"}]},
|
||||||
|
},
|
||||||
|
"msg": {"title": "Message", "type": "string"},
|
||||||
|
"type": {"title": "Error Type", "type": "string"},
|
||||||
|
"input": {"title": "Input"},
|
||||||
|
"ctx": {"title": "Context", "type": "object"},
|
||||||
|
},
|
||||||
|
"required": ["loc", "msg", "type"],
|
||||||
|
}
|
||||||
|
|
||||||
|
validation_error_response_definition = {
|
||||||
|
"title": "HTTPValidationError",
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"detail": {
|
||||||
|
"title": "Detail",
|
||||||
|
"type": "array",
|
||||||
|
"items": {"$ref": REF_PREFIX + "ValidationError"},
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
status_code_ranges: dict[str, str] = {
|
||||||
|
"1XX": "Information",
|
||||||
|
"2XX": "Success",
|
||||||
|
"3XX": "Redirection",
|
||||||
|
"4XX": "Client Error",
|
||||||
|
"5XX": "Server Error",
|
||||||
|
"DEFAULT": "Default Response",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def get_openapi_security_definitions(
|
||||||
|
flat_dependant: Dependant,
|
||||||
|
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
|
||||||
|
security_definitions = {}
|
||||||
|
# Use a dict to merge scopes for same security scheme
|
||||||
|
operation_security_dict: dict[str, list[str]] = {}
|
||||||
|
for security_dependency in flat_dependant._security_dependencies:
|
||||||
|
security_definition = jsonable_encoder(
|
||||||
|
security_dependency._security_scheme.model,
|
||||||
|
by_alias=True,
|
||||||
|
exclude_none=True,
|
||||||
|
)
|
||||||
|
security_name = security_dependency._security_scheme.scheme_name
|
||||||
|
security_definitions[security_name] = security_definition
|
||||||
|
# Merge scopes for the same security scheme
|
||||||
|
if security_name not in operation_security_dict:
|
||||||
|
operation_security_dict[security_name] = []
|
||||||
|
for scope in security_dependency.oauth_scopes or []:
|
||||||
|
if scope not in operation_security_dict[security_name]:
|
||||||
|
operation_security_dict[security_name].append(scope)
|
||||||
|
operation_security = [
|
||||||
|
{name: scopes} for name, scopes in operation_security_dict.items()
|
||||||
|
]
|
||||||
|
return security_definitions, operation_security
|
||||||
|
|
||||||
|
|
||||||
|
def _get_openapi_operation_parameters(
|
||||||
|
*,
|
||||||
|
dependant: Dependant,
|
||||||
|
model_name_map: ModelNameMap,
|
||||||
|
field_mapping: dict[
|
||||||
|
tuple[ModelField, Literal["validation", "serialization"]], dict[str, Any]
|
||||||
|
],
|
||||||
|
separate_input_output_schemas: bool = True,
|
||||||
|
) -> list[dict[str, Any]]:
|
||||||
|
parameters = []
|
||||||
|
flat_dependant = get_flat_dependant(dependant, skip_repeats=True)
|
||||||
|
path_params = _get_flat_fields_from_params(flat_dependant.path_params)
|
||||||
|
query_params = _get_flat_fields_from_params(flat_dependant.query_params)
|
||||||
|
header_params = _get_flat_fields_from_params(flat_dependant.header_params)
|
||||||
|
cookie_params = _get_flat_fields_from_params(flat_dependant.cookie_params)
|
||||||
|
parameter_groups = [
|
||||||
|
(ParamTypes.path, path_params),
|
||||||
|
(ParamTypes.query, query_params),
|
||||||
|
(ParamTypes.header, header_params),
|
||||||
|
(ParamTypes.cookie, cookie_params),
|
||||||
|
]
|
||||||
|
default_convert_underscores = True
|
||||||
|
if len(flat_dependant.header_params) == 1:
|
||||||
|
first_field = flat_dependant.header_params[0]
|
||||||
|
if lenient_issubclass(first_field.field_info.annotation, BaseModel):
|
||||||
|
default_convert_underscores = getattr(
|
||||||
|
first_field.field_info, "convert_underscores", True
|
||||||
|
)
|
||||||
|
for param_type, param_group in parameter_groups:
|
||||||
|
for param in param_group:
|
||||||
|
field_info = param.field_info
|
||||||
|
# field_info = cast(Param, field_info)
|
||||||
|
if not getattr(field_info, "include_in_schema", True):
|
||||||
|
continue
|
||||||
|
param_schema = get_schema_from_model_field(
|
||||||
|
field=param,
|
||||||
|
model_name_map=model_name_map,
|
||||||
|
field_mapping=field_mapping,
|
||||||
|
separate_input_output_schemas=separate_input_output_schemas,
|
||||||
|
)
|
||||||
|
name = get_validation_alias(param)
|
||||||
|
convert_underscores = getattr(
|
||||||
|
param.field_info,
|
||||||
|
"convert_underscores",
|
||||||
|
default_convert_underscores,
|
||||||
|
)
|
||||||
|
if (
|
||||||
|
param_type == ParamTypes.header
|
||||||
|
and name == param.name
|
||||||
|
and convert_underscores
|
||||||
|
):
|
||||||
|
name = param.name.replace("_", "-")
|
||||||
|
|
||||||
|
parameter = {
|
||||||
|
"name": name,
|
||||||
|
"in": param_type.value,
|
||||||
|
"required": param.field_info.is_required(),
|
||||||
|
"schema": param_schema,
|
||||||
|
}
|
||||||
|
if field_info.description:
|
||||||
|
parameter["description"] = field_info.description
|
||||||
|
openapi_examples = getattr(field_info, "openapi_examples", None)
|
||||||
|
example = getattr(field_info, "example", None)
|
||||||
|
if openapi_examples:
|
||||||
|
parameter["examples"] = jsonable_encoder(openapi_examples)
|
||||||
|
elif example is not _Unset:
|
||||||
|
parameter["example"] = jsonable_encoder(example)
|
||||||
|
if getattr(field_info, "deprecated", None):
|
||||||
|
parameter["deprecated"] = True
|
||||||
|
parameters.append(parameter)
|
||||||
|
return parameters
|
||||||
|
|
||||||
|
|
||||||
|
def get_openapi_operation_request_body(
|
||||||
|
*,
|
||||||
|
body_field: ModelField | None,
|
||||||
|
model_name_map: ModelNameMap,
|
||||||
|
field_mapping: dict[
|
||||||
|
tuple[ModelField, Literal["validation", "serialization"]], dict[str, Any]
|
||||||
|
],
|
||||||
|
separate_input_output_schemas: bool = True,
|
||||||
|
) -> dict[str, Any] | None:
|
||||||
|
if not body_field:
|
||||||
|
return None
|
||||||
|
assert isinstance(body_field, ModelField)
|
||||||
|
body_schema = get_schema_from_model_field(
|
||||||
|
field=body_field,
|
||||||
|
model_name_map=model_name_map,
|
||||||
|
field_mapping=field_mapping,
|
||||||
|
separate_input_output_schemas=separate_input_output_schemas,
|
||||||
|
)
|
||||||
|
field_info = cast(Body, body_field.field_info)
|
||||||
|
request_media_type = field_info.media_type
|
||||||
|
required = body_field.field_info.is_required()
|
||||||
|
request_body_oai: dict[str, Any] = {}
|
||||||
|
if required:
|
||||||
|
request_body_oai["required"] = required
|
||||||
|
request_media_content: dict[str, Any] = {"schema": body_schema}
|
||||||
|
if field_info.openapi_examples:
|
||||||
|
request_media_content["examples"] = jsonable_encoder(
|
||||||
|
field_info.openapi_examples
|
||||||
|
)
|
||||||
|
elif field_info.example is not _Unset:
|
||||||
|
request_media_content["example"] = jsonable_encoder(field_info.example)
|
||||||
|
request_body_oai["content"] = {request_media_type: request_media_content}
|
||||||
|
return request_body_oai
|
||||||
|
|
||||||
|
|
||||||
|
def generate_operation_id(
|
||||||
|
*, route: routing._APIRouteLike, method: str
|
||||||
|
) -> str: # pragma: nocover
|
||||||
|
warnings.warn(
|
||||||
|
message="fastapi.openapi.utils.generate_operation_id() was deprecated, "
|
||||||
|
"it is not used internally, and will be removed soon",
|
||||||
|
category=FastAPIDeprecationWarning,
|
||||||
|
stacklevel=2,
|
||||||
|
)
|
||||||
|
if route.operation_id:
|
||||||
|
return route.operation_id
|
||||||
|
path: str = route.path_format
|
||||||
|
return generate_operation_id_for_path(name=route.name, path=path, method=method)
|
||||||
|
|
||||||
|
|
||||||
|
def generate_operation_summary(*, route: routing._APIRouteLike, method: str) -> str:
|
||||||
|
if route.summary:
|
||||||
|
return route.summary
|
||||||
|
return route.name.replace("_", " ").title()
|
||||||
|
|
||||||
|
|
||||||
|
def get_openapi_operation_metadata(
|
||||||
|
*, route: routing._APIRouteLike, method: str, operation_ids: set[str]
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
operation: dict[str, Any] = {}
|
||||||
|
if route.tags:
|
||||||
|
operation["tags"] = route.tags
|
||||||
|
operation["summary"] = generate_operation_summary(route=route, method=method)
|
||||||
|
if route.description:
|
||||||
|
operation["description"] = route.description
|
||||||
|
operation_id = route.operation_id or route.unique_id
|
||||||
|
if operation_id in operation_ids:
|
||||||
|
endpoint_name = getattr(route.endpoint, "__name__", "<unnamed_endpoint>")
|
||||||
|
message = f"Duplicate Operation ID {operation_id} for function {endpoint_name}"
|
||||||
|
file_name = getattr(route.endpoint, "__globals__", {}).get("__file__")
|
||||||
|
if file_name:
|
||||||
|
message += f" at {file_name}"
|
||||||
|
warnings.warn(message, stacklevel=1)
|
||||||
|
operation_ids.add(operation_id)
|
||||||
|
operation["operationId"] = operation_id
|
||||||
|
if route.deprecated:
|
||||||
|
operation["deprecated"] = route.deprecated
|
||||||
|
return operation
|
||||||
|
|
||||||
|
|
||||||
|
def get_openapi_path(
|
||||||
|
*,
|
||||||
|
route: routing._APIRouteLike,
|
||||||
|
operation_ids: set[str],
|
||||||
|
model_name_map: ModelNameMap,
|
||||||
|
field_mapping: dict[
|
||||||
|
tuple[ModelField, Literal["validation", "serialization"]], dict[str, Any]
|
||||||
|
],
|
||||||
|
separate_input_output_schemas: bool = True,
|
||||||
|
) -> tuple[dict[str, Any], dict[str, Any], dict[str, Any]]:
|
||||||
|
path = {}
|
||||||
|
security_schemes: dict[str, Any] = {}
|
||||||
|
definitions: dict[str, Any] = {}
|
||||||
|
assert route.methods is not None, "Methods must be a list"
|
||||||
|
if isinstance(route.response_class, DefaultPlaceholder):
|
||||||
|
current_response_class: type[Response] = route.response_class.value
|
||||||
|
else:
|
||||||
|
current_response_class = route.response_class
|
||||||
|
assert current_response_class, "A response class is needed to generate OpenAPI"
|
||||||
|
route_response_media_type: str | None = current_response_class.media_type
|
||||||
|
if route.include_in_schema:
|
||||||
|
for method in route.methods:
|
||||||
|
operation = get_openapi_operation_metadata(
|
||||||
|
route=route, method=method, operation_ids=operation_ids
|
||||||
|
)
|
||||||
|
parameters: list[dict[str, Any]] = []
|
||||||
|
flat_dependant = get_flat_dependant(route.dependant, skip_repeats=True)
|
||||||
|
security_definitions, operation_security = get_openapi_security_definitions(
|
||||||
|
flat_dependant=flat_dependant
|
||||||
|
)
|
||||||
|
if operation_security:
|
||||||
|
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=cast(routing._APIRouteLike, 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 is_body_allowed_for_status_code(route.status_code):
|
||||||
|
# Check for JSONL streaming (generator endpoints)
|
||||||
|
if route.is_json_stream:
|
||||||
|
jsonl_content: dict[str, Any] = {}
|
||||||
|
if route.stream_item_field:
|
||||||
|
item_schema = get_schema_from_model_field(
|
||||||
|
field=route.stream_item_field,
|
||||||
|
model_name_map=model_name_map,
|
||||||
|
field_mapping=field_mapping,
|
||||||
|
separate_input_output_schemas=separate_input_output_schemas,
|
||||||
|
)
|
||||||
|
jsonl_content["itemSchema"] = item_schema
|
||||||
|
else:
|
||||||
|
jsonl_content["itemSchema"] = {}
|
||||||
|
operation.setdefault("responses", {}).setdefault(
|
||||||
|
status_code, {}
|
||||||
|
).setdefault("content", {})["application/jsonl"] = jsonl_content
|
||||||
|
elif route.is_sse_stream:
|
||||||
|
sse_content: dict[str, Any] = {}
|
||||||
|
item_schema = copy.deepcopy(_SSE_EVENT_SCHEMA)
|
||||||
|
if route.stream_item_field:
|
||||||
|
content_schema = get_schema_from_model_field(
|
||||||
|
field=route.stream_item_field,
|
||||||
|
model_name_map=model_name_map,
|
||||||
|
field_mapping=field_mapping,
|
||||||
|
separate_input_output_schemas=separate_input_output_schemas,
|
||||||
|
)
|
||||||
|
item_schema["required"] = ["data"]
|
||||||
|
item_schema["properties"]["data"] = {
|
||||||
|
"type": "string",
|
||||||
|
"contentMediaType": "application/json",
|
||||||
|
"contentSchema": content_schema,
|
||||||
|
}
|
||||||
|
sse_content["itemSchema"] = item_schema
|
||||||
|
operation.setdefault("responses", {}).setdefault(
|
||||||
|
status_code, {}
|
||||||
|
).setdefault("content", {})["text/event-stream"] = sse_content
|
||||||
|
elif route_response_media_type:
|
||||||
|
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)
|
||||||
|
process_response.pop("model", None)
|
||||||
|
status_code_key = str(additional_status_code).upper()
|
||||||
|
if status_code_key == "DEFAULT":
|
||||||
|
status_code_key = "default"
|
||||||
|
openapi_response = operation_responses.setdefault(
|
||||||
|
status_code_key, {}
|
||||||
|
)
|
||||||
|
assert isinstance(process_response, dict), (
|
||||||
|
"An additional response must be a dict"
|
||||||
|
)
|
||||||
|
field = route.response_fields.get(additional_status_code)
|
||||||
|
additional_field_schema: dict[str, Any] | None = None
|
||||||
|
if field:
|
||||||
|
additional_field_schema = get_schema_from_model_field(
|
||||||
|
field=field,
|
||||||
|
model_name_map=model_name_map,
|
||||||
|
field_mapping=field_mapping,
|
||||||
|
separate_input_output_schemas=separate_input_output_schemas,
|
||||||
|
)
|
||||||
|
media_type = route_response_media_type or "application/json"
|
||||||
|
additional_schema = (
|
||||||
|
process_response.setdefault("content", {})
|
||||||
|
.setdefault(media_type, {})
|
||||||
|
.setdefault("schema", {})
|
||||||
|
)
|
||||||
|
deep_dict_update(additional_schema, additional_field_schema)
|
||||||
|
status_text: str | None = status_code_ranges.get(
|
||||||
|
str(additional_status_code).upper()
|
||||||
|
) or http.client.responses.get(int(additional_status_code))
|
||||||
|
description = (
|
||||||
|
process_response.get("description")
|
||||||
|
or openapi_response.get("description")
|
||||||
|
or status_text
|
||||||
|
or "Additional Response"
|
||||||
|
)
|
||||||
|
deep_dict_update(openapi_response, process_response)
|
||||||
|
openapi_response["description"] = description
|
||||||
|
http422 = "422"
|
||||||
|
all_route_params = get_flat_params(route.dependant)
|
||||||
|
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": {
|
||||||
|
"schema": {"$ref": REF_PREFIX + "HTTPValidationError"}
|
||||||
|
}
|
||||||
|
},
|
||||||
|
}
|
||||||
|
if "ValidationError" not in definitions:
|
||||||
|
definitions.update(
|
||||||
|
{
|
||||||
|
"ValidationError": validation_error_definition,
|
||||||
|
"HTTPValidationError": validation_error_response_definition,
|
||||||
|
}
|
||||||
|
)
|
||||||
|
if route.openapi_extra:
|
||||||
|
deep_dict_update(operation, route.openapi_extra)
|
||||||
|
path[method.lower()] = operation
|
||||||
|
return path, security_schemes, definitions
|
||||||
|
|
||||||
|
|
||||||
|
def _get_api_route_for_openapi(
|
||||||
|
route_context: routing.RouteContext,
|
||||||
|
) -> routing._APIRouteLike | None:
|
||||||
|
if isinstance(route_context.original_route, routing.APIRoute):
|
||||||
|
return cast(routing._APIRouteLike, route_context)
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def get_fields_from_routes(
|
||||||
|
routes: Sequence[BaseRoute | routing.RouteContext],
|
||||||
|
) -> list[ModelField]:
|
||||||
|
body_fields_from_routes: list[ModelField] = []
|
||||||
|
responses_from_routes: list[ModelField] = []
|
||||||
|
request_fields_from_routes: list[ModelField] = []
|
||||||
|
callback_flat_models: list[ModelField] = []
|
||||||
|
for route_context in routing.iter_route_contexts(routes):
|
||||||
|
api_route = _get_api_route_for_openapi(route_context)
|
||||||
|
if api_route is None:
|
||||||
|
continue
|
||||||
|
if api_route.include_in_schema:
|
||||||
|
if api_route.body_field:
|
||||||
|
assert isinstance(api_route.body_field, ModelField), (
|
||||||
|
"A request body must be a Pydantic Field"
|
||||||
|
)
|
||||||
|
body_fields_from_routes.append(api_route.body_field)
|
||||||
|
if api_route.response_field:
|
||||||
|
responses_from_routes.append(api_route.response_field)
|
||||||
|
if api_route.response_fields:
|
||||||
|
responses_from_routes.extend(api_route.response_fields.values())
|
||||||
|
if api_route.stream_item_field:
|
||||||
|
responses_from_routes.append(api_route.stream_item_field)
|
||||||
|
if api_route.callbacks:
|
||||||
|
callback_flat_models.extend(get_fields_from_routes(api_route.callbacks))
|
||||||
|
params = get_flat_params(api_route.dependant)
|
||||||
|
request_fields_from_routes.extend(params)
|
||||||
|
|
||||||
|
flat_models = callback_flat_models + list(
|
||||||
|
body_fields_from_routes + responses_from_routes + request_fields_from_routes
|
||||||
|
)
|
||||||
|
return flat_models
|
||||||
|
|
||||||
|
|
||||||
|
def get_openapi(
|
||||||
|
*,
|
||||||
|
title: str,
|
||||||
|
version: str,
|
||||||
|
openapi_version: str = "3.1.0",
|
||||||
|
summary: str | None = None,
|
||||||
|
description: str | None = None,
|
||||||
|
routes: Sequence[BaseRoute | routing.RouteContext],
|
||||||
|
webhooks: Sequence[BaseRoute | routing.RouteContext] | None = None,
|
||||||
|
tags: list[dict[str, Any]] | None = None,
|
||||||
|
servers: list[dict[str, str | Any]] | None = None,
|
||||||
|
terms_of_service: str | None = None,
|
||||||
|
contact: dict[str, str | Any] | None = None,
|
||||||
|
license_info: dict[str, str | Any] | None = None,
|
||||||
|
separate_input_output_schemas: bool = True,
|
||||||
|
external_docs: dict[str, Any] | None = None,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
info: dict[str, Any] = {"title": title, "version": version}
|
||||||
|
if summary:
|
||||||
|
info["summary"] = summary
|
||||||
|
if description:
|
||||||
|
info["description"] = description
|
||||||
|
if terms_of_service:
|
||||||
|
info["termsOfService"] = terms_of_service
|
||||||
|
if contact:
|
||||||
|
info["contact"] = contact
|
||||||
|
if license_info:
|
||||||
|
info["license"] = license_info
|
||||||
|
output: dict[str, Any] = {"openapi": openapi_version, "info": info}
|
||||||
|
if servers:
|
||||||
|
output["servers"] = servers
|
||||||
|
components: dict[str, dict[str, Any]] = {}
|
||||||
|
paths: dict[str, dict[str, Any]] = {}
|
||||||
|
webhook_paths: dict[str, dict[str, Any]] = {}
|
||||||
|
operation_ids: set[str] = set()
|
||||||
|
all_fields = get_fields_from_routes(list(routes) + list(webhooks or []))
|
||||||
|
flat_models = get_flat_models_from_fields(all_fields, known_models=set())
|
||||||
|
model_name_map = get_model_name_map(flat_models)
|
||||||
|
field_mapping, definitions = get_definitions(
|
||||||
|
fields=all_fields,
|
||||||
|
model_name_map=model_name_map,
|
||||||
|
separate_input_output_schemas=separate_input_output_schemas,
|
||||||
|
)
|
||||||
|
for route_context in routing.iter_route_contexts(routes):
|
||||||
|
api_route = _get_api_route_for_openapi(route_context)
|
||||||
|
if api_route is not None:
|
||||||
|
result = get_openapi_path(
|
||||||
|
route=api_route,
|
||||||
|
operation_ids=operation_ids,
|
||||||
|
model_name_map=model_name_map,
|
||||||
|
field_mapping=field_mapping,
|
||||||
|
separate_input_output_schemas=separate_input_output_schemas,
|
||||||
|
)
|
||||||
|
if result:
|
||||||
|
path, security_schemes, path_definitions = result
|
||||||
|
if path:
|
||||||
|
paths.setdefault(api_route.path_format, {}).update(path)
|
||||||
|
if security_schemes:
|
||||||
|
components.setdefault("securitySchemes", {}).update(
|
||||||
|
security_schemes
|
||||||
|
)
|
||||||
|
if path_definitions:
|
||||||
|
definitions.update(path_definitions)
|
||||||
|
for webhook_context in routing.iter_route_contexts(webhooks or []):
|
||||||
|
api_webhook = _get_api_route_for_openapi(webhook_context)
|
||||||
|
if api_webhook is not None:
|
||||||
|
result = get_openapi_path(
|
||||||
|
route=api_webhook,
|
||||||
|
operation_ids=operation_ids,
|
||||||
|
model_name_map=model_name_map,
|
||||||
|
field_mapping=field_mapping,
|
||||||
|
separate_input_output_schemas=separate_input_output_schemas,
|
||||||
|
)
|
||||||
|
if result:
|
||||||
|
path, security_schemes, path_definitions = result
|
||||||
|
if path:
|
||||||
|
webhook_paths.setdefault(api_webhook.path_format, {}).update(path)
|
||||||
|
if security_schemes:
|
||||||
|
components.setdefault("securitySchemes", {}).update(
|
||||||
|
security_schemes
|
||||||
|
)
|
||||||
|
if path_definitions:
|
||||||
|
definitions.update(path_definitions)
|
||||||
|
if definitions:
|
||||||
|
components["schemas"] = {k: definitions[k] for k in sorted(definitions)}
|
||||||
|
if components:
|
||||||
|
output["components"] = components
|
||||||
|
output["paths"] = paths
|
||||||
|
if webhook_paths:
|
||||||
|
output["webhooks"] = webhook_paths
|
||||||
|
if tags:
|
||||||
|
output["tags"] = tags
|
||||||
|
if external_docs:
|
||||||
|
output["externalDocs"] = external_docs
|
||||||
|
return jsonable_encoder(OpenAPI(**output), by_alias=True, exclude_none=True) # type: ignore[no-any-return]
|
||||||
Reference in New Issue
Block a user