| import contextlib |
| import copy |
| import email.message |
| import errno |
| import functools |
| import inspect |
| import json |
| import os |
| import stat |
| import threading |
| import types |
| from collections.abc import ( |
| AsyncIterator, |
| Awaitable, |
| Callable, |
| Collection, |
| Coroutine, |
| Generator, |
| Iterator, |
| Mapping, |
| Sequence, |
| ) |
| from contextlib import ( |
| AbstractAsyncContextManager, |
| AbstractContextManager, |
| AsyncExitStack, |
| asynccontextmanager, |
| ) |
| from contextvars import ContextVar |
| from dataclasses import dataclass, field |
| from enum import Enum, IntEnum |
| from typing import ( |
| Annotated, |
| Any, |
| Literal, |
| Protocol, |
| TypeVar, |
| cast, |
| ) |
|
|
| import anyio |
| from annotated_doc import Doc |
| from anyio.abc import ObjectReceiveStream |
| from fastapi import params |
| from fastapi._compat import ( |
| ModelField, |
| Undefined, |
| lenient_issubclass, |
| ) |
| from fastapi.datastructures import Default, DefaultPlaceholder |
| from fastapi.dependencies.models import ( |
| Dependant, |
| _is_async_gen_callable, |
| _is_coroutine_callable, |
| _is_gen_callable, |
| ) |
| from fastapi.dependencies.utils import ( |
| _should_embed_body_fields, |
| get_body_field, |
| get_dependant, |
| get_flat_dependant, |
| get_parameterless_sub_dependant, |
| get_stream_item_type, |
| get_typed_return_annotation, |
| solve_dependencies, |
| ) |
| from fastapi.encoders import jsonable_encoder |
| from fastapi.exceptions import ( |
| EndpointContext, |
| FastAPIError, |
| RequestValidationError, |
| ResponseValidationError, |
| WebSocketRequestValidationError, |
| ) |
| from fastapi.sse import ( |
| _PING_INTERVAL, |
| KEEPALIVE_COMMENT, |
| EventSourceResponse, |
| ServerSentEvent, |
| format_sse_event, |
| ) |
| from fastapi.types import DecoratedCallable, IncEx |
| from fastapi.utils import ( |
| create_model_field, |
| generate_unique_id, |
| get_value_or_default, |
| is_body_allowed_for_status_code, |
| ) |
| from starlette import routing |
| from starlette._exception_handler import wrap_app_handling_exceptions |
| from starlette._utils import get_route_path, is_async_callable |
| from starlette.concurrency import iterate_in_threadpool, run_in_threadpool |
| from starlette.datastructures import URL, FormData, URLPath |
| from starlette.exceptions import HTTPException |
| from starlette.requests import Request |
| from starlette.responses import ( |
| JSONResponse, |
| PlainTextResponse, |
| RedirectResponse, |
| Response, |
| StreamingResponse, |
| ) |
| from starlette.routing import ( |
| BaseRoute, |
| Match, |
| NoMatchFound, |
| compile_path, |
| get_name, |
| ) |
| from starlette.routing import Mount as Mount |
| from starlette.staticfiles import StaticFiles |
| from starlette.types import AppType, ASGIApp, Lifespan, Receive, Scope, Send |
| from starlette.websockets import WebSocket |
| from typing_extensions import deprecated |
|
|
|
|
| |
| |
| def request_response( |
| func: Callable[[Request], Awaitable[Response] | Response], |
| ) -> ASGIApp: |
| """ |
| Takes a function or coroutine `func(request) -> response`, |
| and returns an ASGI application. |
| """ |
| f: Callable[[Request], Awaitable[Response]] = ( |
| func |
| if is_async_callable(func) |
| else functools.partial(run_in_threadpool, func) |
| ) |
|
|
| async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| request = Request(scope, receive, send) |
|
|
| async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| |
| response_awaited = False |
| async with AsyncExitStack() as request_stack: |
| scope["fastapi_inner_astack"] = request_stack |
| async with AsyncExitStack() as function_stack: |
| scope["fastapi_function_astack"] = function_stack |
| response = await f(request) |
| await response(scope, receive, send) |
| |
| response_awaited = True |
| if not response_awaited: |
| raise FastAPIError( |
| "Response not awaited. There's a high chance that the " |
| "application code is raising an exception and a dependency with yield " |
| "has a block with a bare except, or a block with except Exception, " |
| "and is not raising the exception again. Read more about it in the " |
| "docs: https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-with-yield/#dependencies-with-yield-and-except" |
| ) |
|
|
| |
| await wrap_app_handling_exceptions(app, request)(scope, receive, send) |
|
|
| return app |
|
|
|
|
| |
| |
| def websocket_session( |
| func: Callable[[WebSocket], Awaitable[None]], |
| ) -> ASGIApp: |
| """ |
| Takes a coroutine `func(session)`, and returns an ASGI application. |
| """ |
| |
|
|
| async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| session = WebSocket(scope, receive=receive, send=send) |
|
|
| async def app(scope: Scope, receive: Receive, send: Send) -> None: |
| async with AsyncExitStack() as request_stack: |
| scope["fastapi_inner_astack"] = request_stack |
| async with AsyncExitStack() as function_stack: |
| scope["fastapi_function_astack"] = function_stack |
| await func(session) |
|
|
| |
| await wrap_app_handling_exceptions(app, session)(scope, receive, send) |
|
|
| return app |
|
|
|
|
| _T = TypeVar("_T") |
|
|
|
|
| |
| class _AsyncLiftContextManager(AbstractAsyncContextManager[_T]): |
| """ |
| Wraps a synchronous context manager to make it async. |
| |
| This is vendored from Starlette to avoid importing private symbols. |
| """ |
|
|
| def __init__(self, cm: AbstractContextManager[_T]) -> None: |
| self._cm = cm |
|
|
| async def __aenter__(self) -> _T: |
| return self._cm.__enter__() |
|
|
| async def __aexit__( |
| self, |
| exc_type: type[BaseException] | None, |
| exc_value: BaseException | None, |
| traceback: types.TracebackType | None, |
| ) -> bool | None: |
| return self._cm.__exit__(exc_type, exc_value, traceback) |
|
|
|
|
| |
| def _wrap_gen_lifespan_context( |
| lifespan_context: Callable[[Any], Generator[Any, Any, Any]], |
| ) -> Callable[[Any], AbstractAsyncContextManager[Any]]: |
| """ |
| Wrap a generator-based lifespan context into an async context manager. |
| |
| This is vendored from Starlette to avoid importing private symbols. |
| """ |
| cmgr = contextlib.contextmanager(lifespan_context) |
|
|
| @functools.wraps(cmgr) |
| def wrapper(app: Any) -> _AsyncLiftContextManager[Any]: |
| return _AsyncLiftContextManager(cmgr(app)) |
|
|
| return wrapper |
|
|
|
|
| def _merge_lifespan_context( |
| original_context: Lifespan[Any], nested_context: Lifespan[Any] |
| ) -> Lifespan[Any]: |
| @asynccontextmanager |
| async def merged_lifespan( |
| app: AppType, |
| ) -> AsyncIterator[Mapping[str, Any] | None]: |
| async with original_context(app) as maybe_original_state: |
| async with nested_context(app) as maybe_nested_state: |
| if maybe_nested_state is None and maybe_original_state is None: |
| yield None |
| else: |
| yield {**(maybe_nested_state or {}), **(maybe_original_state or {})} |
|
|
| return merged_lifespan |
|
|
|
|
| class _DefaultLifespan: |
| """ |
| Default lifespan context manager that runs on_startup and on_shutdown handlers. |
| |
| This is a copy of the Starlette _DefaultLifespan class that was removed |
| in Starlette. FastAPI keeps it to maintain backward compatibility with |
| on_startup and on_shutdown event handlers. |
| |
| Ref: https://github.com/Kludex/starlette/pull/3117 |
| """ |
|
|
| def __init__(self, router: "APIRouter") -> None: |
| self._router = router |
|
|
| async def __aenter__(self) -> None: |
| await self._router._startup() |
|
|
| async def __aexit__(self, *exc_info: object) -> None: |
| await self._router._shutdown() |
|
|
| def __call__(self: _T, app: object) -> _T: |
| return self |
|
|
|
|
| |
| _endpoint_context_cache: dict[int, EndpointContext] = {} |
|
|
|
|
| def _extract_endpoint_context(func: Any) -> EndpointContext: |
| """Extract endpoint context with caching to avoid repeated file I/O.""" |
| func_id = id(func) |
|
|
| if func_id in _endpoint_context_cache: |
| return _endpoint_context_cache[func_id] |
|
|
| try: |
| ctx: EndpointContext = {} |
|
|
| if (source_file := inspect.getsourcefile(func)) is not None: |
| ctx["file"] = source_file |
| if (line_number := inspect.getsourcelines(func)[1]) is not None: |
| ctx["line"] = line_number |
| if (func_name := getattr(func, "__name__", None)) is not None: |
| ctx["function"] = func_name |
| except Exception: |
| ctx = EndpointContext() |
|
|
| _endpoint_context_cache[func_id] = ctx |
| return ctx |
|
|
|
|
| async def serialize_response( |
| *, |
| field: ModelField | None = None, |
| response_content: Any, |
| include: IncEx | None = None, |
| exclude: IncEx | None = None, |
| by_alias: bool = True, |
| exclude_unset: bool = False, |
| exclude_defaults: bool = False, |
| exclude_none: bool = False, |
| is_coroutine: bool = True, |
| endpoint_ctx: EndpointContext | None = None, |
| dump_json: bool = False, |
| ) -> Any: |
| if field: |
| if is_coroutine: |
| value, errors = field.validate(response_content, {}, loc=("response",)) |
| else: |
| value, errors = await run_in_threadpool( |
| field.validate, response_content, {}, loc=("response",) |
| ) |
| if errors: |
| ctx = endpoint_ctx or EndpointContext() |
| raise ResponseValidationError( |
| errors=errors, |
| body=response_content, |
| endpoint_ctx=ctx, |
| ) |
| serializer = field.serialize_json if dump_json else field.serialize |
| return serializer( |
| value, |
| include=include, |
| exclude=exclude, |
| by_alias=by_alias, |
| exclude_unset=exclude_unset, |
| exclude_defaults=exclude_defaults, |
| exclude_none=exclude_none, |
| ) |
|
|
| else: |
| return jsonable_encoder(response_content) |
|
|
|
|
| async def run_endpoint_function( |
| *, dependant: Dependant, values: dict[str, Any], is_coroutine: bool |
| ) -> Any: |
| |
| |
| assert dependant.call is not None, "dependant.call must be a function" |
|
|
| if is_coroutine: |
| return await dependant.call(**values) |
| else: |
| return await run_in_threadpool(dependant.call, **values) |
|
|
|
|
| def _build_response_args( |
| *, status_code: int | None, solved_result: Any |
| ) -> dict[str, Any]: |
| response_args: dict[str, Any] = { |
| "background": solved_result.background_tasks, |
| } |
| |
| |
| current_status_code = ( |
| status_code if status_code else solved_result.response.status_code |
| ) |
| if current_status_code is not None: |
| response_args["status_code"] = current_status_code |
| if solved_result.response.status_code: |
| response_args["status_code"] = solved_result.response.status_code |
| return response_args |
|
|
|
|
| def get_request_handler( |
| dependant: Dependant, |
| body_field: ModelField | None = None, |
| status_code: int | None = None, |
| response_class: type[Response] | DefaultPlaceholder = Default(JSONResponse), |
| response_field: ModelField | None = None, |
| response_model_include: IncEx | None = None, |
| response_model_exclude: IncEx | None = 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, |
| dependency_overrides_provider: Any | None = None, |
| embed_body_fields: bool = False, |
| strict_content_type: bool | DefaultPlaceholder = Default(True), |
| stream_item_field: ModelField | None = None, |
| is_json_stream: bool = False, |
| ) -> Callable[[Request], Coroutine[Any, Any, Response]]: |
| assert dependant.call is not None, "dependant.call must be a function" |
| is_coroutine = _is_coroutine_callable(dependant.call) |
| is_body_form = body_field and isinstance(body_field.field_info, params.Form) |
| if isinstance(response_class, DefaultPlaceholder): |
| actual_response_class: type[Response] = response_class.value |
| else: |
| actual_response_class = response_class |
| is_sse_stream = lenient_issubclass(actual_response_class, EventSourceResponse) |
| if isinstance(strict_content_type, DefaultPlaceholder): |
| actual_strict_content_type: bool = strict_content_type.value |
| else: |
| actual_strict_content_type = strict_content_type |
|
|
| async def app(request: Request) -> Response: |
| response: Response | None = None |
| file_stack = request.scope.get("fastapi_middleware_astack") |
| assert isinstance(file_stack, AsyncExitStack), ( |
| "fastapi_middleware_astack not found in request scope" |
| ) |
|
|
| |
| endpoint_ctx = ( |
| _extract_endpoint_context(dependant.call) |
| if dependant.call |
| else EndpointContext() |
| ) |
|
|
| if dependant.path: |
| |
| mount_path = request.scope.get("root_path", "").rstrip("/") |
| endpoint_ctx["path"] = f"{request.method} {mount_path}{dependant.path}" |
|
|
| |
| try: |
| body: Any = None |
| if body_field: |
| if is_body_form: |
| body = await request.form() |
| file_stack.push_async_callback(body.close) |
| else: |
| body_bytes = await request.body() |
| if body_bytes: |
| json_body: Any = Undefined |
| content_type_value = request.headers.get("content-type") |
| if not content_type_value: |
| if not actual_strict_content_type: |
| json_body = await request.json() |
| else: |
| message = email.message.Message() |
| message["content-type"] = content_type_value |
| if message.get_content_maintype() == "application": |
| subtype = message.get_content_subtype() |
| if subtype == "json" or subtype.endswith("+json"): |
| json_body = await request.json() |
| if json_body != Undefined: |
| body = json_body |
| else: |
| body = body_bytes |
| except json.JSONDecodeError as e: |
| validation_error = RequestValidationError( |
| [ |
| { |
| "type": "json_invalid", |
| "loc": ("body", e.pos), |
| "msg": "JSON decode error", |
| "input": {}, |
| "ctx": {"error": e.msg}, |
| } |
| ], |
| body=e.doc, |
| endpoint_ctx=endpoint_ctx, |
| ) |
| raise validation_error from e |
| except HTTPException: |
| |
| raise |
| except Exception as e: |
| http_error = HTTPException( |
| status_code=400, detail="There was an error parsing the body" |
| ) |
| raise http_error from e |
|
|
| |
| errors: list[Any] = [] |
| async_exit_stack = request.scope.get("fastapi_inner_astack") |
| assert isinstance(async_exit_stack, AsyncExitStack), ( |
| "fastapi_inner_astack not found in request scope" |
| ) |
| solved_result = await solve_dependencies( |
| request=request, |
| dependant=dependant, |
| body=cast(dict[str, Any] | FormData | bytes | None, body), |
| dependency_overrides_provider=dependency_overrides_provider, |
| async_exit_stack=async_exit_stack, |
| embed_body_fields=embed_body_fields, |
| ) |
| errors = solved_result.errors |
| assert dependant.call |
| if not errors: |
| |
| |
| |
| def _serialize_data(data: Any) -> bytes: |
| if stream_item_field: |
| value, errors_ = stream_item_field.validate( |
| data, {}, loc=("response",) |
| ) |
| if errors_: |
| ctx = endpoint_ctx or EndpointContext() |
| raise ResponseValidationError( |
| errors=errors_, |
| body=data, |
| endpoint_ctx=ctx, |
| ) |
| return stream_item_field.serialize_json( |
| value, |
| include=response_model_include, |
| exclude=response_model_exclude, |
| by_alias=response_model_by_alias, |
| exclude_unset=response_model_exclude_unset, |
| exclude_defaults=response_model_exclude_defaults, |
| exclude_none=response_model_exclude_none, |
| ) |
| else: |
| data = jsonable_encoder(data) |
| return json.dumps(data).encode("utf-8") |
|
|
| if is_sse_stream: |
| |
| gen = dependant.call(**solved_result.values) |
|
|
| def _serialize_sse_item(item: Any) -> bytes: |
| if isinstance(item, ServerSentEvent): |
| |
| |
| |
| |
| if item.raw_data is not None: |
| data_str: str | None = item.raw_data |
| elif item.data is not None: |
| if hasattr(item.data, "model_dump_json"): |
| data_str = item.data.model_dump_json() |
| else: |
| data_str = json.dumps(jsonable_encoder(item.data)) |
| else: |
| data_str = None |
| return format_sse_event( |
| data_str=data_str, |
| event=item.event, |
| id=item.id, |
| retry=item.retry, |
| comment=item.comment, |
| ) |
| else: |
| |
| |
| return format_sse_event( |
| data_str=_serialize_data(item).decode("utf-8") |
| ) |
|
|
| if _is_async_gen_callable(dependant.call): |
| sse_aiter: AsyncIterator[Any] = gen.__aiter__() |
| else: |
| sse_aiter = iterate_in_threadpool(gen) |
|
|
| @asynccontextmanager |
| async def _sse_producer_cm() -> AsyncIterator[ |
| ObjectReceiveStream[bytes] |
| ]: |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| send_stream, receive_stream = anyio.create_memory_object_stream[ |
| bytes |
| ](max_buffer_size=1) |
|
|
| async def _producer() -> None: |
| async with send_stream: |
| async for raw_item in sse_aiter: |
| await send_stream.send(_serialize_sse_item(raw_item)) |
|
|
| send_keepalive, receive_keepalive = ( |
| anyio.create_memory_object_stream[bytes](max_buffer_size=1) |
| ) |
|
|
| async def _keepalive_inserter() -> None: |
| """Read from the producer and forward to the output, |
| inserting keepalive comments on timeout.""" |
| async with send_keepalive, receive_stream: |
| try: |
| while True: |
| try: |
| with anyio.fail_after(_PING_INTERVAL): |
| data = await receive_stream.receive() |
| await send_keepalive.send(data) |
| except TimeoutError: |
| await send_keepalive.send(KEEPALIVE_COMMENT) |
| except anyio.EndOfStream: |
| pass |
|
|
| async with anyio.create_task_group() as tg: |
| tg.start_soon(_producer) |
| tg.start_soon(_keepalive_inserter) |
| yield receive_keepalive |
| tg.cancel_scope.cancel() |
|
|
| |
| |
| |
| |
| sse_receive_stream = await async_exit_stack.enter_async_context( |
| _sse_producer_cm() |
| ) |
| |
| |
| async_exit_stack.push_async_callback(sse_receive_stream.aclose) |
|
|
| async def _sse_with_checkpoints( |
| stream: ObjectReceiveStream[bytes], |
| ) -> AsyncIterator[bytes]: |
| async for data in stream: |
| yield data |
| |
| |
| |
| await anyio.sleep(0) |
|
|
| sse_stream_content: AsyncIterator[bytes] | Iterator[bytes] = ( |
| _sse_with_checkpoints(sse_receive_stream) |
| ) |
|
|
| response = StreamingResponse( |
| sse_stream_content, |
| media_type="text/event-stream", |
| background=solved_result.background_tasks, |
| ) |
| response.headers["Cache-Control"] = "no-cache" |
| |
| response.headers["X-Accel-Buffering"] = "no" |
| response.headers.raw.extend(solved_result.response.headers.raw) |
| elif is_json_stream: |
| |
| gen = dependant.call(**solved_result.values) |
|
|
| def _serialize_item(item: Any) -> bytes: |
| return _serialize_data(item) + b"\n" |
|
|
| if _is_async_gen_callable(dependant.call): |
|
|
| async def _async_stream_jsonl() -> AsyncIterator[bytes]: |
| async for item in gen: |
| yield _serialize_item(item) |
| |
| |
| await anyio.sleep(0) |
|
|
| jsonl_stream_content: AsyncIterator[bytes] | Iterator[bytes] = ( |
| _async_stream_jsonl() |
| ) |
| else: |
|
|
| def _sync_stream_jsonl() -> Iterator[bytes]: |
| for item in gen: |
| yield _serialize_item(item) |
|
|
| jsonl_stream_content = _sync_stream_jsonl() |
|
|
| response = StreamingResponse( |
| jsonl_stream_content, |
| media_type="application/jsonl", |
| background=solved_result.background_tasks, |
| ) |
| response.headers.raw.extend(solved_result.response.headers.raw) |
| elif _is_async_gen_callable(dependant.call) or _is_gen_callable( |
| dependant.call |
| ): |
| |
| gen = dependant.call(**solved_result.values) |
| if _is_async_gen_callable(dependant.call): |
|
|
| async def _async_stream_raw( |
| async_gen: AsyncIterator[Any], |
| ) -> AsyncIterator[Any]: |
| async for chunk in async_gen: |
| yield chunk |
| |
| |
| await anyio.sleep(0) |
|
|
| gen = _async_stream_raw(gen) |
| response_args = _build_response_args( |
| status_code=status_code, solved_result=solved_result |
| ) |
| response = actual_response_class(content=gen, **response_args) |
| response.headers.raw.extend(solved_result.response.headers.raw) |
| else: |
| raw_response = await run_endpoint_function( |
| dependant=dependant, |
| values=solved_result.values, |
| is_coroutine=is_coroutine, |
| ) |
| if isinstance(raw_response, Response): |
| if raw_response.background is None: |
| raw_response.background = solved_result.background_tasks |
| response = raw_response |
| else: |
| response_args = _build_response_args( |
| status_code=status_code, solved_result=solved_result |
| ) |
| |
| |
| |
| |
| |
| use_dump_json = response_field is not None and isinstance( |
| response_class, DefaultPlaceholder |
| ) |
| content = await serialize_response( |
| field=response_field, |
| response_content=raw_response, |
| include=response_model_include, |
| exclude=response_model_exclude, |
| by_alias=response_model_by_alias, |
| exclude_unset=response_model_exclude_unset, |
| exclude_defaults=response_model_exclude_defaults, |
| exclude_none=response_model_exclude_none, |
| is_coroutine=is_coroutine, |
| endpoint_ctx=endpoint_ctx, |
| dump_json=use_dump_json, |
| ) |
| if use_dump_json: |
| response = Response( |
| content=content, |
| media_type="application/json", |
| **response_args, |
| ) |
| else: |
| response = actual_response_class(content, **response_args) |
| if not is_body_allowed_for_status_code(response.status_code): |
| response.body = b"" |
| response.headers.raw.extend(solved_result.response.headers.raw) |
| if errors: |
| validation_error = RequestValidationError( |
| errors, body=body, endpoint_ctx=endpoint_ctx |
| ) |
| raise validation_error |
|
|
| |
| assert response |
| return response |
|
|
| return app |
|
|
|
|
| def get_websocket_app( |
| dependant: Dependant, |
| dependency_overrides_provider: Any | None = None, |
| embed_body_fields: bool = False, |
| ) -> Callable[[WebSocket], Coroutine[Any, Any, Any]]: |
| async def app(websocket: WebSocket) -> None: |
| endpoint_ctx = ( |
| _extract_endpoint_context(dependant.call) |
| if dependant.call |
| else EndpointContext() |
| ) |
| if dependant.path: |
| |
| mount_path = websocket.scope.get("root_path", "").rstrip("/") |
| endpoint_ctx["path"] = f"WS {mount_path}{dependant.path}" |
| async_exit_stack = websocket.scope.get("fastapi_inner_astack") |
| assert isinstance(async_exit_stack, AsyncExitStack), ( |
| "fastapi_inner_astack not found in request scope" |
| ) |
| solved_result = await solve_dependencies( |
| request=websocket, |
| dependant=dependant, |
| dependency_overrides_provider=dependency_overrides_provider, |
| async_exit_stack=async_exit_stack, |
| embed_body_fields=embed_body_fields, |
| ) |
| if solved_result.errors: |
| raise WebSocketRequestValidationError( |
| solved_result.errors, |
| endpoint_ctx=endpoint_ctx, |
| ) |
| assert dependant.call is not None, "dependant.call must be a function" |
| await dependant.call(**solved_result.values) |
|
|
| return app |
|
|
|
|
| class APIWebSocketRoute(routing.WebSocketRoute): |
| def __init__( |
| self, |
| path: str, |
| endpoint: Callable[..., Any], |
| *, |
| name: str | None = None, |
| dependencies: Sequence[params.Depends] | None = None, |
| dependency_overrides_provider: Any | None = None, |
| ) -> None: |
| self.path = path |
| self.endpoint = endpoint |
| self.name = get_name(endpoint) if name is None else name |
| self.dependencies = list(dependencies or []) |
| self.path_regex, self.path_format, self.param_convertors = compile_path(path) |
| ( |
| self.dependant, |
| self._flat_dependant, |
| self._embed_body_fields, |
| ) = _build_dependant_with_parameterless_dependencies( |
| path=self.path_format, |
| call=self.endpoint, |
| dependencies=self.dependencies, |
| ) |
| self.app = websocket_session( |
| get_websocket_app( |
| dependant=self.dependant, |
| dependency_overrides_provider=dependency_overrides_provider, |
| embed_body_fields=self._embed_body_fields, |
| ) |
| ) |
|
|
| def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| match, child_scope = super().matches(scope) |
| if match != Match.NONE: |
| child_scope["route"] = self |
| return match, child_scope |
|
|
|
|
| _FASTAPI_SCOPE_KEY = "fastapi" |
| _FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY = "effective_route_context" |
| _FASTAPI_FRONTEND_PATH_KEY = "frontend_path" |
| _FASTAPI_FRONTEND_SPECIFICITY_KEY = "frontend_specificity" |
| _FASTAPI_INCLUDED_ROUTER_KEY = "included_router" |
| _effective_route_context_var: ContextVar[Any | None] = ContextVar( |
| "fastapi_effective_route_context", default=None |
| ) |
| _SCOPE_MISSING = object() |
|
|
|
|
| def _frontend_dependency_endpoint() -> None: |
| pass |
|
|
|
|
| def _build_dependant_with_parameterless_dependencies( |
| *, |
| path: str, |
| call: Callable[..., Any], |
| dependencies: Sequence[params.Depends], |
| ) -> tuple[Dependant, Dependant, bool]: |
| dependant = get_dependant(path=path, call=call, scope="function") |
| for depends in dependencies[::-1]: |
| dependant.dependencies.insert( |
| 0, |
| get_parameterless_sub_dependant(depends=depends, path=path), |
| ) |
| flat_dependant = get_flat_dependant(dependant) |
| embed_body_fields = _should_embed_body_fields(flat_dependant.body_params) |
| return dependant, flat_dependant, embed_body_fields |
|
|
|
|
| class _RouteWithPath(Protocol): |
| path: str |
|
|
|
|
| def _get_fastapi_scope(scope: Scope) -> dict[str, Any]: |
| fastapi_scope = scope.setdefault(_FASTAPI_SCOPE_KEY, {}) |
| assert isinstance(fastapi_scope, dict) |
| return fastapi_scope |
|
|
|
|
| def _update_scope(scope: Scope, child_scope: Scope) -> None: |
| fastapi_child_scope = child_scope.get(_FASTAPI_SCOPE_KEY) |
| for key, value in child_scope.items(): |
| if key != _FASTAPI_SCOPE_KEY: |
| scope[key] = value |
| if isinstance(fastapi_child_scope, dict): |
| _get_fastapi_scope(scope).update(fastapi_child_scope) |
|
|
|
|
| def _get_scope_effective_route_context(scope: Scope) -> Any | None: |
| return scope.get(_FASTAPI_SCOPE_KEY, {}).get(_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY) |
|
|
|
|
| def _get_scope_included_router(scope: Scope) -> Any | None: |
| return scope.get(_FASTAPI_SCOPE_KEY, {}).get(_FASTAPI_INCLUDED_ROUTER_KEY) |
|
|
|
|
| def _frontend_scope_specificity(scope: Scope) -> int | None: |
| specificity = scope.get(_FASTAPI_SCOPE_KEY, {}).get( |
| _FASTAPI_FRONTEND_SPECIFICITY_KEY |
| ) |
| if isinstance(specificity, int): |
| return specificity |
| return None |
|
|
|
|
| def _restore_fastapi_scope_key(scope: Scope, key: str, previous: Any) -> None: |
| fastapi_scope = scope.get(_FASTAPI_SCOPE_KEY) |
| if not isinstance(fastapi_scope, dict): |
| return |
| if previous is _SCOPE_MISSING: |
| fastapi_scope.pop(key, None) |
| else: |
| fastapi_scope[key] = previous |
|
|
|
|
| class _APIRouteLike(Protocol): |
| path: str |
| endpoint: Callable[..., Any] |
| stream_item_type: Any | None |
| response_model: Any |
| summary: str | None |
| response_description: str |
| deprecated: bool | None |
| operation_id: str | None |
| response_model_include: IncEx | None |
| response_model_exclude: IncEx | None |
| response_model_by_alias: bool |
| response_model_exclude_unset: bool |
| response_model_exclude_defaults: bool |
| response_model_exclude_none: bool |
| include_in_schema: bool |
| response_class: type[Response] | DefaultPlaceholder |
| dependency_overrides_provider: Any | None |
| callbacks: list[BaseRoute] | None |
| openapi_extra: dict[str, Any] | None |
| generate_unique_id_function: Callable[[Any], str] | DefaultPlaceholder |
| strict_content_type: bool | DefaultPlaceholder |
| tags: list[str | Enum] |
| responses: dict[int | str, dict[str, Any]] |
| name: str |
| path_regex: Any |
| path_format: str |
| param_convertors: dict[str, Any] |
| methods: set[str] |
| unique_id: str |
| status_code: int | None |
| response_field: ModelField | None |
| stream_item_field: ModelField | None |
| dependencies: list[params.Depends] |
| description: str |
| response_fields: dict[int | str, ModelField] |
| dependant: Dependant |
| _flat_dependant: Dependant |
| _embed_body_fields: bool |
| body_field: ModelField | None |
| is_sse_stream: bool |
| is_json_stream: bool |
|
|
|
|
| def _populate_api_route_state( |
| route: _APIRouteLike, |
| path: str, |
| endpoint: Callable[..., Any], |
| *, |
| response_model: Any = Default(None), |
| status_code: int | None = None, |
| tags: list[str | Enum] | None = None, |
| dependencies: Sequence[params.Depends] | None = None, |
| summary: str | None = None, |
| description: str | None = None, |
| response_description: str = "Successful Response", |
| responses: dict[int | str, dict[str, Any]] | None = None, |
| deprecated: bool | None = None, |
| name: str | None = None, |
| methods: set[str] | list[str] | None = None, |
| operation_id: str | None = None, |
| response_model_include: IncEx | None = None, |
| response_model_exclude: IncEx | None = 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] | DefaultPlaceholder = Default(JSONResponse), |
| dependency_overrides_provider: Any | None = None, |
| callbacks: list[BaseRoute] | None = None, |
| openapi_extra: dict[str, Any] | None = None, |
| generate_unique_id_function: Callable[[Any], str] | DefaultPlaceholder = Default( |
| generate_unique_id |
| ), |
| strict_content_type: bool | DefaultPlaceholder = Default(True), |
| ) -> None: |
| route.path = path |
| route.endpoint = endpoint |
| route.stream_item_type = None |
| if isinstance(response_model, DefaultPlaceholder): |
| return_annotation = get_typed_return_annotation(endpoint) |
| if lenient_issubclass(return_annotation, Response): |
| response_model = None |
| else: |
| stream_item = get_stream_item_type(return_annotation) |
| if stream_item is not None: |
| |
| |
| |
| |
| |
| |
| if ( |
| isinstance(response_class, DefaultPlaceholder) |
| or lenient_issubclass(response_class, EventSourceResponse) |
| ) and not lenient_issubclass(stream_item, ServerSentEvent): |
| route.stream_item_type = stream_item |
| response_model = None |
| else: |
| response_model = return_annotation |
| route.response_model = response_model |
| route.summary = summary |
| route.response_description = response_description |
| route.deprecated = deprecated |
| route.operation_id = operation_id |
| route.response_model_include = response_model_include |
| route.response_model_exclude = response_model_exclude |
| route.response_model_by_alias = response_model_by_alias |
| route.response_model_exclude_unset = response_model_exclude_unset |
| route.response_model_exclude_defaults = response_model_exclude_defaults |
| route.response_model_exclude_none = response_model_exclude_none |
| route.include_in_schema = include_in_schema |
| route.response_class = response_class |
| route.dependency_overrides_provider = dependency_overrides_provider |
| route.callbacks = callbacks |
| route.openapi_extra = openapi_extra |
| route.generate_unique_id_function = generate_unique_id_function |
| route.strict_content_type = strict_content_type |
| route.tags = tags or [] |
| route.responses = responses or {} |
| route.name = get_name(endpoint) if name is None else name |
| route.path_regex, route.path_format, route.param_convertors = compile_path(path) |
| if methods is None: |
| methods = ["GET"] |
| route.methods = {method.upper() for method in methods} |
| if isinstance(generate_unique_id_function, DefaultPlaceholder): |
| current_generate_unique_id: Callable[[Any], str] = ( |
| generate_unique_id_function.value |
| ) |
| else: |
| current_generate_unique_id = generate_unique_id_function |
| route.unique_id = route.operation_id or current_generate_unique_id(route) |
| |
| if isinstance(status_code, IntEnum): |
| status_code = int(status_code) |
| route.status_code = status_code |
| if route.response_model: |
| assert is_body_allowed_for_status_code(status_code), ( |
| f"Status code {status_code} must not have a response body" |
| ) |
| response_name = "Response_" + route.unique_id |
| route.response_field = create_model_field( |
| name=response_name, |
| type_=route.response_model, |
| mode="serialization", |
| ) |
| else: |
| route.response_field = None |
| if route.stream_item_type: |
| stream_item_name = "StreamItem_" + route.unique_id |
| route.stream_item_field = create_model_field( |
| name=stream_item_name, |
| type_=route.stream_item_type, |
| mode="serialization", |
| ) |
| else: |
| route.stream_item_field = None |
| route.dependencies = list(dependencies or []) |
| route.description = description or inspect.cleandoc(route.endpoint.__doc__ or "") |
| |
| |
| route.description = route.description.split("\f")[0].strip() |
| response_fields = {} |
| for additional_status_code, response in route.responses.items(): |
| assert isinstance(response, dict), "An additional response must be a dict" |
| model = response.get("model") |
| if model: |
| assert is_body_allowed_for_status_code(additional_status_code), ( |
| f"Status code {additional_status_code} must not have a response body" |
| ) |
| response_name = f"Response_{additional_status_code}_{route.unique_id}" |
| response_field = create_model_field( |
| name=response_name, type_=model, mode="serialization" |
| ) |
| response_fields[additional_status_code] = response_field |
| if response_fields: |
| route.response_fields = response_fields |
| else: |
| route.response_fields = {} |
|
|
| assert callable(endpoint), "An endpoint must be a callable" |
| ( |
| route.dependant, |
| route._flat_dependant, |
| route._embed_body_fields, |
| ) = _build_dependant_with_parameterless_dependencies( |
| path=route.path_format, |
| call=route.endpoint, |
| dependencies=route.dependencies, |
| ) |
| route.body_field = get_body_field( |
| flat_dependant=route._flat_dependant, |
| name=route.unique_id, |
| embed_body_fields=route._embed_body_fields, |
| ) |
| |
| is_generator = _is_async_gen_callable(route.dependant.call) or _is_gen_callable( |
| route.dependant.call |
| ) |
| route.is_sse_stream = is_generator and lenient_issubclass( |
| response_class, EventSourceResponse |
| ) |
| route.is_json_stream = is_generator and isinstance( |
| response_class, DefaultPlaceholder |
| ) |
|
|
|
|
| class APIRoute(routing.Route): |
| stream_item_type: Any | None |
| response_model: Any |
| summary: str | None |
| response_description: str |
| deprecated: bool | None |
| operation_id: str | None |
| response_model_include: IncEx | None |
| response_model_exclude: IncEx | None |
| response_model_by_alias: bool |
| response_model_exclude_unset: bool |
| response_model_exclude_defaults: bool |
| response_model_exclude_none: bool |
| include_in_schema: bool |
| response_class: type[Response] | DefaultPlaceholder |
| dependency_overrides_provider: Any | None |
| callbacks: list[BaseRoute] | None |
| openapi_extra: dict[str, Any] | None |
| generate_unique_id_function: Callable[[Any], str] | DefaultPlaceholder |
| strict_content_type: bool | DefaultPlaceholder |
| tags: list[str | Enum] |
| responses: dict[int | str, dict[str, Any]] |
| unique_id: str |
| status_code: int | None |
| response_field: ModelField | None |
| stream_item_field: ModelField | None |
| dependencies: list[params.Depends] |
| description: str |
| response_fields: dict[int | str, ModelField] |
| dependant: Dependant |
| _flat_dependant: Dependant |
| _embed_body_fields: bool |
| body_field: ModelField | None |
| is_sse_stream: bool |
| is_json_stream: bool |
|
|
| def __init__( |
| self, |
| path: str, |
| endpoint: Callable[..., Any], |
| *, |
| response_model: Any = Default(None), |
| status_code: int | None = None, |
| tags: list[str | Enum] | None = None, |
| dependencies: Sequence[params.Depends] | None = None, |
| summary: str | None = None, |
| description: str | None = None, |
| response_description: str = "Successful Response", |
| responses: dict[int | str, dict[str, Any]] | None = None, |
| deprecated: bool | None = None, |
| name: str | None = None, |
| methods: set[str] | list[str] | None = None, |
| operation_id: str | None = None, |
| response_model_include: IncEx | None = None, |
| response_model_exclude: IncEx | None = 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] | DefaultPlaceholder = Default(JSONResponse), |
| dependency_overrides_provider: Any | None = None, |
| callbacks: list[BaseRoute] | None = None, |
| openapi_extra: dict[str, Any] | None = None, |
| generate_unique_id_function: Callable[["APIRoute"], str] |
| | DefaultPlaceholder = Default(generate_unique_id), |
| strict_content_type: bool | DefaultPlaceholder = Default(True), |
| ) -> None: |
| _populate_api_route_state( |
| cast(_APIRouteLike, self), |
| path, |
| 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, |
| name=name, |
| 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, |
| dependency_overrides_provider=dependency_overrides_provider, |
| callbacks=callbacks, |
| openapi_extra=openapi_extra, |
| generate_unique_id_function=generate_unique_id_function, |
| strict_content_type=strict_content_type, |
| ) |
| self.app = request_response(self.get_route_handler()) |
|
|
| def get_route_handler(self) -> Callable[[Request], Coroutine[Any, Any, Response]]: |
| route = cast(_APIRouteLike, self) |
| |
| |
| effective_context = _effective_route_context_var.get() |
| if effective_context is not None and effective_context.original_route is self: |
| route = cast(_APIRouteLike, effective_context) |
| return get_request_handler( |
| dependant=route.dependant, |
| body_field=route.body_field, |
| status_code=route.status_code, |
| response_class=route.response_class, |
| response_field=route.response_field, |
| response_model_include=route.response_model_include, |
| response_model_exclude=route.response_model_exclude, |
| response_model_by_alias=route.response_model_by_alias, |
| response_model_exclude_unset=route.response_model_exclude_unset, |
| response_model_exclude_defaults=route.response_model_exclude_defaults, |
| response_model_exclude_none=route.response_model_exclude_none, |
| dependency_overrides_provider=route.dependency_overrides_provider, |
| embed_body_fields=route._embed_body_fields, |
| strict_content_type=route.strict_content_type, |
| stream_item_field=route.stream_item_field, |
| is_json_stream=route.is_json_stream, |
| ) |
|
|
| def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| effective_context = _get_scope_effective_route_context(scope) |
| if effective_context is not None and effective_context.original_route is self: |
| match, child_scope = effective_context.matches(scope) |
| else: |
| match, child_scope = super().matches(scope) |
| if match != Match.NONE: |
| child_scope["route"] = self |
| return match, child_scope |
|
|
| async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: |
| effective_context = _get_scope_effective_route_context(scope) |
| if effective_context is not None and effective_context.original_route is self: |
| methods = effective_context.methods |
| if methods and scope["method"] not in methods: |
| headers = {"Allow": ", ".join(methods)} |
| if "app" in scope: |
| raise HTTPException(status_code=405, headers=headers) |
| response = PlainTextResponse( |
| "Method Not Allowed", status_code=405, headers=headers |
| ) |
| await response(scope, receive, send) |
| return |
| token = _effective_route_context_var.set(effective_context) |
| try: |
| app = request_response(self.get_route_handler()) |
| finally: |
| _effective_route_context_var.reset(token) |
| await app(scope, receive, send) |
| return |
| await super().handle(scope, receive, send) |
|
|
|
|
| @dataclass |
| class _RouterIncludeContext: |
| included_router: "APIRouter" |
| prefix: str = "" |
| tags: list[str | Enum] = field(default_factory=list) |
| dependencies: list[params.Depends] = field(default_factory=list) |
| default_response_class: type[Response] | DefaultPlaceholder = field( |
| default_factory=lambda: Default(JSONResponse) |
| ) |
| responses: dict[int | str, dict[str, Any]] = field(default_factory=dict) |
| callbacks: list[BaseRoute] = field(default_factory=list) |
| deprecated: bool | None = None |
| include_in_schema: bool = True |
| generate_unique_id_function: Callable[[APIRoute], str] | DefaultPlaceholder = field( |
| default_factory=lambda: Default(generate_unique_id) |
| ) |
| strict_content_type: bool | DefaultPlaceholder = field( |
| default_factory=lambda: Default(True) |
| ) |
| dependency_overrides_provider: Any | None = None |
|
|
| @classmethod |
| def for_include( |
| cls, |
| *, |
| parent_router: "APIRouter", |
| included_router: "APIRouter", |
| prefix: str = "", |
| tags: list[str | Enum] | None = None, |
| dependencies: Sequence[params.Depends] | None = None, |
| default_response_class: type[Response] | DefaultPlaceholder = Default( |
| JSONResponse |
| ), |
| responses: dict[int | str, dict[str, Any]] | None = None, |
| callbacks: list[BaseRoute] | None = None, |
| deprecated: bool | None = None, |
| include_in_schema: bool = True, |
| generate_unique_id_function: Callable[[APIRoute], str] |
| | DefaultPlaceholder = Default(generate_unique_id), |
| ) -> "_RouterIncludeContext": |
| return cls( |
| included_router=included_router, |
| prefix=parent_router.prefix + prefix, |
| tags=[*parent_router.tags, *(tags or [])], |
| dependencies=[*parent_router.dependencies, *(dependencies or [])], |
| default_response_class=get_value_or_default( |
| default_response_class, parent_router.default_response_class |
| ), |
| responses={**parent_router.responses, **(responses or {})}, |
| callbacks=[*parent_router.callbacks, *(callbacks or [])], |
| deprecated=deprecated or parent_router.deprecated, |
| include_in_schema=parent_router.include_in_schema and include_in_schema, |
| generate_unique_id_function=get_value_or_default( |
| generate_unique_id_function, parent_router.generate_unique_id_function |
| ), |
| strict_content_type=parent_router.strict_content_type, |
| dependency_overrides_provider=parent_router.dependency_overrides_provider, |
| ) |
|
|
| def combine( |
| self, child_context: "_RouterIncludeContext" |
| ) -> "_RouterIncludeContext": |
| return _RouterIncludeContext( |
| included_router=child_context.included_router, |
| prefix=self.prefix + child_context.prefix, |
| tags=[*self.tags, *child_context.tags], |
| dependencies=[*self.dependencies, *child_context.dependencies], |
| default_response_class=get_value_or_default( |
| child_context.default_response_class, self.default_response_class |
| ), |
| responses={**self.responses, **child_context.responses}, |
| callbacks=[*self.callbacks, *child_context.callbacks], |
| deprecated=self.deprecated or child_context.deprecated, |
| include_in_schema=self.include_in_schema |
| and child_context.include_in_schema, |
| generate_unique_id_function=get_value_or_default( |
| child_context.generate_unique_id_function, |
| self.generate_unique_id_function, |
| ), |
| strict_content_type=get_value_or_default( |
| child_context.strict_content_type, self.strict_content_type |
| ), |
| dependency_overrides_provider=self.dependency_overrides_provider, |
| ) |
|
|
| def path_for(self, route: _RouteWithPath) -> str: |
| return self.prefix + route.path |
|
|
|
|
| @dataclass |
| class _EffectiveRouteContext: |
| original_route: BaseRoute |
| starlette_route: BaseRoute | None = None |
| frontend_prefix: str = "" |
| path: str = "" |
| endpoint: Callable[..., Any] | None = None |
| stream_item_type: Any | None = None |
| response_model: Any = None |
| summary: str | None = None |
| response_description: str = "Successful Response" |
| deprecated: bool | None = None |
| operation_id: str | None = None |
| response_model_include: IncEx | None = None |
| response_model_exclude: IncEx | None = 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] | DefaultPlaceholder = field( |
| default_factory=lambda: Default(JSONResponse) |
| ) |
| dependency_overrides_provider: Any | None = None |
| callbacks: list[BaseRoute] | None = None |
| openapi_extra: dict[str, Any] | None = None |
| generate_unique_id_function: Callable[[Any], str] | DefaultPlaceholder = field( |
| default_factory=lambda: Default(generate_unique_id) |
| ) |
| strict_content_type: bool | DefaultPlaceholder = field( |
| default_factory=lambda: Default(True) |
| ) |
| tags: list[str | Enum] = field(default_factory=list) |
| responses: dict[int | str, dict[str, Any]] = field(default_factory=dict) |
| name: str = "" |
| path_regex: Any = None |
| path_format: str = "" |
| param_convertors: dict[str, Any] = field(default_factory=dict) |
| methods: set[str] = field(default_factory=set) |
| unique_id: str = "" |
| status_code: int | None = None |
| response_field: ModelField | None = None |
| stream_item_field: ModelField | None = None |
| dependencies: list[params.Depends] = field(default_factory=list) |
| description: str = "" |
| response_fields: dict[int | str, ModelField] = field(default_factory=dict) |
| dependant: Dependant | None = None |
| _flat_dependant: Dependant | None = None |
| _embed_body_fields: bool = False |
| body_field: ModelField | None = None |
| is_sse_stream: bool = False |
| is_json_stream: bool = False |
|
|
| @classmethod |
| def from_api_route( |
| cls, |
| *, |
| original_route: APIRoute, |
| include_context: _RouterIncludeContext, |
| ) -> "_EffectiveRouteContext": |
| route = cast(_APIRouteLike, original_route) |
| context = cls(original_route=original_route) |
| _populate_api_route_state( |
| cast(_APIRouteLike, context), |
| include_context.path_for(original_route), |
| route.endpoint, |
| response_model=route.response_model, |
| status_code=route.status_code, |
| tags=[*include_context.tags, *route.tags], |
| dependencies=[*include_context.dependencies, *route.dependencies], |
| summary=route.summary, |
| description=route.description, |
| response_description=route.response_description, |
| responses={**include_context.responses, **route.responses}, |
| deprecated=route.deprecated or include_context.deprecated, |
| methods=route.methods, |
| operation_id=route.operation_id, |
| response_model_include=route.response_model_include, |
| response_model_exclude=route.response_model_exclude, |
| response_model_by_alias=route.response_model_by_alias, |
| response_model_exclude_unset=route.response_model_exclude_unset, |
| response_model_exclude_defaults=route.response_model_exclude_defaults, |
| response_model_exclude_none=route.response_model_exclude_none, |
| include_in_schema=route.include_in_schema |
| and include_context.include_in_schema, |
| response_class=get_value_or_default( |
| route.response_class, |
| include_context.included_router.default_response_class, |
| include_context.default_response_class, |
| ), |
| name=route.name, |
| dependency_overrides_provider=include_context.dependency_overrides_provider, |
| callbacks=[*include_context.callbacks, *(route.callbacks or [])], |
| openapi_extra=route.openapi_extra, |
| generate_unique_id_function=get_value_or_default( |
| route.generate_unique_id_function, |
| include_context.included_router.generate_unique_id_function, |
| include_context.generate_unique_id_function, |
| ), |
| strict_content_type=get_value_or_default( |
| route.strict_content_type, |
| include_context.included_router.strict_content_type, |
| include_context.strict_content_type, |
| ), |
| ) |
| return context |
|
|
| @classmethod |
| def from_frontend_route_group( |
| cls, |
| *, |
| original_route: "_FrontendRouteGroup", |
| include_context: _RouterIncludeContext, |
| ) -> "_EffectiveRouteContext": |
| dependencies = [*include_context.dependencies, *original_route.dependencies] |
| context = cls( |
| original_route=original_route, |
| frontend_prefix=include_context.prefix, |
| dependencies=dependencies, |
| dependency_overrides_provider=include_context.dependency_overrides_provider, |
| ) |
| ( |
| context.dependant, |
| context._flat_dependant, |
| context._embed_body_fields, |
| ) = _build_dependant_with_parameterless_dependencies( |
| path="", |
| call=_frontend_dependency_endpoint, |
| dependencies=dependencies, |
| ) |
| return context |
|
|
| def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| if isinstance(self.original_route, _FrontendRouteGroup): |
| return self.original_route.matches_with_prefix(scope, self.frontend_prefix) |
| if not isinstance(self.original_route, APIRoute): |
| assert self.starlette_route is not None |
| return self.starlette_route.matches(scope) |
| if scope["type"] != "http": |
| return Match.NONE, {} |
| route_path = get_route_path(scope) |
| match = self.path_regex.match(route_path) |
| if not match: |
| return Match.NONE, {} |
| matched_params = match.groupdict() |
| for key, value in matched_params.items(): |
| matched_params[key] = self.param_convertors[key].convert(value) |
| path_params = dict(scope.get("path_params", {})) |
| path_params.update(matched_params) |
| child_scope = {"endpoint": self.endpoint, "path_params": path_params} |
| methods = self.methods |
| if methods and scope["method"] not in methods: |
| return Match.PARTIAL, child_scope |
| return Match.FULL, child_scope |
|
|
| def url_path_for(self, name: str, /, **path_params: Any) -> Any: |
| if not isinstance(self.original_route, APIRoute): |
| assert self.starlette_route is not None |
| return self.starlette_route.url_path_for(name, **path_params) |
| seen_params = set(path_params.keys()) |
| param_convertors = self.param_convertors |
| expected_params = set(param_convertors.keys()) |
| if name != self.name or seen_params != expected_params: |
| raise routing.NoMatchFound(name, path_params) |
| path, remaining_params = routing.replace_params( |
| self.path_format, param_convertors, path_params |
| ) |
| assert not remaining_params |
| return URLPath(path=path, protocol="http") |
|
|
|
|
| @dataclass(frozen=True) |
| class RouteContext: |
| route: BaseRoute |
| _route_context: _EffectiveRouteContext | None = field(default=None, repr=False) |
|
|
| @property |
| def original_route(self) -> BaseRoute: |
| if self._route_context is not None: |
| return self._route_context.original_route |
| return self.route |
|
|
| @property |
| def _effective_route(self) -> BaseRoute | _EffectiveRouteContext: |
| if self._route_context is not None: |
| return self._route_context |
| return self.route |
|
|
| @property |
| def path(self) -> str | None: |
| return getattr(self._effective_route, "path", None) |
|
|
| @property |
| def path_format(self) -> str | None: |
| return getattr(self._effective_route, "path_format", None) |
|
|
| @property |
| def name(self) -> str | None: |
| return getattr(self._effective_route, "name", None) |
|
|
| @property |
| def methods(self) -> set[str] | None: |
| return getattr(self._effective_route, "methods", None) |
|
|
| @property |
| def endpoint(self) -> Callable[..., Any] | None: |
| return getattr(self._effective_route, "endpoint", None) |
|
|
| def __getattr__(self, name: str) -> Any: |
| return getattr(self._effective_route, name) |
|
|
|
|
| @dataclass |
| class _IncludedRouter(BaseRoute): |
| original_router: "APIRouter" |
| include_context: _RouterIncludeContext |
| _effective_routes_lock: Any = field( |
| default_factory=threading.Lock, repr=False, compare=False |
| ) |
| _effective_candidates: list["_EffectiveRouteContext | _IncludedRouter"] = field( |
| default_factory=list |
| ) |
| _effective_candidates_version: int | None = None |
| _effective_low_priority_routes: list["_EffectiveRouteContext"] = field( |
| default_factory=list |
| ) |
| _effective_low_priority_routes_version: int | None = None |
|
|
| def effective_candidates(self) -> list["_EffectiveRouteContext | _IncludedRouter"]: |
| routes_version = self.original_router._get_routes_version() |
| if routes_version == self._effective_candidates_version: |
| return self._effective_candidates |
| with self._effective_routes_lock: |
| routes_version = self.original_router._get_routes_version() |
| if routes_version == self._effective_candidates_version: |
| return self._effective_candidates |
| effective_candidates: list[_EffectiveRouteContext | _IncludedRouter] = [] |
| for route in self.original_router.routes: |
| if isinstance(route, _IncludedRouter): |
| child_context = self.include_context.combine(route.include_context) |
| child_branch = _IncludedRouter( |
| original_router=route.original_router, |
| include_context=child_context, |
| ) |
| effective_candidates.append(child_branch) |
| continue |
| route_context = self._build_effective_context(route) |
| if route_context is not None: |
| effective_candidates.append(route_context) |
| self._effective_candidates = effective_candidates |
| self._effective_candidates_version = routes_version |
| return effective_candidates |
|
|
| def effective_low_priority_routes(self) -> list["_EffectiveRouteContext"]: |
| routes_version = self.original_router._get_routes_version() |
| if routes_version == self._effective_low_priority_routes_version: |
| return self._effective_low_priority_routes |
| with self._effective_routes_lock: |
| routes_version = self.original_router._get_routes_version() |
| if routes_version == self._effective_low_priority_routes_version: |
| return self._effective_low_priority_routes |
| effective_low_priority_routes: list[_EffectiveRouteContext] = [] |
| for route in self.original_router._low_priority_routes: |
| route_context = self._build_effective_context(route) |
| if route_context is not None: |
| effective_low_priority_routes.append(route_context) |
| for route in self.original_router.routes: |
| if isinstance(route, _IncludedRouter): |
| child_context = self.include_context.combine(route.include_context) |
| child_branch = _IncludedRouter( |
| original_router=route.original_router, |
| include_context=child_context, |
| ) |
| effective_low_priority_routes.extend( |
| child_branch.effective_low_priority_routes() |
| ) |
| self._effective_low_priority_routes = effective_low_priority_routes |
| self._effective_low_priority_routes_version = routes_version |
| return effective_low_priority_routes |
|
|
| def _build_effective_context( |
| self, route: BaseRoute |
| ) -> _EffectiveRouteContext | None: |
| if isinstance(route, APIRoute): |
| return _EffectiveRouteContext.from_api_route( |
| original_route=route, |
| include_context=self.include_context, |
| ) |
| if isinstance(route, _FrontendRouteGroup): |
| return _EffectiveRouteContext.from_frontend_route_group( |
| original_route=route, |
| include_context=self.include_context, |
| ) |
| if isinstance(route, routing.Route): |
| starlette_route: BaseRoute = routing.Route( |
| self.include_context.path_for(route), |
| endpoint=route.endpoint, |
| methods=list(route.methods or []), |
| name=route.name, |
| include_in_schema=route.include_in_schema, |
| ) |
| return _EffectiveRouteContext( |
| original_route=route, |
| starlette_route=starlette_route, |
| ) |
| if isinstance(route, APIWebSocketRoute): |
| starlette_route = APIWebSocketRoute( |
| self.include_context.path_for(route), |
| endpoint=route.endpoint, |
| name=route.name, |
| dependencies=[*self.include_context.dependencies, *route.dependencies], |
| dependency_overrides_provider=( |
| self.include_context.dependency_overrides_provider |
| ), |
| ) |
| return _EffectiveRouteContext( |
| original_route=route, |
| starlette_route=starlette_route, |
| ) |
| if isinstance(route, routing.WebSocketRoute): |
| starlette_route = routing.WebSocketRoute( |
| self.include_context.path_for(route), route.endpoint, name=route.name |
| ) |
| return _EffectiveRouteContext( |
| original_route=route, |
| starlette_route=starlette_route, |
| ) |
| if isinstance(route, routing.Mount): |
| starlette_route = copy.copy(route) |
| starlette_route.path = self.include_context.path_for(route).rstrip("/") |
| ( |
| starlette_route.path_regex, |
| starlette_route.path_format, |
| starlette_route.param_convertors, |
| ) = compile_path(starlette_route.path + "/{path:path}") |
| return _EffectiveRouteContext( |
| original_route=route, |
| starlette_route=starlette_route, |
| ) |
| if isinstance(route, routing.Host): |
| if self.include_context.prefix: |
| prefixed_app: ASGIApp = routing.Router( |
| routes=[routing.Mount(self.include_context.prefix, app=route.app)] |
| ) |
| else: |
| prefixed_app = route.app |
| starlette_route = routing.Host( |
| route.host, app=prefixed_app, name=route.name |
| ) |
| return _EffectiveRouteContext( |
| original_route=route, |
| starlette_route=starlette_route, |
| ) |
| return None |
|
|
| def _match( |
| self, scope: Scope |
| ) -> tuple[Match, Scope, BaseRoute | None, _EffectiveRouteContext | None]: |
| partial: tuple[Scope, BaseRoute, _EffectiveRouteContext | None] | None = None |
| for candidate in self.effective_candidates(): |
| if isinstance(candidate, _IncludedRouter): |
| match, child_scope = candidate.matches(scope) |
| route: BaseRoute = candidate |
| route_context = None |
| elif isinstance(candidate.original_route, APIRoute): |
| route_context = candidate |
| fastapi_scope = _get_fastapi_scope(scope) |
| previous_context = fastapi_scope.get( |
| _FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY, _SCOPE_MISSING |
| ) |
| fastapi_scope[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = route_context |
| try: |
| match, child_scope = candidate.original_route.matches(scope) |
| finally: |
| _restore_fastapi_scope_key( |
| scope, _FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY, previous_context |
| ) |
| route = candidate.original_route |
| else: |
| route_context = candidate |
| match, child_scope = candidate.matches(scope) |
| route = candidate.starlette_route or candidate.original_route |
| if match == Match.FULL: |
| return match, child_scope, route, route_context |
| if match == Match.PARTIAL and partial is None: |
| partial = (child_scope, route, route_context) |
| if partial is not None: |
| child_scope, route, route_context = partial |
| return Match.PARTIAL, child_scope, route, route_context |
| return Match.NONE, {}, None, None |
|
|
| def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| fastapi_scope = _get_fastapi_scope(scope) |
| previous_router = fastapi_scope.get( |
| _FASTAPI_INCLUDED_ROUTER_KEY, _SCOPE_MISSING |
| ) |
| fastapi_scope[_FASTAPI_INCLUDED_ROUTER_KEY] = self |
| try: |
| match, _ = self.original_router.matches(scope) |
| return match, {} |
| finally: |
| _restore_fastapi_scope_key( |
| scope, _FASTAPI_INCLUDED_ROUTER_KEY, previous_router |
| ) |
|
|
| async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: |
| _get_fastapi_scope(scope)[_FASTAPI_INCLUDED_ROUTER_KEY] = self |
| await self.original_router.handle(scope, receive, send) |
|
|
| async def _handle_selected( |
| self, scope: Scope, receive: Receive, send: Send |
| ) -> None: |
| match, child_scope, route, effective_context = self._match(scope) |
| if match == Match.NONE or route is None: |
| await self.original_router.default(scope, receive, send) |
| return |
| scope.update(child_scope) |
| if isinstance(route, _IncludedRouter): |
| await route.handle(scope, receive, send) |
| return |
| if effective_context is not None: |
| _get_fastapi_scope(scope)[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = ( |
| effective_context |
| ) |
| original_route = effective_context.original_route |
| if isinstance(original_route, APIRoute): |
| scope["route"] = original_route |
| await original_route.handle(scope, receive, send) |
| return |
| await route.handle(scope, receive, send) |
|
|
| def effective_route_contexts(self) -> Iterator[_EffectiveRouteContext]: |
| for candidate in self.effective_candidates(): |
| if isinstance(candidate, _IncludedRouter): |
| yield from candidate.effective_route_contexts() |
| else: |
| yield candidate |
|
|
| def url_path_for(self, name: str, /, **path_params: Any) -> Any: |
| for route_context in self.effective_route_contexts(): |
| try: |
| return route_context.url_path_for(name, **path_params) |
| except routing.NoMatchFound: |
| pass |
| raise routing.NoMatchFound(name, path_params) |
|
|
|
|
| def _iter_included_route_candidates(routes: Sequence[BaseRoute]) -> Iterator[BaseRoute]: |
| for route, route_context in _iter_routes_with_context(routes): |
| if route_context is not None and route_context.starlette_route is not None: |
| yield route_context.starlette_route |
| else: |
| yield route |
|
|
|
|
| def iter_route_contexts( |
| routes: Sequence[BaseRoute | RouteContext], |
| ) -> Iterator[RouteContext]: |
| for route in routes: |
| if isinstance(route, RouteContext): |
| yield route |
| continue |
| for original_route, route_context in _iter_routes_with_context([route]): |
| if route_context is None: |
| yield RouteContext(original_route) |
| else: |
| yield RouteContext(original_route, route_context) |
|
|
|
|
| def _iter_routes_with_context( |
| routes: Sequence[BaseRoute], |
| ) -> Iterator[tuple[BaseRoute, _EffectiveRouteContext | None]]: |
| for route in routes: |
| if isinstance(route, _IncludedRouter): |
| for route_context in route.effective_route_contexts(): |
| yield route_context.original_route, route_context |
| else: |
| yield route, None |
|
|
|
|
| def _normalize_frontend_path(path: str) -> str: |
| if not path: |
| raise AssertionError("A frontend path cannot be empty") |
| if not path.startswith("/"): |
| raise AssertionError("A frontend path must start with '/'") |
| if path != "/": |
| path = path.rstrip("/") |
| return path |
|
|
|
|
| def _join_frontend_paths(prefix: str, path: str) -> str: |
| if not prefix: |
| return path |
| if path == "/": |
| return prefix |
| return prefix + path |
|
|
|
|
| def _frontend_path_specificity(path: str) -> int: |
| if path == "/": |
| return 0 |
| return len(path) |
|
|
|
|
| def _get_resolved_absolute_path(path: str | os.PathLike[str]) -> str: |
| return os.path.realpath(os.fspath(path)) |
|
|
|
|
| class _FrontendStaticFiles(StaticFiles): |
| def __init__( |
| self, |
| *, |
| directory: str | os.PathLike[str], |
| fallback: Literal["auto", "index.html", "404.html"] | None, |
| check_dir: bool = True, |
| ) -> None: |
| self.fallback = fallback |
| if check_dir and not os.path.isdir(directory): |
| raise RuntimeError( |
| f"Frontend directory '{directory}' does not exist. " |
| f"Resolved absolute path: '{_get_resolved_absolute_path(directory)}'" |
| ) |
| super().__init__( |
| directory=directory, |
| html=True, |
| check_dir=check_dir, |
| follow_symlink=False, |
| ) |
| if check_dir and fallback in {"index.html", "404.html"}: |
| self._check_fallback_file(fallback) |
|
|
| def _check_fallback_file(self, fallback: str) -> None: |
| _, stat_result = self.lookup_path(fallback) |
| if stat_result is None or not stat.S_ISREG(stat_result.st_mode): |
| raise RuntimeError( |
| f"Frontend fallback file '{fallback}' does not exist in " |
| f"directory '{self.directory}'. Resolved absolute directory: " |
| f"'{self._get_resolved_directory()}'" |
| ) |
|
|
| def _get_resolved_directory(self) -> str: |
| assert self.directory is not None |
| return _get_resolved_absolute_path(self.directory) |
|
|
| def get_path(self, scope: Scope) -> str: |
| path = _get_fastapi_scope(scope).get(_FASTAPI_FRONTEND_PATH_KEY, "") |
| assert isinstance(path, str) |
| return os.path.normpath(os.path.join(*path.split("/"))) |
|
|
| async def get_response(self, path: str, scope: Scope) -> Response: |
| if scope["method"] not in ("GET", "HEAD"): |
| if await self._lookup_static_resource(path) is not None: |
| raise HTTPException(status_code=405) |
| raise HTTPException(status_code=404) |
|
|
| static_resource = await self._lookup_static_resource(path) |
| if static_resource is not None: |
| full_path, stat_result, is_directory_index = static_resource |
| if is_directory_index and not scope["path"].endswith("/"): |
| url = URL(scope=scope) |
| url = url.replace(path=url.path + "/") |
| return RedirectResponse(url=url) |
| return self.file_response(full_path, stat_result, scope) |
|
|
| if self.fallback == "404.html" or ( |
| self.fallback == "auto" and self._fallback_file_exists("404.html") |
| ): |
| return await self._fallback_response("404.html", scope, status_code=404) |
|
|
| if ( |
| self.fallback == "index.html" |
| or (self.fallback == "auto" and self._fallback_file_exists("index.html")) |
| ) and _is_frontend_navigation_request(scope): |
| return await self._fallback_response("index.html", scope, status_code=200) |
|
|
| raise HTTPException(status_code=404) |
|
|
| async def _lookup_path(self, path: str) -> tuple[str, os.stat_result | None]: |
| try: |
| return await run_in_threadpool(self.lookup_path, path) |
| except PermissionError: |
| raise HTTPException(status_code=401) from None |
| except OSError as exc: |
| if exc.errno == errno.ENAMETOOLONG: |
| raise HTTPException(status_code=404) from None |
| raise exc |
| except ValueError: |
| raise HTTPException(status_code=404) from None |
|
|
| async def _lookup_static_resource( |
| self, path: str |
| ) -> tuple[str, os.stat_result, bool] | None: |
| full_path, stat_result = await self._lookup_path(path) |
| if stat_result is None: |
| return None |
| if stat.S_ISREG(stat_result.st_mode): |
| return full_path, stat_result, False |
| if stat.S_ISDIR(stat_result.st_mode): |
| index_path = os.path.join(path, "index.html") |
| full_path, stat_result = await self._lookup_path(index_path) |
| if stat_result is not None and stat.S_ISREG(stat_result.st_mode): |
| return full_path, stat_result, True |
| return None |
|
|
| def _fallback_file_exists(self, fallback: str) -> bool: |
| _, stat_result = self.lookup_path(fallback) |
| return stat_result is not None and stat.S_ISREG(stat_result.st_mode) |
|
|
| async def _fallback_response( |
| self, fallback: str, scope: Scope, *, status_code: int |
| ) -> Response: |
| full_path, stat_result = await run_in_threadpool(self.lookup_path, fallback) |
| if stat_result is None or not stat.S_ISREG(stat_result.st_mode): |
| raise RuntimeError( |
| f"Frontend fallback file '{fallback}' does not exist in " |
| f"directory '{self.directory}'. Resolved absolute directory: " |
| f"'{self._get_resolved_directory()}'" |
| ) |
| return self.file_response( |
| full_path, stat_result, scope, status_code=status_code |
| ) |
|
|
|
|
| def _iter_accept_media_types(accept: str) -> Iterator[tuple[str, float]]: |
| for raw_value in accept.split(","): |
| message = email.message.Message() |
| message["content-type"] = raw_value.strip() |
| q = message.get_param("q") |
| quality = 1.0 |
| if isinstance(q, str): |
| try: |
| quality = float(q) |
| except ValueError: |
| pass |
| yield ( |
| f"{message.get_content_maintype()}/{message.get_content_subtype()}", |
| quality, |
| ) |
|
|
|
|
| def _is_frontend_navigation_request(scope: Scope) -> bool: |
| request = Request(scope) |
| for media_type, quality in _iter_accept_media_types( |
| request.headers.get("accept", "") |
| ): |
| if media_type in {"text/html", "application/xhtml+xml"} and quality != 0: |
| return True |
| return False |
|
|
|
|
| class _FrontendRoute(BaseRoute): |
| def __init__( |
| self, |
| path: str, |
| *, |
| directory: str | os.PathLike[str], |
| fallback: Literal["auto", "index.html", "404.html"] | None = "auto", |
| check_dir: bool = True, |
| ) -> None: |
| if fallback not in {"auto", "index.html", "404.html", None}: |
| raise AssertionError( |
| "fallback must be 'auto', 'index.html', '404.html', or None" |
| ) |
| self.path = _normalize_frontend_path(path) |
| self.methods = {"GET", "HEAD"} |
| self.app = _FrontendStaticFiles( |
| directory=directory, fallback=fallback, check_dir=check_dir |
| ) |
|
|
| def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| return self.matches_with_path(scope, self.path) |
|
|
| def matches_with_path(self, scope: Scope, path: str) -> tuple[Match, Scope]: |
| if scope["type"] != "http": |
| return Match.NONE, {} |
| frontend_path = self._get_frontend_path(path, get_route_path(scope)) |
| if frontend_path is None: |
| return Match.NONE, {} |
| child_scope = { |
| _FASTAPI_SCOPE_KEY: { |
| _FASTAPI_FRONTEND_PATH_KEY: frontend_path, |
| _FASTAPI_FRONTEND_SPECIFICITY_KEY: _frontend_path_specificity(path), |
| } |
| } |
| if scope["method"] not in self.methods: |
| return Match.PARTIAL, child_scope |
| return Match.FULL, child_scope |
|
|
| def _get_frontend_path(self, path: str, route_path: str) -> str | None: |
| if path == "/": |
| return route_path.lstrip("/") |
| if route_path == path: |
| return "" |
| prefix = path + "/" |
| if route_path.startswith(prefix): |
| return route_path[len(prefix) :] |
| return None |
|
|
| async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: |
| await self.app(scope, receive, send) |
|
|
| def url_path_for(self, name: str, /, **path_params: Any) -> URLPath: |
| raise NoMatchFound(name, path_params) |
|
|
|
|
| class _FrontendRouteGroup(BaseRoute): |
| def __init__( |
| self, |
| *, |
| dependencies: Sequence[params.Depends] | None = None, |
| dependency_overrides_provider: Any | None = None, |
| ) -> None: |
| self.routes: list[_FrontendRoute] = [] |
| self.dependencies = list(dependencies or []) |
| self.dependency_overrides_provider = dependency_overrides_provider |
| ( |
| self.dependant, |
| self._flat_dependant, |
| self._embed_body_fields, |
| ) = _build_dependant_with_parameterless_dependencies( |
| path="", |
| call=_frontend_dependency_endpoint, |
| dependencies=self.dependencies, |
| ) |
|
|
| def add_frontend_route( |
| self, |
| path: str, |
| *, |
| directory: str | os.PathLike[str], |
| fallback: Literal["auto", "index.html", "404.html"] | None = "auto", |
| check_dir: bool = True, |
| ) -> None: |
| self.routes.append( |
| _FrontendRoute( |
| path, |
| directory=directory, |
| fallback=fallback, |
| check_dir=check_dir, |
| ) |
| ) |
|
|
| def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| match, child_scope, _ = self._match(scope, prefix="") |
| return match, child_scope |
|
|
| def matches_with_prefix(self, scope: Scope, prefix: str) -> tuple[Match, Scope]: |
| match, child_scope, _ = self._match(scope, prefix=prefix) |
| return match, child_scope |
|
|
| def _match( |
| self, scope: Scope, *, prefix: str |
| ) -> tuple[Match, Scope, _FrontendRoute | None]: |
| full: tuple[Scope, _FrontendRoute, int] | None = None |
| partial: tuple[Scope, _FrontendRoute, int] | None = None |
| for route in self.routes: |
| path = _join_frontend_paths(prefix, route.path) |
| match, child_scope = route.matches_with_path(scope, path) |
| specificity = _frontend_path_specificity(path) |
| if match == Match.FULL: |
| if full is None or specificity > full[2]: |
| full = (child_scope, route, specificity) |
| elif match == Match.PARTIAL: |
| if partial is None or specificity > partial[2]: |
| partial = (child_scope, route, specificity) |
| if full is not None: |
| child_scope, route, _ = full |
| return Match.FULL, child_scope, route |
| if partial is not None: |
| child_scope, route, _ = partial |
| return Match.PARTIAL, child_scope, route |
| return Match.NONE, {}, None |
|
|
| async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: |
| effective_context = _get_scope_effective_route_context(scope) |
| if ( |
| isinstance(effective_context, _EffectiveRouteContext) |
| and effective_context.original_route is self |
| ): |
| prefix = effective_context.frontend_prefix |
| dependant = effective_context.dependant |
| dependency_overrides_provider = ( |
| effective_context.dependency_overrides_provider |
| ) |
| embed_body_fields = effective_context._embed_body_fields |
| else: |
| prefix = "" |
| dependant = self.dependant |
| dependency_overrides_provider = self.dependency_overrides_provider |
| embed_body_fields = self._embed_body_fields |
| match, child_scope, route = self._match(scope, prefix=prefix) |
| if match == Match.NONE or route is None: |
| raise HTTPException(status_code=404) |
| _update_scope(scope, child_scope) |
| if match == Match.FULL and dependant and dependant.dependencies: |
| async with self._solve_dependencies( |
| scope, |
| receive, |
| send, |
| dependant=dependant, |
| dependency_overrides_provider=dependency_overrides_provider, |
| embed_body_fields=embed_body_fields, |
| ): |
| await route.handle(scope, receive, send) |
| return |
| await route.handle(scope, receive, send) |
|
|
| def url_path_for(self, name: str, /, **path_params: Any) -> URLPath: |
| raise NoMatchFound(name, path_params) |
|
|
| |
| |
| |
| @asynccontextmanager |
| async def _solve_dependencies( |
| self, |
| scope: Scope, |
| receive: Receive, |
| send: Send, |
| *, |
| dependant: Dependant, |
| dependency_overrides_provider: Any | None, |
| embed_body_fields: bool, |
| ) -> AsyncIterator[None]: |
| request = Request(scope, receive, send) |
| previous_inner_astack = scope.get("fastapi_inner_astack", _SCOPE_MISSING) |
| previous_function_astack = scope.get("fastapi_function_astack", _SCOPE_MISSING) |
| try: |
| async with AsyncExitStack() as request_stack: |
| scope["fastapi_inner_astack"] = request_stack |
| async with AsyncExitStack() as function_stack: |
| scope["fastapi_function_astack"] = function_stack |
| solved_result = await solve_dependencies( |
| request=request, |
| dependant=dependant, |
| dependency_overrides_provider=dependency_overrides_provider, |
| async_exit_stack=request_stack, |
| embed_body_fields=embed_body_fields, |
| ) |
| if solved_result.errors: |
| raise RequestValidationError(solved_result.errors) |
| yield |
| finally: |
| if previous_inner_astack is _SCOPE_MISSING: |
| scope.pop("fastapi_inner_astack", None) |
| else: |
| scope["fastapi_inner_astack"] = previous_inner_astack |
| if previous_function_astack is _SCOPE_MISSING: |
| scope.pop("fastapi_function_astack", None) |
| else: |
| scope["fastapi_function_astack"] = previous_function_astack |
|
|
|
|
| class APIRouter(routing.Router): |
| """ |
| `APIRouter` class, used to group *path operations*, for example to structure |
| an app in multiple files. It would then be included in the `FastAPI` app, or |
| in another `APIRouter` (ultimately included in the app). |
| |
| Read more about it in the |
| [FastAPI docs for Bigger Applications - Multiple Files](https://fastapi.tiangolo.com/tutorial/bigger-applications/). |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| |
| @router.get("/users/", tags=["users"]) |
| async def read_users(): |
| return [{"username": "Rick"}, {"username": "Morty"}] |
| |
| |
| app.include_router(router) |
| ``` |
| """ |
|
|
| def __init__( |
| self, |
| *, |
| prefix: Annotated[str, Doc("An optional path prefix for the router.")] = "", |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to all the *path operations* in this |
| router. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to all the |
| *path operations* in this router. |
| |
| Read more about it in the |
| [FastAPI docs for Bigger Applications - Multiple Files](https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies). |
| """ |
| ), |
| ] = None, |
| default_response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| The default response class to be used. |
| |
| Read more in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#default-response-class). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses to be shown in OpenAPI. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Additional Responses in OpenAPI](https://fastapi.tiangolo.com/advanced/additional-responses/). |
| |
| And in the |
| [FastAPI docs for Bigger Applications](https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies). |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| OpenAPI callbacks that should apply to all *path operations* in this |
| router. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| routes: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| **Note**: you probably shouldn't use this parameter, it is inherited |
| from Starlette and supported for compatibility. |
| |
| --- |
| |
| A list of routes to serve incoming HTTP and WebSocket requests. |
| """ |
| ), |
| deprecated( |
| """ |
| You normally wouldn't use this parameter with FastAPI, it is inherited |
| from Starlette and supported for compatibility. |
| |
| In FastAPI, you normally would use the *path operation methods*, |
| like `router.get()`, `router.post()`, etc. |
| """ |
| ), |
| ] = None, |
| redirect_slashes: Annotated[ |
| bool, |
| Doc( |
| """ |
| Whether to detect and redirect slashes in URLs when the client doesn't |
| use the same format. |
| """ |
| ), |
| ] = True, |
| default: Annotated[ |
| ASGIApp | None, |
| Doc( |
| """ |
| Default function handler for this router. Used to handle |
| 404 Not Found errors. |
| """ |
| ), |
| ] = None, |
| dependency_overrides_provider: Annotated[ |
| Any | None, |
| Doc( |
| """ |
| Only used internally by FastAPI to handle dependency overrides. |
| |
| You shouldn't need to use it. It normally points to the `FastAPI` app |
| object. |
| """ |
| ), |
| ] = None, |
| route_class: Annotated[ |
| type[APIRoute], |
| Doc( |
| """ |
| Custom route (*path operation*) class to be used by this router. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Request and APIRoute class](https://fastapi.tiangolo.com/how-to/custom-request-and-route/#custom-apiroute-class-in-a-router). |
| """ |
| ), |
| ] = APIRoute, |
| on_startup: Annotated[ |
| Sequence[Callable[[], Any]] | None, |
| Doc( |
| """ |
| A list of startup event handler functions. |
| |
| You should instead use the `lifespan` handlers. |
| |
| Read more in the [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/). |
| """ |
| ), |
| ] = None, |
| on_shutdown: Annotated[ |
| Sequence[Callable[[], Any]] | None, |
| Doc( |
| """ |
| A list of shutdown event handler functions. |
| |
| You should instead use the `lifespan` handlers. |
| |
| Read more in the |
| [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/). |
| """ |
| ), |
| ] = None, |
| |
| |
| lifespan: Annotated[ |
| Lifespan[Any] | None, |
| Doc( |
| """ |
| A `Lifespan` context manager handler. This replaces `startup` and |
| `shutdown` functions with a single context manager. |
| |
| Read more in the |
| [FastAPI docs for `lifespan`](https://fastapi.tiangolo.com/advanced/events/). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark all *path operations* in this router as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| To include (or not) all the *path operations* in this router in the |
| generated OpenAPI. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| strict_content_type: Annotated[ |
| bool, |
| Doc( |
| """ |
| Enable strict checking for request Content-Type headers. |
| |
| When `True` (the default), requests with a body that do not include |
| a `Content-Type` header will **not** be parsed as JSON. |
| |
| This prevents potential cross-site request forgery (CSRF) attacks |
| that exploit the browser's ability to send requests without a |
| Content-Type header, bypassing CORS preflight checks. In particular |
| applicable for apps that need to be run locally (in localhost). |
| |
| When `False`, requests without a `Content-Type` header will have |
| their body parsed as JSON, which maintains compatibility with |
| certain clients that don't send `Content-Type` headers. |
| |
| Read more about it in the |
| [FastAPI docs for Strict Content-Type](https://fastapi.tiangolo.com/advanced/strict-content-type/). |
| """ |
| ), |
| ] = Default(True), |
| ) -> None: |
| |
| if lifespan is None: |
| |
| lifespan_context: Lifespan[Any] = _DefaultLifespan(self) |
| elif inspect.isasyncgenfunction(lifespan): |
| lifespan_context = asynccontextmanager(lifespan) |
| elif inspect.isgeneratorfunction(lifespan): |
| lifespan_context = _wrap_gen_lifespan_context(lifespan) |
| else: |
| lifespan_context = lifespan |
| self.lifespan_context = lifespan_context |
|
|
| super().__init__( |
| routes=routes, |
| redirect_slashes=redirect_slashes, |
| default=default, |
| lifespan=lifespan_context, |
| ) |
| if prefix: |
| assert prefix.startswith("/"), "A path prefix must start with '/'" |
| assert not prefix.endswith("/"), ( |
| "A path prefix must not end with '/', as the routes will start with '/'" |
| ) |
|
|
| |
| |
| |
| self.on_startup: list[Callable[[], Any]] = ( |
| [] if on_startup is None else list(on_startup) |
| ) |
| self.on_shutdown: list[Callable[[], Any]] = ( |
| [] if on_shutdown is None else list(on_shutdown) |
| ) |
|
|
| self.prefix = prefix |
| self.tags: list[str | Enum] = tags or [] |
| self.dependencies = list(dependencies or []) |
| self.deprecated = deprecated |
| self.include_in_schema = include_in_schema |
| self.responses = responses or {} |
| self.callbacks = callbacks or [] |
| self.dependency_overrides_provider = dependency_overrides_provider |
| self.route_class = route_class |
| self.default_response_class = default_response_class |
| self.generate_unique_id_function = generate_unique_id_function |
| self.strict_content_type = strict_content_type |
| self._routes_version = 0 |
| self._low_priority_routes: list[BaseRoute] = [] |
| self._frontend_routes: _FrontendRouteGroup | None = None |
|
|
| def _mark_routes_changed(self) -> None: |
| self._routes_version += 1 |
|
|
| def _get_routes_version(self, seen: set[int] | None = None) -> int: |
| if seen is None: |
| seen = set() |
| router_id = id(self) |
| if router_id in seen: |
| return self._routes_version |
| seen.add(router_id) |
| version = self._routes_version |
| for route in self.routes: |
| if isinstance(route, _IncludedRouter): |
| version += route.original_router._get_routes_version(seen) |
| return version |
|
|
| def _contains_router( |
| self, router: "APIRouter", seen: set[int] | None = None |
| ) -> bool: |
| if seen is None: |
| seen = set() |
| router_id = id(self) |
| if router_id in seen: |
| return False |
| seen.add(router_id) |
| for route in self.routes: |
| if not isinstance(route, _IncludedRouter): |
| continue |
| if route.original_router is router: |
| return True |
| if route.original_router._contains_router(router, seen): |
| return True |
| return False |
|
|
| def add_route( |
| self, |
| path: str, |
| endpoint: Callable[[Request], Awaitable[Response] | Response], |
| methods: Collection[str] | None = None, |
| name: str | None = None, |
| include_in_schema: bool = True, |
| ) -> None: |
| super().add_route( |
| path, |
| endpoint, |
| methods=methods, |
| name=name, |
| include_in_schema=include_in_schema, |
| ) |
| self._mark_routes_changed() |
|
|
| def add_websocket_route( |
| self, |
| path: str, |
| endpoint: Callable[[WebSocket], Awaitable[None]], |
| name: str | None = None, |
| ) -> None: |
| super().add_websocket_route(path, endpoint, name=name) |
| self._mark_routes_changed() |
|
|
| def frontend( |
| self, |
| path: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path prefix where the frontend build should be served. |
| """ |
| ), |
| ], |
| *, |
| directory: Annotated[ |
| str | os.PathLike[str], |
| Doc( |
| """ |
| The directory containing the static frontend build output. |
| """ |
| ), |
| ], |
| fallback: Annotated[ |
| Literal["auto", "index.html", "404.html"] | None, |
| Doc( |
| """ |
| The fallback file behavior for missing frontend paths. |
| """ |
| ), |
| ] = "auto", |
| check_dir: Annotated[ |
| bool, |
| Doc( |
| """ |
| Check that the frontend directory exists when the app is created. |
| """ |
| ), |
| ] = True, |
| ) -> None: |
| """ |
| Serve a static frontend build as low-priority routes. |
| |
| Use this for frontend tools that build static files into a directory, |
| such as `dist`. **FastAPI** path operations are checked first, and |
| the frontend files are checked only if no normal route matched. |
| |
| A typical project could look like this: |
| |
| ```text |
| . |
| ├── pyproject.toml |
| ├── app |
| │ ├── __init__.py |
| │ └── main.py |
| └── dist |
| ├── index.html |
| └── assets |
| └── app.js |
| ``` |
| |
| Then in `app/main.py`: |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| |
| app = FastAPI() |
| router = APIRouter() |
| router.frontend("/", directory="dist") |
| app.include_router(router) |
| ``` |
| """ |
| normalized_path = _normalize_frontend_path(path) |
| if self._frontend_routes is None: |
| self._frontend_routes = _FrontendRouteGroup( |
| dependencies=self.dependencies, |
| dependency_overrides_provider=self.dependency_overrides_provider, |
| ) |
| self._low_priority_routes.append(self._frontend_routes) |
| self._frontend_routes.add_frontend_route( |
| _join_frontend_paths(self.prefix, normalized_path), |
| directory=directory, |
| fallback=fallback, |
| check_dir=check_dir, |
| ) |
| self._mark_routes_changed() |
|
|
| async def app(self, scope: Scope, receive: Receive, send: Send) -> None: |
| assert scope["type"] in ("http", "websocket", "lifespan") |
|
|
| if "router" not in scope: |
| scope["router"] = self |
|
|
| if scope["type"] == "lifespan": |
| await self.lifespan(scope, receive, send) |
| return |
|
|
| partial: tuple[BaseRoute, Scope] | None = None |
| for route in self.routes: |
| match, child_scope = route.matches(scope) |
| if match == Match.FULL: |
| scope.update(child_scope) |
| await route.handle(scope, receive, send) |
| return |
| if match == Match.PARTIAL and partial is None: |
| partial = (route, child_scope) |
|
|
| if partial is not None: |
| route, child_scope = partial |
| scope.update(child_scope) |
| await route.handle(scope, receive, send) |
| return |
|
|
| route_path = get_route_path(scope) |
| if scope["type"] == "http" and self.redirect_slashes and route_path != "/": |
| redirect_scope = dict(scope) |
| if route_path.endswith("/"): |
| redirect_scope["path"] = redirect_scope["path"].rstrip("/") |
| else: |
| redirect_scope["path"] = redirect_scope["path"] + "/" |
|
|
| for route in self.routes: |
| match, _ = route.matches(redirect_scope) |
| if match != Match.NONE: |
| redirect_url = URL(scope=redirect_scope) |
| response = RedirectResponse(url=str(redirect_url)) |
| await response(scope, receive, send) |
| return |
|
|
| ( |
| low_priority_match, |
| low_priority_scope, |
| low_priority_route, |
| low_priority_context, |
| ) = self._match_low_priority(scope) |
| if low_priority_match != Match.NONE and low_priority_route is not None: |
| _update_scope(scope, low_priority_scope) |
| if low_priority_context is not None: |
| _get_fastapi_scope(scope)[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = ( |
| low_priority_context |
| ) |
| original_route = low_priority_context.original_route |
| if isinstance(original_route, APIRoute): |
| scope["route"] = original_route |
| await original_route.handle(scope, receive, send) |
| return |
| await low_priority_route.handle(scope, receive, send) |
| return |
|
|
| await self.default(scope, receive, send) |
|
|
| async def handle(self, scope: Scope, receive: Receive, send: Send) -> None: |
| included_router = _get_scope_included_router(scope) |
| if ( |
| isinstance(included_router, _IncludedRouter) |
| and included_router.original_router is self |
| ): |
| await included_router._handle_selected(scope, receive, send) |
| return |
| await self.app(scope, receive, send) |
|
|
| def matches(self, scope: Scope) -> tuple[Match, Scope]: |
| included_router = _get_scope_included_router(scope) |
| if ( |
| isinstance(included_router, _IncludedRouter) |
| and included_router.original_router is self |
| ): |
| match, child_scope, _, _ = included_router._match(scope) |
| return match, child_scope |
| return Match.NONE, {} |
|
|
| def _iter_low_priority_routes( |
| self, |
| ) -> Iterator[BaseRoute | _EffectiveRouteContext]: |
| yield from self._low_priority_routes |
| for route in self.routes: |
| if isinstance(route, _IncludedRouter): |
| yield from route.effective_low_priority_routes() |
|
|
| def _match_low_priority( |
| self, scope: Scope |
| ) -> tuple[Match, Scope, BaseRoute | None, _EffectiveRouteContext | None]: |
| full: tuple[Scope, BaseRoute, _EffectiveRouteContext | None] | None = None |
| partial: tuple[Scope, BaseRoute, _EffectiveRouteContext | None] | None = None |
| for candidate in self._iter_low_priority_routes(): |
| route: BaseRoute |
| if isinstance(candidate, _EffectiveRouteContext): |
| route_context: _EffectiveRouteContext | None = candidate |
| original_route = candidate.original_route |
| if isinstance(original_route, APIRoute): |
| fastapi_scope = _get_fastapi_scope(scope) |
| previous_context = fastapi_scope.get( |
| _FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY, _SCOPE_MISSING |
| ) |
| fastapi_scope[_FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY] = route_context |
| try: |
| match, child_scope = original_route.matches(scope) |
| finally: |
| _restore_fastapi_scope_key( |
| scope, |
| _FASTAPI_EFFECTIVE_ROUTE_CONTEXT_KEY, |
| previous_context, |
| ) |
| route = original_route |
| else: |
| match, child_scope = candidate.matches(scope) |
| route = candidate.starlette_route or original_route |
| else: |
| route_context = None |
| match, child_scope = candidate.matches(scope) |
| route = candidate |
| if match == Match.FULL: |
| if full is None or self._frontend_match_is_more_specific( |
| child_scope, full[0] |
| ): |
| full = (child_scope, route, route_context) |
| elif match == Match.PARTIAL: |
| if partial is None or self._frontend_match_is_more_specific( |
| child_scope, partial[0] |
| ): |
| partial = (child_scope, route, route_context) |
| if full is not None: |
| child_scope, route, route_context = full |
| return Match.FULL, child_scope, route, route_context |
| if partial is not None: |
| child_scope, route, route_context = partial |
| return Match.PARTIAL, child_scope, route, route_context |
| return Match.NONE, {}, None, None |
|
|
| def _frontend_match_is_more_specific( |
| self, child_scope: Scope, previous_child_scope: Scope |
| ) -> bool: |
| specificity = _frontend_scope_specificity(child_scope) |
| previous_specificity = _frontend_scope_specificity(previous_child_scope) |
| if specificity is None or previous_specificity is None: |
| return False |
| return specificity > previous_specificity |
|
|
| def route( |
| self, |
| path: str, |
| methods: Collection[str] | None = None, |
| name: str | None = None, |
| include_in_schema: bool = True, |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| def decorator(func: DecoratedCallable) -> DecoratedCallable: |
| self.add_route( |
| path, |
| func, |
| methods=methods, |
| name=name, |
| include_in_schema=include_in_schema, |
| ) |
| return func |
|
|
| return decorator |
|
|
| def add_api_route( |
| self, |
| path: str, |
| endpoint: Callable[..., Any], |
| *, |
| response_model: Any = Default(None), |
| status_code: int | None = None, |
| tags: list[str | Enum] | None = None, |
| dependencies: Sequence[params.Depends] | None = None, |
| summary: str | None = None, |
| description: str | None = None, |
| response_description: str = "Successful Response", |
| responses: dict[int | str, dict[str, Any]] | None = None, |
| deprecated: bool | None = None, |
| methods: set[str] | list[str] | None = None, |
| operation_id: str | None = None, |
| response_model_include: IncEx | None = None, |
| response_model_exclude: IncEx | None = 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] | DefaultPlaceholder = Default(JSONResponse), |
| name: str | None = None, |
| route_class_override: type[APIRoute] | None = None, |
| callbacks: list[BaseRoute] | None = None, |
| openapi_extra: dict[str, Any] | None = None, |
| generate_unique_id_function: Callable[[APIRoute], str] |
| | DefaultPlaceholder = Default(generate_unique_id), |
| strict_content_type: bool | DefaultPlaceholder = Default(True), |
| ) -> None: |
| route_class = route_class_override or self.route_class |
| responses = responses or {} |
| combined_responses = {**self.responses, **responses} |
| current_response_class = get_value_or_default( |
| response_class, self.default_response_class |
| ) |
| current_tags = self.tags.copy() |
| if tags: |
| current_tags.extend(tags) |
| current_dependencies = self.dependencies.copy() |
| if dependencies: |
| current_dependencies.extend(dependencies) |
| current_callbacks = self.callbacks.copy() |
| if callbacks: |
| current_callbacks.extend(callbacks) |
| current_generate_unique_id = get_value_or_default( |
| generate_unique_id_function, self.generate_unique_id_function |
| ) |
| route = route_class( |
| self.prefix + path, |
| endpoint=endpoint, |
| response_model=response_model, |
| status_code=status_code, |
| tags=current_tags, |
| dependencies=current_dependencies, |
| summary=summary, |
| description=description, |
| response_description=response_description, |
| responses=combined_responses, |
| deprecated=deprecated or self.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 and self.include_in_schema, |
| response_class=current_response_class, |
| name=name, |
| dependency_overrides_provider=self.dependency_overrides_provider, |
| callbacks=current_callbacks, |
| openapi_extra=openapi_extra, |
| generate_unique_id_function=current_generate_unique_id, |
| strict_content_type=get_value_or_default( |
| strict_content_type, self.strict_content_type |
| ), |
| ) |
| self.routes.append(route) |
| self._mark_routes_changed() |
|
|
| def api_route( |
| self, |
| path: str, |
| *, |
| response_model: Any = Default(None), |
| status_code: int | None = None, |
| tags: list[str | Enum] | None = None, |
| dependencies: Sequence[params.Depends] | None = None, |
| summary: str | None = None, |
| description: str | None = None, |
| response_description: str = "Successful Response", |
| responses: dict[int | str, dict[str, Any]] | None = None, |
| deprecated: bool | None = None, |
| methods: list[str] | None = None, |
| operation_id: str | None = None, |
| response_model_include: IncEx | None = None, |
| response_model_exclude: IncEx | None = 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: str | None = None, |
| callbacks: list[BaseRoute] | None = None, |
| openapi_extra: dict[str, Any] | None = None, |
| generate_unique_id_function: Callable[[APIRoute], str] = Default( |
| generate_unique_id |
| ), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| def decorator(func: DecoratedCallable) -> DecoratedCallable: |
| self.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, |
| callbacks=callbacks, |
| 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: str | None = None, |
| *, |
| dependencies: Sequence[params.Depends] | None = None, |
| ) -> None: |
| current_dependencies = self.dependencies.copy() |
| if dependencies: |
| current_dependencies.extend(dependencies) |
|
|
| route = APIWebSocketRoute( |
| self.prefix + path, |
| endpoint=endpoint, |
| name=name, |
| dependencies=current_dependencies, |
| dependency_overrides_provider=self.dependency_overrides_provider, |
| ) |
| self.routes.append(route) |
| self._mark_routes_changed() |
|
|
| def websocket( |
| self, |
| path: Annotated[ |
| str, |
| Doc( |
| """ |
| WebSocket path. |
| """ |
| ), |
| ], |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A name for the WebSocket. Only used internally. |
| """ |
| ), |
| ] = None, |
| *, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be used for this |
| WebSocket. |
| |
| Read more about it in the |
| [FastAPI docs for WebSockets](https://fastapi.tiangolo.com/advanced/websockets/). |
| """ |
| ), |
| ] = None, |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Decorate a WebSocket function. |
| |
| Read more about it in the |
| [FastAPI docs for WebSockets](https://fastapi.tiangolo.com/advanced/websockets/). |
| |
| **Example** |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI, WebSocket |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.websocket("/ws") |
| async def websocket_endpoint(websocket: WebSocket): |
| await websocket.accept() |
| while True: |
| data = await websocket.receive_text() |
| await websocket.send_text(f"Message text was: {data}") |
| |
| app.include_router(router) |
| ``` |
| """ |
|
|
| def decorator(func: DecoratedCallable) -> DecoratedCallable: |
| self.add_api_websocket_route( |
| path, func, name=name, dependencies=dependencies |
| ) |
| return func |
|
|
| return decorator |
|
|
| def websocket_route( |
| self, path: str, name: str | None = None |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| def decorator(func: DecoratedCallable) -> DecoratedCallable: |
| self.add_websocket_route(path, func, name=name) |
| return func |
|
|
| return decorator |
|
|
| def include_router( |
| self, |
| router: Annotated["APIRouter", Doc("The `APIRouter` to include.")], |
| *, |
| prefix: Annotated[str, Doc("An optional path prefix for the router.")] = "", |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to all the *path operations* in this |
| router. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to all the |
| *path operations* in this router. |
| |
| Read more about it in the |
| [FastAPI docs for Bigger Applications - Multiple Files](https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies). |
| """ |
| ), |
| ] = None, |
| default_response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| The default response class to be used. |
| |
| Read more in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#default-response-class). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses to be shown in OpenAPI. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Additional Responses in OpenAPI](https://fastapi.tiangolo.com/advanced/additional-responses/). |
| |
| And in the |
| [FastAPI docs for Bigger Applications](https://fastapi.tiangolo.com/tutorial/bigger-applications/#include-an-apirouter-with-a-custom-prefix-tags-responses-and-dependencies). |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| OpenAPI callbacks that should apply to all *path operations* in this |
| router. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark all *path operations* in this router as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include (or not) all the *path operations* in this router in the |
| generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = True, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> None: |
| """ |
| Include another `APIRouter` in the same current `APIRouter`. |
| |
| Read more about it in the |
| [FastAPI docs for Bigger Applications](https://fastapi.tiangolo.com/tutorial/bigger-applications/). |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| |
| app = FastAPI() |
| internal_router = APIRouter() |
| users_router = APIRouter() |
| |
| @users_router.get("/users/") |
| def read_users(): |
| return [{"name": "Rick"}, {"name": "Morty"}] |
| |
| internal_router.include_router(users_router) |
| app.include_router(internal_router) |
| ``` |
| """ |
| assert self is not router, ( |
| "Cannot include the same APIRouter instance into itself. " |
| "Did you mean to include a different router?" |
| ) |
| assert not router._contains_router(self), ( |
| "Cannot include an APIRouter instance that already includes this router. " |
| "Did you mean to include a different router?" |
| ) |
| if prefix: |
| assert prefix.startswith("/"), "A path prefix must start with '/'" |
| assert not prefix.endswith("/"), ( |
| "A path prefix must not end with '/', as the routes will start with '/'" |
| ) |
| else: |
| for route, route_context in _iter_routes_with_context(router.routes): |
| if route_context is None: |
| path = getattr(route, "path", None) |
| name = getattr(route, "name", "unknown") |
| elif route_context.starlette_route is not None: |
| path = getattr(route_context.starlette_route, "path", None) |
| name = getattr(route_context.starlette_route, "name", "unknown") |
| else: |
| path = route_context.path |
| name = route_context.name |
| if path is not None and not path: |
| raise FastAPIError( |
| f"Prefix and path cannot be both empty (path operation: {name})" |
| ) |
| include_context = _RouterIncludeContext.for_include( |
| parent_router=self, |
| included_router=router, |
| prefix=prefix, |
| tags=tags, |
| dependencies=dependencies, |
| default_response_class=default_response_class, |
| responses=responses, |
| callbacks=callbacks, |
| deprecated=deprecated, |
| include_in_schema=include_in_schema, |
| generate_unique_id_function=generate_unique_id_function, |
| ) |
| self.routes.append( |
| _IncludedRouter(original_router=router, include_context=include_context) |
| ) |
| self._mark_routes_changed() |
| for handler in router.on_startup: |
| self.add_event_handler("startup", handler) |
| for handler in router.on_shutdown: |
| self.add_event_handler("shutdown", handler) |
| self.lifespan_context = _merge_lifespan_context( |
| self.lifespan_context, |
| router.lifespan_context, |
| ) |
|
|
| def get( |
| self, |
| path: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path to be used for this *path operation*. |
| |
| For example, in `http://example.com/items`, the path is `/items`. |
| """ |
| ), |
| ], |
| *, |
| response_model: Annotated[ |
| Any, |
| Doc( |
| """ |
| The type to use for the response. |
| |
| It could be any valid Pydantic *field* type. So, it doesn't have to |
| be a Pydantic model, it could be other things, like a `list`, `dict`, |
| etc. |
| |
| It will be used for: |
| |
| * Documentation: the generated OpenAPI (and the UI at `/docs`) will |
| show it as the response (JSON Schema). |
| * Serialization: you could return an arbitrary object and the |
| `response_model` would be used to serialize that object into the |
| corresponding JSON. |
| * Filtering: the JSON sent to the client will only contain the data |
| (fields) defined in the `response_model`. If you returned an object |
| that contains an attribute `password` but the `response_model` does |
| not include that field, the JSON sent to the client would not have |
| that `password`. |
| * Validation: whatever you return will be serialized with the |
| `response_model`, converting any data as necessary to generate the |
| corresponding JSON. But if the data in the object returned is not |
| valid, that would mean a violation of the contract with the client, |
| so it's an error from the API developer. So, FastAPI will raise an |
| error and return a 500 error code (Internal Server Error). |
| |
| Read more about it in the |
| [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/). |
| """ |
| ), |
| ] = Default(None), |
| status_code: Annotated[ |
| int | None, |
| Doc( |
| """ |
| The default status code to be used for the response. |
| |
| You could override the status code by returning a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/). |
| """ |
| ), |
| ] = None, |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to the |
| *path operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/). |
| """ |
| ), |
| ] = None, |
| summary: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A summary for the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| description: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A description for the *path operation*. |
| |
| If not provided, it will be extracted automatically from the docstring |
| of the *path operation function*. |
| |
| It can contain Markdown. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| response_description: Annotated[ |
| str, |
| Doc( |
| """ |
| The description for the default response. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = "Successful Response", |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses that could be returned by this *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark this *path operation* as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| operation_id: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Custom operation ID to be used by this *path operation*. |
| |
| By default, it is generated automatically. |
| |
| If you provide a custom operation ID, you need to make sure it is |
| unique for the whole API. |
| |
| You can customize the |
| operation ID generation with the parameter |
| `generate_unique_id_function` in the `FastAPI` class. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = None, |
| response_model_include: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to include only certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_exclude: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to exclude certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_by_alias: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response model |
| should be serialized by alias when an alias is used. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = True, |
| response_model_exclude_unset: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that were not set and |
| have their default values. This is different from |
| `response_model_exclude_defaults` in that if the fields are set, |
| they will be included in the response, even if the value is the same |
| as the default. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_defaults: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that have the same value |
| as the default. This is different from `response_model_exclude_unset` |
| in that if the fields are set but contain the same default values, |
| they will be excluded from the response. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_none: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data should |
| exclude fields set to `None`. |
| |
| This is much simpler (less smart) than `response_model_exclude_unset` |
| and `response_model_exclude_defaults`. You probably want to use one of |
| those two instead of this one, as those allow returning `None` values |
| when it makes sense. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none). |
| """ |
| ), |
| ] = False, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include this *path operation* in the generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| Response class to be used for this *path operation*. |
| |
| This will not be used if you return a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Name for this *path operation*. Only used internally. |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| List of *path operations* that will be used as OpenAPI callbacks. |
| |
| This is only for OpenAPI documentation, the callbacks won't be used |
| directly. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| openapi_extra: Annotated[ |
| dict[str, Any] | None, |
| Doc( |
| """ |
| Extra metadata to be included in the OpenAPI schema for this *path |
| operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema). |
| """ |
| ), |
| ] = None, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add a *path operation* using an HTTP GET operation. |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.get("/items/") |
| def read_items(): |
| return [{"name": "Empanada"}, {"name": "Arepa"}] |
| |
| app.include_router(router) |
| ``` |
| """ |
| return self.api_route( |
| path=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, |
| methods=["GET"], |
| 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: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path to be used for this *path operation*. |
| |
| For example, in `http://example.com/items`, the path is `/items`. |
| """ |
| ), |
| ], |
| *, |
| response_model: Annotated[ |
| Any, |
| Doc( |
| """ |
| The type to use for the response. |
| |
| It could be any valid Pydantic *field* type. So, it doesn't have to |
| be a Pydantic model, it could be other things, like a `list`, `dict`, |
| etc. |
| |
| It will be used for: |
| |
| * Documentation: the generated OpenAPI (and the UI at `/docs`) will |
| show it as the response (JSON Schema). |
| * Serialization: you could return an arbitrary object and the |
| `response_model` would be used to serialize that object into the |
| corresponding JSON. |
| * Filtering: the JSON sent to the client will only contain the data |
| (fields) defined in the `response_model`. If you returned an object |
| that contains an attribute `password` but the `response_model` does |
| not include that field, the JSON sent to the client would not have |
| that `password`. |
| * Validation: whatever you return will be serialized with the |
| `response_model`, converting any data as necessary to generate the |
| corresponding JSON. But if the data in the object returned is not |
| valid, that would mean a violation of the contract with the client, |
| so it's an error from the API developer. So, FastAPI will raise an |
| error and return a 500 error code (Internal Server Error). |
| |
| Read more about it in the |
| [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/). |
| """ |
| ), |
| ] = Default(None), |
| status_code: Annotated[ |
| int | None, |
| Doc( |
| """ |
| The default status code to be used for the response. |
| |
| You could override the status code by returning a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/). |
| """ |
| ), |
| ] = None, |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to the |
| *path operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/). |
| """ |
| ), |
| ] = None, |
| summary: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A summary for the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| description: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A description for the *path operation*. |
| |
| If not provided, it will be extracted automatically from the docstring |
| of the *path operation function*. |
| |
| It can contain Markdown. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| response_description: Annotated[ |
| str, |
| Doc( |
| """ |
| The description for the default response. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = "Successful Response", |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses that could be returned by this *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark this *path operation* as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| operation_id: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Custom operation ID to be used by this *path operation*. |
| |
| By default, it is generated automatically. |
| |
| If you provide a custom operation ID, you need to make sure it is |
| unique for the whole API. |
| |
| You can customize the |
| operation ID generation with the parameter |
| `generate_unique_id_function` in the `FastAPI` class. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = None, |
| response_model_include: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to include only certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_exclude: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to exclude certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_by_alias: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response model |
| should be serialized by alias when an alias is used. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = True, |
| response_model_exclude_unset: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that were not set and |
| have their default values. This is different from |
| `response_model_exclude_defaults` in that if the fields are set, |
| they will be included in the response, even if the value is the same |
| as the default. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_defaults: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that have the same value |
| as the default. This is different from `response_model_exclude_unset` |
| in that if the fields are set but contain the same default values, |
| they will be excluded from the response. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_none: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data should |
| exclude fields set to `None`. |
| |
| This is much simpler (less smart) than `response_model_exclude_unset` |
| and `response_model_exclude_defaults`. You probably want to use one of |
| those two instead of this one, as those allow returning `None` values |
| when it makes sense. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none). |
| """ |
| ), |
| ] = False, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include this *path operation* in the generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| Response class to be used for this *path operation*. |
| |
| This will not be used if you return a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Name for this *path operation*. Only used internally. |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| List of *path operations* that will be used as OpenAPI callbacks. |
| |
| This is only for OpenAPI documentation, the callbacks won't be used |
| directly. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| openapi_extra: Annotated[ |
| dict[str, Any] | None, |
| Doc( |
| """ |
| Extra metadata to be included in the OpenAPI schema for this *path |
| operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema). |
| """ |
| ), |
| ] = None, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add a *path operation* using an HTTP PUT operation. |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| from pydantic import BaseModel |
| |
| class Item(BaseModel): |
| name: str |
| description: str | None = None |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.put("/items/{item_id}") |
| def replace_item(item_id: str, item: Item): |
| return {"message": "Item replaced", "id": item_id} |
| |
| app.include_router(router) |
| ``` |
| """ |
| return self.api_route( |
| path=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, |
| methods=["PUT"], |
| 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: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path to be used for this *path operation*. |
| |
| For example, in `http://example.com/items`, the path is `/items`. |
| """ |
| ), |
| ], |
| *, |
| response_model: Annotated[ |
| Any, |
| Doc( |
| """ |
| The type to use for the response. |
| |
| It could be any valid Pydantic *field* type. So, it doesn't have to |
| be a Pydantic model, it could be other things, like a `list`, `dict`, |
| etc. |
| |
| It will be used for: |
| |
| * Documentation: the generated OpenAPI (and the UI at `/docs`) will |
| show it as the response (JSON Schema). |
| * Serialization: you could return an arbitrary object and the |
| `response_model` would be used to serialize that object into the |
| corresponding JSON. |
| * Filtering: the JSON sent to the client will only contain the data |
| (fields) defined in the `response_model`. If you returned an object |
| that contains an attribute `password` but the `response_model` does |
| not include that field, the JSON sent to the client would not have |
| that `password`. |
| * Validation: whatever you return will be serialized with the |
| `response_model`, converting any data as necessary to generate the |
| corresponding JSON. But if the data in the object returned is not |
| valid, that would mean a violation of the contract with the client, |
| so it's an error from the API developer. So, FastAPI will raise an |
| error and return a 500 error code (Internal Server Error). |
| |
| Read more about it in the |
| [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/). |
| """ |
| ), |
| ] = Default(None), |
| status_code: Annotated[ |
| int | None, |
| Doc( |
| """ |
| The default status code to be used for the response. |
| |
| You could override the status code by returning a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/). |
| """ |
| ), |
| ] = None, |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to the |
| *path operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/). |
| """ |
| ), |
| ] = None, |
| summary: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A summary for the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| description: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A description for the *path operation*. |
| |
| If not provided, it will be extracted automatically from the docstring |
| of the *path operation function*. |
| |
| It can contain Markdown. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| response_description: Annotated[ |
| str, |
| Doc( |
| """ |
| The description for the default response. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = "Successful Response", |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses that could be returned by this *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark this *path operation* as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| operation_id: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Custom operation ID to be used by this *path operation*. |
| |
| By default, it is generated automatically. |
| |
| If you provide a custom operation ID, you need to make sure it is |
| unique for the whole API. |
| |
| You can customize the |
| operation ID generation with the parameter |
| `generate_unique_id_function` in the `FastAPI` class. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = None, |
| response_model_include: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to include only certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_exclude: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to exclude certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_by_alias: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response model |
| should be serialized by alias when an alias is used. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = True, |
| response_model_exclude_unset: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that were not set and |
| have their default values. This is different from |
| `response_model_exclude_defaults` in that if the fields are set, |
| they will be included in the response, even if the value is the same |
| as the default. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_defaults: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that have the same value |
| as the default. This is different from `response_model_exclude_unset` |
| in that if the fields are set but contain the same default values, |
| they will be excluded from the response. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_none: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data should |
| exclude fields set to `None`. |
| |
| This is much simpler (less smart) than `response_model_exclude_unset` |
| and `response_model_exclude_defaults`. You probably want to use one of |
| those two instead of this one, as those allow returning `None` values |
| when it makes sense. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none). |
| """ |
| ), |
| ] = False, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include this *path operation* in the generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| Response class to be used for this *path operation*. |
| |
| This will not be used if you return a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Name for this *path operation*. Only used internally. |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| List of *path operations* that will be used as OpenAPI callbacks. |
| |
| This is only for OpenAPI documentation, the callbacks won't be used |
| directly. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| openapi_extra: Annotated[ |
| dict[str, Any] | None, |
| Doc( |
| """ |
| Extra metadata to be included in the OpenAPI schema for this *path |
| operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema). |
| """ |
| ), |
| ] = None, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add a *path operation* using an HTTP POST operation. |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| from pydantic import BaseModel |
| |
| class Item(BaseModel): |
| name: str |
| description: str | None = None |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.post("/items/") |
| def create_item(item: Item): |
| return {"message": "Item created"} |
| |
| app.include_router(router) |
| ``` |
| """ |
| return self.api_route( |
| path=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, |
| methods=["POST"], |
| 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: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path to be used for this *path operation*. |
| |
| For example, in `http://example.com/items`, the path is `/items`. |
| """ |
| ), |
| ], |
| *, |
| response_model: Annotated[ |
| Any, |
| Doc( |
| """ |
| The type to use for the response. |
| |
| It could be any valid Pydantic *field* type. So, it doesn't have to |
| be a Pydantic model, it could be other things, like a `list`, `dict`, |
| etc. |
| |
| It will be used for: |
| |
| * Documentation: the generated OpenAPI (and the UI at `/docs`) will |
| show it as the response (JSON Schema). |
| * Serialization: you could return an arbitrary object and the |
| `response_model` would be used to serialize that object into the |
| corresponding JSON. |
| * Filtering: the JSON sent to the client will only contain the data |
| (fields) defined in the `response_model`. If you returned an object |
| that contains an attribute `password` but the `response_model` does |
| not include that field, the JSON sent to the client would not have |
| that `password`. |
| * Validation: whatever you return will be serialized with the |
| `response_model`, converting any data as necessary to generate the |
| corresponding JSON. But if the data in the object returned is not |
| valid, that would mean a violation of the contract with the client, |
| so it's an error from the API developer. So, FastAPI will raise an |
| error and return a 500 error code (Internal Server Error). |
| |
| Read more about it in the |
| [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/). |
| """ |
| ), |
| ] = Default(None), |
| status_code: Annotated[ |
| int | None, |
| Doc( |
| """ |
| The default status code to be used for the response. |
| |
| You could override the status code by returning a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/). |
| """ |
| ), |
| ] = None, |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to the |
| *path operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/). |
| """ |
| ), |
| ] = None, |
| summary: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A summary for the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| description: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A description for the *path operation*. |
| |
| If not provided, it will be extracted automatically from the docstring |
| of the *path operation function*. |
| |
| It can contain Markdown. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| response_description: Annotated[ |
| str, |
| Doc( |
| """ |
| The description for the default response. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = "Successful Response", |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses that could be returned by this *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark this *path operation* as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| operation_id: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Custom operation ID to be used by this *path operation*. |
| |
| By default, it is generated automatically. |
| |
| If you provide a custom operation ID, you need to make sure it is |
| unique for the whole API. |
| |
| You can customize the |
| operation ID generation with the parameter |
| `generate_unique_id_function` in the `FastAPI` class. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = None, |
| response_model_include: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to include only certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_exclude: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to exclude certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_by_alias: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response model |
| should be serialized by alias when an alias is used. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = True, |
| response_model_exclude_unset: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that were not set and |
| have their default values. This is different from |
| `response_model_exclude_defaults` in that if the fields are set, |
| they will be included in the response, even if the value is the same |
| as the default. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_defaults: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that have the same value |
| as the default. This is different from `response_model_exclude_unset` |
| in that if the fields are set but contain the same default values, |
| they will be excluded from the response. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_none: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data should |
| exclude fields set to `None`. |
| |
| This is much simpler (less smart) than `response_model_exclude_unset` |
| and `response_model_exclude_defaults`. You probably want to use one of |
| those two instead of this one, as those allow returning `None` values |
| when it makes sense. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none). |
| """ |
| ), |
| ] = False, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include this *path operation* in the generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| Response class to be used for this *path operation*. |
| |
| This will not be used if you return a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Name for this *path operation*. Only used internally. |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| List of *path operations* that will be used as OpenAPI callbacks. |
| |
| This is only for OpenAPI documentation, the callbacks won't be used |
| directly. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| openapi_extra: Annotated[ |
| dict[str, Any] | None, |
| Doc( |
| """ |
| Extra metadata to be included in the OpenAPI schema for this *path |
| operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema). |
| """ |
| ), |
| ] = None, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add a *path operation* using an HTTP DELETE operation. |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.delete("/items/{item_id}") |
| def delete_item(item_id: str): |
| return {"message": "Item deleted"} |
| |
| app.include_router(router) |
| ``` |
| """ |
| return self.api_route( |
| path=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, |
| methods=["DELETE"], |
| 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: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path to be used for this *path operation*. |
| |
| For example, in `http://example.com/items`, the path is `/items`. |
| """ |
| ), |
| ], |
| *, |
| response_model: Annotated[ |
| Any, |
| Doc( |
| """ |
| The type to use for the response. |
| |
| It could be any valid Pydantic *field* type. So, it doesn't have to |
| be a Pydantic model, it could be other things, like a `list`, `dict`, |
| etc. |
| |
| It will be used for: |
| |
| * Documentation: the generated OpenAPI (and the UI at `/docs`) will |
| show it as the response (JSON Schema). |
| * Serialization: you could return an arbitrary object and the |
| `response_model` would be used to serialize that object into the |
| corresponding JSON. |
| * Filtering: the JSON sent to the client will only contain the data |
| (fields) defined in the `response_model`. If you returned an object |
| that contains an attribute `password` but the `response_model` does |
| not include that field, the JSON sent to the client would not have |
| that `password`. |
| * Validation: whatever you return will be serialized with the |
| `response_model`, converting any data as necessary to generate the |
| corresponding JSON. But if the data in the object returned is not |
| valid, that would mean a violation of the contract with the client, |
| so it's an error from the API developer. So, FastAPI will raise an |
| error and return a 500 error code (Internal Server Error). |
| |
| Read more about it in the |
| [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/). |
| """ |
| ), |
| ] = Default(None), |
| status_code: Annotated[ |
| int | None, |
| Doc( |
| """ |
| The default status code to be used for the response. |
| |
| You could override the status code by returning a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/). |
| """ |
| ), |
| ] = None, |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to the |
| *path operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/). |
| """ |
| ), |
| ] = None, |
| summary: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A summary for the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| description: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A description for the *path operation*. |
| |
| If not provided, it will be extracted automatically from the docstring |
| of the *path operation function*. |
| |
| It can contain Markdown. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| response_description: Annotated[ |
| str, |
| Doc( |
| """ |
| The description for the default response. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = "Successful Response", |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses that could be returned by this *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark this *path operation* as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| operation_id: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Custom operation ID to be used by this *path operation*. |
| |
| By default, it is generated automatically. |
| |
| If you provide a custom operation ID, you need to make sure it is |
| unique for the whole API. |
| |
| You can customize the |
| operation ID generation with the parameter |
| `generate_unique_id_function` in the `FastAPI` class. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = None, |
| response_model_include: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to include only certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_exclude: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to exclude certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_by_alias: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response model |
| should be serialized by alias when an alias is used. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = True, |
| response_model_exclude_unset: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that were not set and |
| have their default values. This is different from |
| `response_model_exclude_defaults` in that if the fields are set, |
| they will be included in the response, even if the value is the same |
| as the default. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_defaults: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that have the same value |
| as the default. This is different from `response_model_exclude_unset` |
| in that if the fields are set but contain the same default values, |
| they will be excluded from the response. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_none: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data should |
| exclude fields set to `None`. |
| |
| This is much simpler (less smart) than `response_model_exclude_unset` |
| and `response_model_exclude_defaults`. You probably want to use one of |
| those two instead of this one, as those allow returning `None` values |
| when it makes sense. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none). |
| """ |
| ), |
| ] = False, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include this *path operation* in the generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| Response class to be used for this *path operation*. |
| |
| This will not be used if you return a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Name for this *path operation*. Only used internally. |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| List of *path operations* that will be used as OpenAPI callbacks. |
| |
| This is only for OpenAPI documentation, the callbacks won't be used |
| directly. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| openapi_extra: Annotated[ |
| dict[str, Any] | None, |
| Doc( |
| """ |
| Extra metadata to be included in the OpenAPI schema for this *path |
| operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema). |
| """ |
| ), |
| ] = None, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add a *path operation* using an HTTP OPTIONS operation. |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.options("/items/") |
| def get_item_options(): |
| return {"additions": ["Aji", "Guacamole"]} |
| |
| app.include_router(router) |
| ``` |
| """ |
| return self.api_route( |
| path=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, |
| methods=["OPTIONS"], |
| 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: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path to be used for this *path operation*. |
| |
| For example, in `http://example.com/items`, the path is `/items`. |
| """ |
| ), |
| ], |
| *, |
| response_model: Annotated[ |
| Any, |
| Doc( |
| """ |
| The type to use for the response. |
| |
| It could be any valid Pydantic *field* type. So, it doesn't have to |
| be a Pydantic model, it could be other things, like a `list`, `dict`, |
| etc. |
| |
| It will be used for: |
| |
| * Documentation: the generated OpenAPI (and the UI at `/docs`) will |
| show it as the response (JSON Schema). |
| * Serialization: you could return an arbitrary object and the |
| `response_model` would be used to serialize that object into the |
| corresponding JSON. |
| * Filtering: the JSON sent to the client will only contain the data |
| (fields) defined in the `response_model`. If you returned an object |
| that contains an attribute `password` but the `response_model` does |
| not include that field, the JSON sent to the client would not have |
| that `password`. |
| * Validation: whatever you return will be serialized with the |
| `response_model`, converting any data as necessary to generate the |
| corresponding JSON. But if the data in the object returned is not |
| valid, that would mean a violation of the contract with the client, |
| so it's an error from the API developer. So, FastAPI will raise an |
| error and return a 500 error code (Internal Server Error). |
| |
| Read more about it in the |
| [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/). |
| """ |
| ), |
| ] = Default(None), |
| status_code: Annotated[ |
| int | None, |
| Doc( |
| """ |
| The default status code to be used for the response. |
| |
| You could override the status code by returning a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/). |
| """ |
| ), |
| ] = None, |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to the |
| *path operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/). |
| """ |
| ), |
| ] = None, |
| summary: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A summary for the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| description: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A description for the *path operation*. |
| |
| If not provided, it will be extracted automatically from the docstring |
| of the *path operation function*. |
| |
| It can contain Markdown. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| response_description: Annotated[ |
| str, |
| Doc( |
| """ |
| The description for the default response. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = "Successful Response", |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses that could be returned by this *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark this *path operation* as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| operation_id: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Custom operation ID to be used by this *path operation*. |
| |
| By default, it is generated automatically. |
| |
| If you provide a custom operation ID, you need to make sure it is |
| unique for the whole API. |
| |
| You can customize the |
| operation ID generation with the parameter |
| `generate_unique_id_function` in the `FastAPI` class. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = None, |
| response_model_include: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to include only certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_exclude: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to exclude certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_by_alias: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response model |
| should be serialized by alias when an alias is used. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = True, |
| response_model_exclude_unset: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that were not set and |
| have their default values. This is different from |
| `response_model_exclude_defaults` in that if the fields are set, |
| they will be included in the response, even if the value is the same |
| as the default. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_defaults: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that have the same value |
| as the default. This is different from `response_model_exclude_unset` |
| in that if the fields are set but contain the same default values, |
| they will be excluded from the response. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_none: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data should |
| exclude fields set to `None`. |
| |
| This is much simpler (less smart) than `response_model_exclude_unset` |
| and `response_model_exclude_defaults`. You probably want to use one of |
| those two instead of this one, as those allow returning `None` values |
| when it makes sense. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none). |
| """ |
| ), |
| ] = False, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include this *path operation* in the generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| Response class to be used for this *path operation*. |
| |
| This will not be used if you return a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Name for this *path operation*. Only used internally. |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| List of *path operations* that will be used as OpenAPI callbacks. |
| |
| This is only for OpenAPI documentation, the callbacks won't be used |
| directly. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| openapi_extra: Annotated[ |
| dict[str, Any] | None, |
| Doc( |
| """ |
| Extra metadata to be included in the OpenAPI schema for this *path |
| operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema). |
| """ |
| ), |
| ] = None, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add a *path operation* using an HTTP HEAD operation. |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| from pydantic import BaseModel |
| |
| class Item(BaseModel): |
| name: str |
| description: str | None = None |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.head("/items/", status_code=204) |
| def get_items_headers(response: Response): |
| response.headers["X-Cat-Dog"] = "Alone in the world" |
| |
| app.include_router(router) |
| ``` |
| """ |
| return self.api_route( |
| path=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, |
| methods=["HEAD"], |
| 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: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path to be used for this *path operation*. |
| |
| For example, in `http://example.com/items`, the path is `/items`. |
| """ |
| ), |
| ], |
| *, |
| response_model: Annotated[ |
| Any, |
| Doc( |
| """ |
| The type to use for the response. |
| |
| It could be any valid Pydantic *field* type. So, it doesn't have to |
| be a Pydantic model, it could be other things, like a `list`, `dict`, |
| etc. |
| |
| It will be used for: |
| |
| * Documentation: the generated OpenAPI (and the UI at `/docs`) will |
| show it as the response (JSON Schema). |
| * Serialization: you could return an arbitrary object and the |
| `response_model` would be used to serialize that object into the |
| corresponding JSON. |
| * Filtering: the JSON sent to the client will only contain the data |
| (fields) defined in the `response_model`. If you returned an object |
| that contains an attribute `password` but the `response_model` does |
| not include that field, the JSON sent to the client would not have |
| that `password`. |
| * Validation: whatever you return will be serialized with the |
| `response_model`, converting any data as necessary to generate the |
| corresponding JSON. But if the data in the object returned is not |
| valid, that would mean a violation of the contract with the client, |
| so it's an error from the API developer. So, FastAPI will raise an |
| error and return a 500 error code (Internal Server Error). |
| |
| Read more about it in the |
| [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/). |
| """ |
| ), |
| ] = Default(None), |
| status_code: Annotated[ |
| int | None, |
| Doc( |
| """ |
| The default status code to be used for the response. |
| |
| You could override the status code by returning a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/). |
| """ |
| ), |
| ] = None, |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to the |
| *path operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/). |
| """ |
| ), |
| ] = None, |
| summary: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A summary for the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| description: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A description for the *path operation*. |
| |
| If not provided, it will be extracted automatically from the docstring |
| of the *path operation function*. |
| |
| It can contain Markdown. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| response_description: Annotated[ |
| str, |
| Doc( |
| """ |
| The description for the default response. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = "Successful Response", |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses that could be returned by this *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark this *path operation* as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| operation_id: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Custom operation ID to be used by this *path operation*. |
| |
| By default, it is generated automatically. |
| |
| If you provide a custom operation ID, you need to make sure it is |
| unique for the whole API. |
| |
| You can customize the |
| operation ID generation with the parameter |
| `generate_unique_id_function` in the `FastAPI` class. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = None, |
| response_model_include: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to include only certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_exclude: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to exclude certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_by_alias: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response model |
| should be serialized by alias when an alias is used. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = True, |
| response_model_exclude_unset: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that were not set and |
| have their default values. This is different from |
| `response_model_exclude_defaults` in that if the fields are set, |
| they will be included in the response, even if the value is the same |
| as the default. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_defaults: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that have the same value |
| as the default. This is different from `response_model_exclude_unset` |
| in that if the fields are set but contain the same default values, |
| they will be excluded from the response. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_none: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data should |
| exclude fields set to `None`. |
| |
| This is much simpler (less smart) than `response_model_exclude_unset` |
| and `response_model_exclude_defaults`. You probably want to use one of |
| those two instead of this one, as those allow returning `None` values |
| when it makes sense. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none). |
| """ |
| ), |
| ] = False, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include this *path operation* in the generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| Response class to be used for this *path operation*. |
| |
| This will not be used if you return a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Name for this *path operation*. Only used internally. |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| List of *path operations* that will be used as OpenAPI callbacks. |
| |
| This is only for OpenAPI documentation, the callbacks won't be used |
| directly. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| openapi_extra: Annotated[ |
| dict[str, Any] | None, |
| Doc( |
| """ |
| Extra metadata to be included in the OpenAPI schema for this *path |
| operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema). |
| """ |
| ), |
| ] = None, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add a *path operation* using an HTTP PATCH operation. |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| from pydantic import BaseModel |
| |
| class Item(BaseModel): |
| name: str |
| description: str | None = None |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.patch("/items/") |
| def update_item(item: Item): |
| return {"message": "Item updated in place"} |
| |
| app.include_router(router) |
| ``` |
| """ |
| return self.api_route( |
| path=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, |
| methods=["PATCH"], |
| 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: Annotated[ |
| str, |
| Doc( |
| """ |
| The URL path to be used for this *path operation*. |
| |
| For example, in `http://example.com/items`, the path is `/items`. |
| """ |
| ), |
| ], |
| *, |
| response_model: Annotated[ |
| Any, |
| Doc( |
| """ |
| The type to use for the response. |
| |
| It could be any valid Pydantic *field* type. So, it doesn't have to |
| be a Pydantic model, it could be other things, like a `list`, `dict`, |
| etc. |
| |
| It will be used for: |
| |
| * Documentation: the generated OpenAPI (and the UI at `/docs`) will |
| show it as the response (JSON Schema). |
| * Serialization: you could return an arbitrary object and the |
| `response_model` would be used to serialize that object into the |
| corresponding JSON. |
| * Filtering: the JSON sent to the client will only contain the data |
| (fields) defined in the `response_model`. If you returned an object |
| that contains an attribute `password` but the `response_model` does |
| not include that field, the JSON sent to the client would not have |
| that `password`. |
| * Validation: whatever you return will be serialized with the |
| `response_model`, converting any data as necessary to generate the |
| corresponding JSON. But if the data in the object returned is not |
| valid, that would mean a violation of the contract with the client, |
| so it's an error from the API developer. So, FastAPI will raise an |
| error and return a 500 error code (Internal Server Error). |
| |
| Read more about it in the |
| [FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/). |
| """ |
| ), |
| ] = Default(None), |
| status_code: Annotated[ |
| int | None, |
| Doc( |
| """ |
| The default status code to be used for the response. |
| |
| You could override the status code by returning a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/). |
| """ |
| ), |
| ] = None, |
| tags: Annotated[ |
| list[str | Enum] | None, |
| Doc( |
| """ |
| A list of tags to be applied to the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags). |
| """ |
| ), |
| ] = None, |
| dependencies: Annotated[ |
| Sequence[params.Depends] | None, |
| Doc( |
| """ |
| A list of dependencies (using `Depends()`) to be applied to the |
| *path operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/). |
| """ |
| ), |
| ] = None, |
| summary: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A summary for the *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| description: Annotated[ |
| str | None, |
| Doc( |
| """ |
| A description for the *path operation*. |
| |
| If not provided, it will be extracted automatically from the docstring |
| of the *path operation function*. |
| |
| It can contain Markdown. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/). |
| """ |
| ), |
| ] = None, |
| response_description: Annotated[ |
| str, |
| Doc( |
| """ |
| The description for the default response. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = "Successful Response", |
| responses: Annotated[ |
| dict[int | str, dict[str, Any]] | None, |
| Doc( |
| """ |
| Additional responses that could be returned by this *path operation*. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| deprecated: Annotated[ |
| bool | None, |
| Doc( |
| """ |
| Mark this *path operation* as deprecated. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| """ |
| ), |
| ] = None, |
| operation_id: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Custom operation ID to be used by this *path operation*. |
| |
| By default, it is generated automatically. |
| |
| If you provide a custom operation ID, you need to make sure it is |
| unique for the whole API. |
| |
| You can customize the |
| operation ID generation with the parameter |
| `generate_unique_id_function` in the `FastAPI` class. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = None, |
| response_model_include: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to include only certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_exclude: Annotated[ |
| IncEx | None, |
| Doc( |
| """ |
| Configuration passed to Pydantic to exclude certain fields in the |
| response data. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = None, |
| response_model_by_alias: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response model |
| should be serialized by alias when an alias is used. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude). |
| """ |
| ), |
| ] = True, |
| response_model_exclude_unset: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that were not set and |
| have their default values. This is different from |
| `response_model_exclude_defaults` in that if the fields are set, |
| they will be included in the response, even if the value is the same |
| as the default. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_defaults: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data |
| should have all the fields, including the ones that have the same value |
| as the default. This is different from `response_model_exclude_unset` |
| in that if the fields are set but contain the same default values, |
| they will be excluded from the response. |
| |
| When `True`, default values are omitted from the response. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter). |
| """ |
| ), |
| ] = False, |
| response_model_exclude_none: Annotated[ |
| bool, |
| Doc( |
| """ |
| Configuration passed to Pydantic to define if the response data should |
| exclude fields set to `None`. |
| |
| This is much simpler (less smart) than `response_model_exclude_unset` |
| and `response_model_exclude_defaults`. You probably want to use one of |
| those two instead of this one, as those allow returning `None` values |
| when it makes sense. |
| |
| Read more about it in the |
| [FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none). |
| """ |
| ), |
| ] = False, |
| include_in_schema: Annotated[ |
| bool, |
| Doc( |
| """ |
| Include this *path operation* in the generated OpenAPI schema. |
| |
| This affects the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi). |
| """ |
| ), |
| ] = True, |
| response_class: Annotated[ |
| type[Response], |
| Doc( |
| """ |
| Response class to be used for this *path operation*. |
| |
| This will not be used if you return a response directly. |
| |
| Read more about it in the |
| [FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse). |
| """ |
| ), |
| ] = Default(JSONResponse), |
| name: Annotated[ |
| str | None, |
| Doc( |
| """ |
| Name for this *path operation*. Only used internally. |
| """ |
| ), |
| ] = None, |
| callbacks: Annotated[ |
| list[BaseRoute] | None, |
| Doc( |
| """ |
| List of *path operations* that will be used as OpenAPI callbacks. |
| |
| This is only for OpenAPI documentation, the callbacks won't be used |
| directly. |
| |
| It will be added to the generated OpenAPI (e.g. visible at `/docs`). |
| |
| Read more about it in the |
| [FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/). |
| """ |
| ), |
| ] = None, |
| openapi_extra: Annotated[ |
| dict[str, Any] | None, |
| Doc( |
| """ |
| Extra metadata to be included in the OpenAPI schema for this *path |
| operation*. |
| |
| Read more about it in the |
| [FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema). |
| """ |
| ), |
| ] = None, |
| generate_unique_id_function: Annotated[ |
| Callable[[APIRoute], str], |
| Doc( |
| """ |
| Customize the function used to generate unique IDs for the *path |
| operations* shown in the generated OpenAPI. |
| |
| This is particularly useful when automatically generating clients or |
| SDKs for your API. |
| |
| Read more about it in the |
| [FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function). |
| """ |
| ), |
| ] = Default(generate_unique_id), |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add a *path operation* using an HTTP TRACE operation. |
| |
| ## Example |
| |
| ```python |
| from fastapi import APIRouter, FastAPI |
| from pydantic import BaseModel |
| |
| class Item(BaseModel): |
| name: str |
| description: str | None = None |
| |
| app = FastAPI() |
| router = APIRouter() |
| |
| @router.trace("/items/{item_id}") |
| def trace_item(item_id: str): |
| return None |
| |
| app.include_router(router) |
| ``` |
| """ |
| return self.api_route( |
| path=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, |
| methods=["TRACE"], |
| 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, |
| ) |
|
|
| |
| async def _startup(self) -> None: |
| """ |
| Run any `.on_startup` event handlers. |
| |
| This method is kept for backward compatibility after Starlette removed |
| support for on_startup/on_shutdown handlers. |
| |
| Ref: https://github.com/Kludex/starlette/pull/3117 |
| """ |
| for handler in self.on_startup: |
| if is_async_callable(handler): |
| await handler() |
| else: |
| handler() |
|
|
| |
| async def _shutdown(self) -> None: |
| """ |
| Run any `.on_shutdown` event handlers. |
| |
| This method is kept for backward compatibility after Starlette removed |
| support for on_startup/on_shutdown handlers. |
| |
| Ref: https://github.com/Kludex/starlette/pull/3117 |
| """ |
| for handler in self.on_shutdown: |
| if is_async_callable(handler): |
| await handler() |
| else: |
| handler() |
|
|
| |
| def add_event_handler( |
| self, |
| event_type: str, |
| func: Callable[[], Any], |
| ) -> None: |
| """ |
| Add an event handler function for startup or shutdown. |
| |
| This method is kept for backward compatibility after Starlette removed |
| support for on_startup/on_shutdown handlers. |
| |
| Ref: https://github.com/Kludex/starlette/pull/3117 |
| """ |
| assert event_type in ("startup", "shutdown") |
| if event_type == "startup": |
| self.on_startup.append(func) |
| else: |
| self.on_shutdown.append(func) |
|
|
| @deprecated( |
| """ |
| on_event is deprecated, use lifespan event handlers instead. |
| |
| Read more about it in the |
| [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/). |
| """ |
| ) |
| def on_event( |
| self, |
| event_type: Annotated[ |
| str, |
| Doc( |
| """ |
| The type of event. `startup` or `shutdown`. |
| """ |
| ), |
| ], |
| ) -> Callable[[DecoratedCallable], DecoratedCallable]: |
| """ |
| Add an event handler for the router. |
| |
| `on_event` is deprecated, use `lifespan` event handlers instead. |
| |
| Read more about it in the |
| [FastAPI docs for Lifespan Events](https://fastapi.tiangolo.com/advanced/events/#alternative-events-deprecated). |
| """ |
|
|
| def decorator(func: DecoratedCallable) -> DecoratedCallable: |
| self.add_event_handler(event_type, func) |
| return func |
|
|
| return decorator |
|
|