Browse Source

📝 Add docstrings to exception handlers, security utils, and utility functions

pull/15281/head
Brandon Sheffield 3 months ago
parent
commit
04393d2794
  1. 26
      fastapi/exception_handlers.py
  2. 6
      fastapi/security/utils.py
  3. 22
      fastapi/utils.py

26
fastapi/exception_handlers.py

@ -9,6 +9,16 @@ from starlette.status import WS_1008_POLICY_VIOLATION
async def http_exception_handler(request: Request, exc: HTTPException) -> Response:
"""
Default exception handler for `HTTPException`.
Returns a JSON response with the exception `detail` and the appropriate
status code and headers. For status codes that do not allow a response body
(1xx, 204, 205, 304), a plain `Response` without a body is returned.
Read more about it in the
[FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/).
"""
headers = getattr(exc, "headers", None)
if not is_body_allowed_for_status_code(exc.status_code):
return Response(status_code=exc.status_code, headers=headers)
@ -20,6 +30,16 @@ async def http_exception_handler(request: Request, exc: HTTPException) -> Respon
async def request_validation_exception_handler(
request: Request, exc: RequestValidationError
) -> JSONResponse:
"""
Default exception handler for `RequestValidationError`.
Returns a 422 JSON response containing the validation error details produced
when the request data (path parameters, query parameters, headers, or body)
does not match the endpoint's declared schema.
Read more about it in the
[FastAPI docs for Handling Errors](https://fastapi.tiangolo.com/tutorial/handling-errors/#override-request-validation-exceptions).
"""
return JSONResponse(
status_code=422,
content={"detail": jsonable_encoder(exc.errors())},
@ -29,6 +49,12 @@ async def request_validation_exception_handler(
async def websocket_request_validation_exception_handler(
websocket: WebSocket, exc: WebSocketRequestValidationError
) -> None:
"""
Default exception handler for `WebSocketRequestValidationError`.
Closes the WebSocket connection with a 1008 (Policy Violation) close code
and includes the validation error details as the close reason.
"""
await websocket.close(
code=WS_1008_POLICY_VIOLATION, reason=jsonable_encoder(exc.errors())
)

6
fastapi/security/utils.py

@ -1,6 +1,12 @@
def get_authorization_scheme_param(
authorization_header_value: str | None,
) -> tuple[str, str]:
"""
Parse an ``Authorization`` header value into its scheme and parameter.
Returns a tuple of ``(scheme, credentials)``. If the header is ``None``
or empty, returns ``("", "")``.
"""
if not authorization_header_value:
return "", ""
scheme, _, param = authorization_header_value.partition(" ")

22
fastapi/utils.py

@ -24,6 +24,13 @@ if TYPE_CHECKING: # pragma: nocover
def is_body_allowed_for_status_code(status_code: int | str | None) -> bool:
"""
Return ``True`` if the given HTTP status code allows a response body.
Status codes 1xx, 204, 205, and 304 do not allow a body per the HTTP
specification. Wildcard codes (``"1XX"``-``"5XX"``) and ``"default"``
always return ``True``.
"""
if status_code is None:
return True
# Ref: https://github.com/OAI/OpenAPI-Specification/blob/main/versions/3.1.0.md#patterned-fields-1
@ -41,6 +48,7 @@ def is_body_allowed_for_status_code(status_code: int | str | None) -> bool:
def get_path_param_names(path: str) -> set[str]:
"""Extract the set of path parameter names from a URL path template."""
return set(re.findall("{(.*?)}", path))
@ -63,6 +71,13 @@ def create_model_field(
alias: str | None = None,
mode: Literal["validation", "serialization"] = "validation",
) -> ModelField:
"""
Create a Pydantic ``ModelField`` for use in request/response validation.
Raises ``FastAPIError`` if the given type cannot produce a valid Pydantic
schema, and ``PydanticV1NotSupportedError`` if a ``pydantic.v1`` model is
passed.
"""
if annotation_is_pydantic_v1(type_):
raise PydanticV1NotSupportedError(
"pydantic.v1 models are no longer supported by FastAPI."
@ -93,6 +108,7 @@ def generate_operation_id_for_path(
def generate_unique_id(route: "APIRoute") -> str:
"""Generate a unique OpenAPI ``operationId`` for the given route."""
operation_id = f"{route.name}{route.path_format}"
operation_id = re.sub(r"\W", "_", operation_id)
assert route.methods
@ -101,6 +117,12 @@ def generate_unique_id(route: "APIRoute") -> str:
def deep_dict_update(main_dict: dict[Any, Any], update_dict: dict[Any, Any]) -> None:
"""
Recursively merge ``update_dict`` into ``main_dict`` in place.
Nested dicts are merged recursively and nested lists are concatenated;
all other values are overwritten.
"""
for key, value in update_dict.items():
if (
key in main_dict

Loading…
Cancel
Save