Browse Source

🎨 Auto format

pull/15584/head
pre-commit-ci-lite[bot] 2 months ago
committed by GitHub
parent
commit
30ef0e833a
No known key found for this signature in database GPG Key ID: B5690EEEBB952194
  1. 70
      fastapi/_compat/v2.py
  2. 12
      fastapi/dependencies/utils.py
  3. 7
      fastapi/routing.py

70
fastapi/_compat/v2.py

@ -516,43 +516,43 @@ def _regenerate_error_with_loc(
) -> list[dict[str, Any]]:
updated_loc_errors: list[Any] = []
for err in errors:
if decoded_value is not Undefined:
new_err = {**err, "loc": loc_prefix + err.get("loc", ())}
# If we are validating a Body with multiple fields, Pydantic's
# "input" might be just the value of one field.
# But when we decode the whole body, we might need to drill down.
# However, for tutorials, usually "input" matches the decoded value
# if the error is at the top level of the body.
# Let's try to match Pydantic's behavior more closely but with
# decoded values.
curr_input = decoded_value
# If the error is inside the body, try to find the specific input
# that caused it, based on the relative location from the body root.
# loc_prefix is usually ('body',)
rel_loc = err.get("loc", ())
for path_item in rel_loc:
if path_item == "[key]":
# For dict key errors, Pydantic includes "[key]" in the loc.
# The "input" should be the key itself, which was the previous
# path_item.
break
try:
if isinstance(curr_input, (dict, list)):
curr_input = curr_input[path_item] # type: ignore[index]
else:
break
except (KeyError, IndexError, TypeError):
if decoded_value is not Undefined:
new_err = {**err, "loc": loc_prefix + err.get("loc", ())}
# If we are validating a Body with multiple fields, Pydantic's
# "input" might be just the value of one field.
# But when we decode the whole body, we might need to drill down.
# However, for tutorials, usually "input" matches the decoded value
# if the error is at the top level of the body.
# Let's try to match Pydantic's behavior more closely but with
# decoded values.
curr_input = decoded_value
# If the error is inside the body, try to find the specific input
# that caused it, based on the relative location from the body root.
# loc_prefix is usually ('body',)
rel_loc = err.get("loc", ())
for path_item in rel_loc:
if path_item == "[key]":
# For dict key errors, Pydantic includes "[key]" in the loc.
# The "input" should be the key itself, which was the previous
# path_item.
break
try:
if isinstance(curr_input, (dict, list)):
curr_input = curr_input[path_item] # type: ignore[index]
else:
break
except (KeyError, IndexError, TypeError):
break
# If it's a key error, the input is the key which is the last path item before "[key]"
if rel_loc and rel_loc[-1] == "[key]":
new_err["input"] = rel_loc[-2]
else:
new_err["input"] = curr_input
if new_err.get("msg") == "Input should be a valid array":
new_err["msg"] = "Input should be a valid list"
# If it's a key error, the input is the key which is the last path item before "[key]"
if rel_loc and rel_loc[-1] == "[key]":
new_err["input"] = rel_loc[-2]
else:
new_err = {**err, "loc": loc_prefix + err.get("loc", ())}
updated_loc_errors.append(new_err)
new_err["input"] = curr_input
if new_err.get("msg") == "Input should be a valid array":
new_err["msg"] = "Input should be a valid list"
else:
new_err = {**err, "loc": loc_prefix + err.get("loc", ())}
updated_loc_errors.append(new_err)
return updated_loc_errors

12
fastapi/dependencies/utils.py

@ -15,6 +15,7 @@ from collections.abc import (
from contextlib import AsyncExitStack, contextmanager
from copy import copy, deepcopy
from dataclasses import dataclass
from functools import lru_cache
from typing import (
Annotated,
Any,
@ -64,11 +65,10 @@ from fastapi.utils import (
create_model_field,
get_path_param_names,
)
from pydantic import BaseModel, Json, TypeAdapter, create_model, ValidationError
from pydantic import BaseModel, Json, TypeAdapter, ValidationError, create_model
from pydantic.fields import FieldInfo
from starlette.background import BackgroundTasks as StarletteBackgroundTasks
from starlette.concurrency import run_in_threadpool
from functools import lru_cache
from starlette.datastructures import (
FormData,
Headers,
@ -760,9 +760,12 @@ def _validate_value_with_model_field(
# If it's a scalar and we have bytes, we MUST decode it first because Pydantic's
# validate_python doesn't handle JSON-encoded scalar bytes (like b'"-1"')
if isinstance(value, bytes) and field_annotation_is_scalar(field.field_info.annotation):
if isinstance(value, bytes) and field_annotation_is_scalar(
field.field_info.annotation
):
try:
import json
value = json.loads(value)
except json.JSONDecodeError:
pass
@ -899,7 +902,7 @@ def request_params_to_args(
# model_dump might convert inner models, which we don't want (FastAPI keeps them as objects)
values.update(validated_data.__dict__)
else:
values.update(validated_data) # type: ignore
values.update(validated_data) # type: ignore
except ValidationError as exc:
field_in = fields[0].field_info.in_.value
@ -1051,6 +1054,7 @@ async def request_body_to_args(
if isinstance(received_body, bytes):
try:
import json
body_to_process = json.loads(received_body)
except json.JSONDecodeError as e:
return values, [

7
fastapi/routing.py

@ -35,7 +35,6 @@ from anyio.abc import ObjectReceiveStream
from fastapi import params
from fastapi._compat import (
ModelField,
Undefined,
lenient_issubclass,
)
from fastapi.datastructures import Default, DefaultPlaceholder
@ -73,6 +72,7 @@ from fastapi.utils import (
get_value_or_default,
is_body_allowed_for_status_code,
)
from pydantic import TypeAdapter
from starlette import routing
from starlette._exception_handler import wrap_app_handling_exceptions
from starlette._utils import is_async_callable
@ -91,7 +91,6 @@ from starlette.routing import Mount as Mount # noqa
from starlette.types import AppType, ASGIApp, Lifespan, Receive, Scope, Send
from starlette.websockets import WebSocket
from typing_extensions import deprecated
from pydantic import TypeAdapter
_any_type_adapter = TypeAdapter(Any)
@ -515,7 +514,9 @@ def get_request_handler(
if hasattr(item.data, "model_dump_json"):
data_str = item.data.model_dump_json()
else:
data_str = _any_type_adapter.dump_json(item.data).decode("utf-8")
data_str = _any_type_adapter.dump_json(
item.data
).decode("utf-8")
else:
data_str = None
return format_sse_event(

Loading…
Cancel
Save