mirror of
https://github.com/fastapi/fastapi.git
synced 2026-09-14 13:36:21 +08:00
Sebastián Ramírez
7dad5a820b
* ✨ Update OpenAPI models for JSON Schema 2020-12 and OpenAPI 3.1.0 * ✨ Add support for summary and webhooks * ✨ Update JSON Schema for UploadFiles * ⏪️ Revert making paths optional, to ensure always correctness * ⏪️ Keep UploadFile as format: binary for compatibility with the rest of Pydantic bytes fields in v1 * ✨ Update version of OpenAPI generated to 3.1.0 * ✨ Update the version of Swagger UI * 📝 Update docs about extending OpenAPI * 📝 Update docs and links to refer to OpenAPI 3.1.0 * ✨ Update logic for handling webhooks * ♻️ Update parameter functions and classes, deprecate example and make examples the main field * ✅ Update tests for OpenAPI 3.1.0 * 📝 Update examples for OpenAPI metadata * ✅ Add and update tests for OpenAPI metadata * 📝 Add source example for webhooks * 📝 Update docs for metadata * 📝 Update docs for Schema extra * 📝 Add docs for webhooks * 🔧 Add webhooks docs to MkDocs * ✅ Update tests for extending OpenAPI * ✅ Add tests for webhooks * ♻️ Refactor generation of OpenAPI and JSON Schema with params * 📝 Update source examples for field examples * ✅ Update tests for examples * ➕ Make sure the minimum version of typing-extensions installed has deprecated() (already a dependency of Pydantic) * ✏️ Fix typo in Webhooks example code * 🔥 Remove commented out code of removed nullable field * 🗑️ Add deprecation warnings for example argument * ✅ Update tests to check for deprecation warnings * ✅ Add test for webhooks with security schemes, for coverage * 🍱 Update image for metadata, with new summary * 🍱 Add docs image for Webhooks * 📝 Update docs for webhooks, add docs UI image
944 lines
40 KiB
Python
944 lines
40 KiB
Python
from enum import Enum
|
|
from typing import (
|
|
Any,
|
|
Awaitable,
|
|
Callable,
|
|
Coroutine,
|
|
Dict,
|
|
List,
|
|
Optional,
|
|
Sequence,
|
|
Type,
|
|
TypeVar,
|
|
Union,
|
|
)
|
|
|
|
from fastapi import routing
|
|
from fastapi.datastructures import Default, DefaultPlaceholder
|
|
from fastapi.encoders import DictIntStrAny, SetIntStr
|
|
from fastapi.exception_handlers import (
|
|
http_exception_handler,
|
|
request_validation_exception_handler,
|
|
websocket_request_validation_exception_handler,
|
|
)
|
|
from fastapi.exceptions import RequestValidationError, WebSocketRequestValidationError
|
|
from fastapi.logger import logger
|
|
from fastapi.middleware.asyncexitstack import AsyncExitStackMiddleware
|
|
from fastapi.openapi.docs import (
|
|
get_redoc_html,
|
|
get_swagger_ui_html,
|
|
get_swagger_ui_oauth2_redirect_html,
|
|
)
|
|
from fastapi.openapi.utils import get_openapi
|
|
from fastapi.params import Depends
|
|
from fastapi.types import DecoratedCallable
|
|
from fastapi.utils import generate_unique_id
|
|
from starlette.applications import Starlette
|
|
from starlette.datastructures import State
|
|
from starlette.exceptions import HTTPException
|
|
from starlette.middleware import Middleware
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
from starlette.middleware.errors import ServerErrorMiddleware
|
|
from starlette.middleware.exceptions import ExceptionMiddleware
|
|
from starlette.requests import Request
|
|
from starlette.responses import HTMLResponse, JSONResponse, Response
|
|
from starlette.routing import BaseRoute
|
|
from starlette.types import ASGIApp, Lifespan, Receive, Scope, Send
|
|
|
|
AppType = TypeVar("AppType", bound="FastAPI")
|
|
|
|
|
|
class FastAPI(Starlette):
|
|
def __init__(
|
|
self: AppType,
|
|
*,
|
|
debug: bool = False,
|
|
routes: Optional[List[BaseRoute]] = None,
|
|
title: str = "FastAPI",
|
|
summary: Optional[str] = None,
|
|
description: str = "",
|
|
version: str = "0.1.0",
|
|
openapi_url: Optional[str] = "/openapi.json",
|
|
openapi_tags: Optional[List[Dict[str, Any]]] = None,
|
|
servers: Optional[List[Dict[str, Union[str, Any]]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
default_response_class: Type[Response] = Default(JSONResponse),
|
|
redirect_slashes: bool = True,
|
|
docs_url: Optional[str] = "/docs",
|
|
redoc_url: Optional[str] = "/redoc",
|
|
swagger_ui_oauth2_redirect_url: Optional[str] = "/docs/oauth2-redirect",
|
|
swagger_ui_init_oauth: Optional[Dict[str, Any]] = None,
|
|
middleware: Optional[Sequence[Middleware]] = None,
|
|
exception_handlers: Optional[
|
|
Dict[
|
|
Union[int, Type[Exception]],
|
|
Callable[[Request, Any], Coroutine[Any, Any, Response]],
|
|
]
|
|
] = None,
|
|
on_startup: Optional[Sequence[Callable[[], Any]]] = None,
|
|
on_shutdown: Optional[Sequence[Callable[[], Any]]] = None,
|
|
lifespan: Optional[Lifespan[AppType]] = None,
|
|
terms_of_service: Optional[str] = None,
|
|
contact: Optional[Dict[str, Union[str, Any]]] = None,
|
|
license_info: Optional[Dict[str, Union[str, Any]]] = None,
|
|
openapi_prefix: str = "",
|
|
root_path: str = "",
|
|
root_path_in_servers: bool = True,
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
webhooks: Optional[routing.APIRouter] = None,
|
|
deprecated: Optional[bool] = None,
|
|
include_in_schema: bool = True,
|
|
swagger_ui_parameters: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
**extra: Any,
|
|
) -> None:
|
|
self.debug = debug
|
|
self.title = title
|
|
self.summary = summary
|
|
self.description = description
|
|
self.version = version
|
|
self.terms_of_service = terms_of_service
|
|
self.contact = contact
|
|
self.license_info = license_info
|
|
self.openapi_url = openapi_url
|
|
self.openapi_tags = openapi_tags
|
|
self.root_path_in_servers = root_path_in_servers
|
|
self.docs_url = docs_url
|
|
self.redoc_url = redoc_url
|
|
self.swagger_ui_oauth2_redirect_url = swagger_ui_oauth2_redirect_url
|
|
self.swagger_ui_init_oauth = swagger_ui_init_oauth
|
|
self.swagger_ui_parameters = swagger_ui_parameters
|
|
self.servers = servers or []
|
|
self.extra = extra
|
|
self.openapi_version = "3.1.0"
|
|
self.openapi_schema: Optional[Dict[str, Any]] = None
|
|
if self.openapi_url:
|
|
assert self.title, "A title must be provided for OpenAPI, e.g.: 'My API'"
|
|
assert self.version, "A version must be provided for OpenAPI, e.g.: '2.1.0'"
|
|
# TODO: remove when discarding the openapi_prefix parameter
|
|
if openapi_prefix:
|
|
logger.warning(
|
|
'"openapi_prefix" has been deprecated in favor of "root_path", which '
|
|
"follows more closely the ASGI standard, is simpler, and more "
|
|
"automatic. Check the docs at "
|
|
"https://fastapi.tiangolo.com/advanced/sub-applications/"
|
|
)
|
|
self.webhooks = webhooks or routing.APIRouter()
|
|
self.root_path = root_path or openapi_prefix
|
|
self.state: State = State()
|
|
self.dependency_overrides: Dict[Callable[..., Any], Callable[..., Any]] = {}
|
|
self.router: routing.APIRouter = routing.APIRouter(
|
|
routes=routes,
|
|
redirect_slashes=redirect_slashes,
|
|
dependency_overrides_provider=self,
|
|
on_startup=on_startup,
|
|
on_shutdown=on_shutdown,
|
|
lifespan=lifespan,
|
|
default_response_class=default_response_class,
|
|
dependencies=dependencies,
|
|
callbacks=callbacks,
|
|
deprecated=deprecated,
|
|
include_in_schema=include_in_schema,
|
|
responses=responses,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
self.exception_handlers: Dict[
|
|
Any, Callable[[Request, Any], Union[Response, Awaitable[Response]]]
|
|
] = ({} if exception_handlers is None else dict(exception_handlers))
|
|
self.exception_handlers.setdefault(HTTPException, http_exception_handler)
|
|
self.exception_handlers.setdefault(
|
|
RequestValidationError, request_validation_exception_handler
|
|
)
|
|
self.exception_handlers.setdefault(
|
|
WebSocketRequestValidationError,
|
|
# Starlette still has incorrect type specification for the handlers
|
|
websocket_request_validation_exception_handler, # type: ignore
|
|
)
|
|
|
|
self.user_middleware: List[Middleware] = (
|
|
[] if middleware is None else list(middleware)
|
|
)
|
|
self.middleware_stack: Union[ASGIApp, None] = None
|
|
self.setup()
|
|
|
|
def build_middleware_stack(self) -> ASGIApp:
|
|
# Duplicate/override from Starlette to add AsyncExitStackMiddleware
|
|
# inside of ExceptionMiddleware, inside of custom user middlewares
|
|
debug = self.debug
|
|
error_handler = None
|
|
exception_handlers = {}
|
|
|
|
for key, value in self.exception_handlers.items():
|
|
if key in (500, Exception):
|
|
error_handler = value
|
|
else:
|
|
exception_handlers[key] = value
|
|
|
|
middleware = (
|
|
[Middleware(ServerErrorMiddleware, handler=error_handler, debug=debug)]
|
|
+ self.user_middleware
|
|
+ [
|
|
Middleware(
|
|
ExceptionMiddleware, handlers=exception_handlers, debug=debug
|
|
),
|
|
# Add FastAPI-specific AsyncExitStackMiddleware for dependencies with
|
|
# contextvars.
|
|
# This needs to happen after user middlewares because those create a
|
|
# new contextvars context copy by using a new AnyIO task group.
|
|
# The initial part of dependencies with yield is executed in the
|
|
# FastAPI code, inside all the middlewares, but the teardown part
|
|
# (after yield) is executed in the AsyncExitStack in this middleware,
|
|
# if the AsyncExitStack lived outside of the custom middlewares and
|
|
# contextvars were set in a dependency with yield in that internal
|
|
# contextvars context, the values would not be available in the
|
|
# outside context of the AsyncExitStack.
|
|
# By putting the middleware and the AsyncExitStack here, inside all
|
|
# user middlewares, the code before and after yield in dependencies
|
|
# with yield is executed in the same contextvars context, so all values
|
|
# set in contextvars before yield is still available after yield as
|
|
# would be expected.
|
|
# Additionally, by having this AsyncExitStack here, after the
|
|
# ExceptionMiddleware, now dependencies can catch handled exceptions,
|
|
# e.g. HTTPException, to customize the teardown code (e.g. DB session
|
|
# rollback).
|
|
Middleware(AsyncExitStackMiddleware),
|
|
]
|
|
)
|
|
|
|
app = self.router
|
|
for cls, options in reversed(middleware):
|
|
app = cls(app=app, **options)
|
|
return app
|
|
|
|
def openapi(self) -> Dict[str, Any]:
|
|
if not self.openapi_schema:
|
|
self.openapi_schema = get_openapi(
|
|
title=self.title,
|
|
version=self.version,
|
|
openapi_version=self.openapi_version,
|
|
summary=self.summary,
|
|
description=self.description,
|
|
terms_of_service=self.terms_of_service,
|
|
contact=self.contact,
|
|
license_info=self.license_info,
|
|
routes=self.routes,
|
|
webhooks=self.webhooks.routes,
|
|
tags=self.openapi_tags,
|
|
servers=self.servers,
|
|
)
|
|
return self.openapi_schema
|
|
|
|
def setup(self) -> None:
|
|
if self.openapi_url:
|
|
urls = (server_data.get("url") for server_data in self.servers)
|
|
server_urls = {url for url in urls if url}
|
|
|
|
async def openapi(req: Request) -> JSONResponse:
|
|
root_path = req.scope.get("root_path", "").rstrip("/")
|
|
if root_path not in server_urls:
|
|
if root_path and self.root_path_in_servers:
|
|
self.servers.insert(0, {"url": root_path})
|
|
server_urls.add(root_path)
|
|
return JSONResponse(self.openapi())
|
|
|
|
self.add_route(self.openapi_url, openapi, include_in_schema=False)
|
|
if self.openapi_url and self.docs_url:
|
|
|
|
async def swagger_ui_html(req: Request) -> HTMLResponse:
|
|
root_path = req.scope.get("root_path", "").rstrip("/")
|
|
openapi_url = root_path + self.openapi_url
|
|
oauth2_redirect_url = self.swagger_ui_oauth2_redirect_url
|
|
if oauth2_redirect_url:
|
|
oauth2_redirect_url = root_path + oauth2_redirect_url
|
|
return get_swagger_ui_html(
|
|
openapi_url=openapi_url,
|
|
title=self.title + " - Swagger UI",
|
|
oauth2_redirect_url=oauth2_redirect_url,
|
|
init_oauth=self.swagger_ui_init_oauth,
|
|
swagger_ui_parameters=self.swagger_ui_parameters,
|
|
)
|
|
|
|
self.add_route(self.docs_url, swagger_ui_html, include_in_schema=False)
|
|
|
|
if self.swagger_ui_oauth2_redirect_url:
|
|
|
|
async def swagger_ui_redirect(req: Request) -> HTMLResponse:
|
|
return get_swagger_ui_oauth2_redirect_html()
|
|
|
|
self.add_route(
|
|
self.swagger_ui_oauth2_redirect_url,
|
|
swagger_ui_redirect,
|
|
include_in_schema=False,
|
|
)
|
|
if self.openapi_url and self.redoc_url:
|
|
|
|
async def redoc_html(req: Request) -> HTMLResponse:
|
|
root_path = req.scope.get("root_path", "").rstrip("/")
|
|
openapi_url = root_path + self.openapi_url
|
|
return get_redoc_html(
|
|
openapi_url=openapi_url, title=self.title + " - ReDoc"
|
|
)
|
|
|
|
self.add_route(self.redoc_url, redoc_html, include_in_schema=False)
|
|
|
|
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
|
|
if self.root_path:
|
|
scope["root_path"] = self.root_path
|
|
await super().__call__(scope, receive, send)
|
|
|
|
def add_api_route(
|
|
self,
|
|
path: str,
|
|
endpoint: Callable[..., Coroutine[Any, Any, Response]],
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
methods: Optional[List[str]] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Union[Type[Response], DefaultPlaceholder] = Default(
|
|
JSONResponse
|
|
),
|
|
name: Optional[str] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> None:
|
|
self.router.add_api_route(
|
|
path,
|
|
endpoint=endpoint,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
methods=methods,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def api_route(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
methods: Optional[List[str]] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
def decorator(func: DecoratedCallable) -> DecoratedCallable:
|
|
self.router.add_api_route(
|
|
path,
|
|
func,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
methods=methods,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
return func
|
|
|
|
return decorator
|
|
|
|
def add_api_websocket_route(
|
|
self,
|
|
path: str,
|
|
endpoint: Callable[..., Any],
|
|
name: Optional[str] = None,
|
|
*,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
) -> None:
|
|
self.router.add_api_websocket_route(
|
|
path,
|
|
endpoint,
|
|
name=name,
|
|
dependencies=dependencies,
|
|
)
|
|
|
|
def websocket(
|
|
self,
|
|
path: str,
|
|
name: Optional[str] = None,
|
|
*,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
def decorator(func: DecoratedCallable) -> DecoratedCallable:
|
|
self.add_api_websocket_route(
|
|
path,
|
|
func,
|
|
name=name,
|
|
dependencies=dependencies,
|
|
)
|
|
return func
|
|
|
|
return decorator
|
|
|
|
def include_router(
|
|
self,
|
|
router: routing.APIRouter,
|
|
*,
|
|
prefix: str = "",
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
include_in_schema: bool = True,
|
|
default_response_class: Type[Response] = Default(JSONResponse),
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> None:
|
|
self.router.include_router(
|
|
router,
|
|
prefix=prefix,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
include_in_schema=include_in_schema,
|
|
default_response_class=default_response_class,
|
|
callbacks=callbacks,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def get(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.get(
|
|
path,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
callbacks=callbacks,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def put(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.put(
|
|
path,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
callbacks=callbacks,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def post(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.post(
|
|
path,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
callbacks=callbacks,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def delete(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.delete(
|
|
path,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
callbacks=callbacks,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def options(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.options(
|
|
path,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
callbacks=callbacks,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def head(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.head(
|
|
path,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
callbacks=callbacks,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def patch(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.patch(
|
|
path,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
callbacks=callbacks,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def trace(
|
|
self,
|
|
path: str,
|
|
*,
|
|
response_model: Any = Default(None),
|
|
status_code: Optional[int] = None,
|
|
tags: Optional[List[Union[str, Enum]]] = None,
|
|
dependencies: Optional[Sequence[Depends]] = None,
|
|
summary: Optional[str] = None,
|
|
description: Optional[str] = None,
|
|
response_description: str = "Successful Response",
|
|
responses: Optional[Dict[Union[int, str], Dict[str, Any]]] = None,
|
|
deprecated: Optional[bool] = None,
|
|
operation_id: Optional[str] = None,
|
|
response_model_include: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_exclude: Optional[Union[SetIntStr, DictIntStrAny]] = None,
|
|
response_model_by_alias: bool = True,
|
|
response_model_exclude_unset: bool = False,
|
|
response_model_exclude_defaults: bool = False,
|
|
response_model_exclude_none: bool = False,
|
|
include_in_schema: bool = True,
|
|
response_class: Type[Response] = Default(JSONResponse),
|
|
name: Optional[str] = None,
|
|
callbacks: Optional[List[BaseRoute]] = None,
|
|
openapi_extra: Optional[Dict[str, Any]] = None,
|
|
generate_unique_id_function: Callable[[routing.APIRoute], str] = Default(
|
|
generate_unique_id
|
|
),
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.trace(
|
|
path,
|
|
response_model=response_model,
|
|
status_code=status_code,
|
|
tags=tags,
|
|
dependencies=dependencies,
|
|
summary=summary,
|
|
description=description,
|
|
response_description=response_description,
|
|
responses=responses,
|
|
deprecated=deprecated,
|
|
operation_id=operation_id,
|
|
response_model_include=response_model_include,
|
|
response_model_exclude=response_model_exclude,
|
|
response_model_by_alias=response_model_by_alias,
|
|
response_model_exclude_unset=response_model_exclude_unset,
|
|
response_model_exclude_defaults=response_model_exclude_defaults,
|
|
response_model_exclude_none=response_model_exclude_none,
|
|
include_in_schema=include_in_schema,
|
|
response_class=response_class,
|
|
name=name,
|
|
callbacks=callbacks,
|
|
openapi_extra=openapi_extra,
|
|
generate_unique_id_function=generate_unique_id_function,
|
|
)
|
|
|
|
def websocket_route(
|
|
self, path: str, name: Union[str, None] = None
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
def decorator(func: DecoratedCallable) -> DecoratedCallable:
|
|
self.router.add_websocket_route(path, func, name=name)
|
|
return func
|
|
|
|
return decorator
|
|
|
|
def on_event(
|
|
self, event_type: str
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
return self.router.on_event(event_type)
|
|
|
|
def middleware(
|
|
self, middleware_type: str
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
def decorator(func: DecoratedCallable) -> DecoratedCallable:
|
|
self.add_middleware(BaseHTTPMiddleware, dispatch=func)
|
|
return func
|
|
|
|
return decorator
|
|
|
|
def exception_handler(
|
|
self, exc_class_or_status_code: Union[int, Type[Exception]]
|
|
) -> Callable[[DecoratedCallable], DecoratedCallable]:
|
|
def decorator(func: DecoratedCallable) -> DecoratedCallable:
|
|
self.add_exception_handler(exc_class_or_status_code, func)
|
|
return func
|
|
|
|
return decorator
|