mirror
/
tiangolo.fastapi
mirror of https://github.com/tiangolo/fastapi
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

♻️ Refactor internals to use new _compat utils

pull/14168/head
Sebastián Ramírez 10 months ago
parent
ac47072c6e
commit
22a1ccab6f
5 changed files with 54 additions and 49 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 3
      fastapi/datastructures.py
  2. 15
      fastapi/dependencies/utils.py
  3. 11
      fastapi/encoders.py
  4. 21
      fastapi/openapi/utils.py
  5. 53
      fastapi/utils.py

3
fastapi/datastructures.py
View File

@ -15,7 +15,6 @@ from fastapi._compat import (
CoreSchema,
GetJsonSchemaHandler,
JsonSchemaValue,
with_info_plain_validator_function,
)
from starlette.datastructures import URL as URL # noqa: F401
from starlette.datastructures import Address as Address # noqa: F401
@ -170,6 +169,8 @@ class UploadFile(StarletteUploadFile):
def __get_pydantic_core_schema__(
cls, source: Type[Any], handler: Callable[[Any], CoreSchema]
) -> CoreSchema:
from ._compat.v2 import with_info_plain_validator_function
return with_info_plain_validator_function(cls._validate)

15
fastapi/dependencies/utils.py
View File

@ -23,11 +23,11 @@ import anyio
from fastapi import params
from fastapi._compat import (
PYDANTIC_V2,
ErrorWrapper,
ModelField,
RequiredParam,
Undefined,
_regenerate_error_with_loc,
_is_error_wrapper,
_is_model_class,
copy_field_info,
create_body_model,
evaluate_forwardref,
@ -45,6 +45,7 @@ from fastapi._compat import (
lenient_issubclass,
sequence_types,
serialize_sequence_value,
v1,
value_is_sequence,
)
from fastapi.background import BackgroundTasks
@ -219,7 +220,7 @@ def _get_flat_fields_from_params(fields: List[ModelField]) -> List[ModelField]:
if not fields:
return fields
first_field = fields[0]
if len(fields) == 1 and lenient_issubclass(first_field.type_, BaseModel):
if len(fields) == 1 and _is_model_class(first_field.type_):
fields_to_extract = get_cached_model_fields(first_field.type_)
return fields_to_extract
return fields
@ -498,7 +499,7 @@ def analyze_param(
type_=use_annotation_from_field_info,
default=field_info.default,
alias=alias,
required=field_info.default in (RequiredParam, Undefined),
required=field_info.default in (RequiredParam, v1.RequiredParam, Undefined),
field_info=field_info,
)
if is_path_param:
@ -510,7 +511,7 @@ def analyze_param(
is_scalar_field(field)
or is_scalar_sequence_field(field)
or (
lenient_issubclass(field.type_, BaseModel)
_is_model_class(field.type_)
# For Pydantic v1
and getattr(field, "shape", 1) == 1
)
@ -712,10 +713,10 @@ def _validate_value_with_model_field(
else:
return deepcopy(field.default), []
v_, errors_ = field.validate(value, values, loc=loc)
if isinstance(errors_, ErrorWrapper):
if _is_error_wrapper(errors_):
return None, [errors_]
elif isinstance(errors_, list):
new_errors = _regenerate_error_with_loc(errors=errors_, loc_prefix=())
new_errors = v1._regenerate_error_with_loc(errors=errors_, loc_prefix=())
return None, new_errors
else:
return v_, []

11
fastapi/encoders.py
View File

@ -17,6 +17,7 @@ from types import GeneratorType
from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union
from uuid import UUID
from fastapi._compat import v1
from fastapi.types import IncEx
from pydantic import BaseModel
from pydantic.color import Color
@ -24,7 +25,7 @@ from pydantic.networks import AnyUrl, NameEmail
from pydantic.types import SecretBytes, SecretStr
from typing_extensions import Annotated, Doc
from ._compat import PYDANTIC_V2, UndefinedType, Url, _model_dump
from ._compat import PYDANTIC_V2, Url, _is_undefined, _model_dump
# Taken from Pydantic v1 as is
@ -58,6 +59,7 @@ def decimal_encoder(dec_value: Decimal) -> Union[int, float]:
ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = {
bytes: lambda o: o.decode(),
Color: str,
v1.Color: str,
datetime.date: isoformat,
datetime.datetime: isoformat,
datetime.time: isoformat,
@ -74,14 +76,19 @@ ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = {
IPv6Interface: str,
IPv6Network: str,
NameEmail: str,
v1.NameEmail: str,
Path: str,
Pattern: lambda o: o.pattern,
SecretBytes: str,
v1.SecretBytes: str,
SecretStr: str,
v1.SecretStr: str,
set: list,
UUID: str,
Url: str,
v1.Url: str,
AnyUrl: str,
v1.AnyUrl: str,
}
@ -260,7 +267,7 @@ def jsonable_encoder(
return str(obj)
if isinstance(obj, (str, int, float, type(None))):
return obj
if isinstance(obj, UndefinedType):
if _is_undefined(obj):
return None
if isinstance(obj, dict):
encoded_dict = {}

21
fastapi/openapi/utils.py
View File

@ -5,7 +5,6 @@ from typing import Any, Dict, List, Optional, Sequence, Set, Tuple, Type, Union,
from fastapi import routing
from fastapi._compat import (
GenerateJsonSchema,
JsonSchemaValue,
ModelField,
Undefined,
@ -22,7 +21,7 @@ from fastapi.dependencies.utils import (
get_flat_params,
)
from fastapi.encoders import jsonable_encoder
from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX, REF_TEMPLATE
from fastapi.openapi.constants import METHODS_WITH_BODY, REF_PREFIX
from fastapi.openapi.models import OpenAPI
from fastapi.params import Body, ParamTypes
from fastapi.responses import Response
@ -37,6 +36,8 @@ from starlette.responses import JSONResponse
from starlette.routing import BaseRoute
from typing_extensions import Literal
from .._compat import _is_model_field
validation_error_definition = {
"title": "ValidationError",
"type": "object",
@ -94,7 +95,6 @@ def get_openapi_security_definitions(
def _get_openapi_operation_parameters(
*,
dependant: Dependant,
schema_generator: GenerateJsonSchema,
model_name_map: ModelNameMap,
field_mapping: Dict[
Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue
@ -128,7 +128,6 @@ def _get_openapi_operation_parameters(
continue
param_schema = get_schema_from_model_field(
field=param,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
@ -169,7 +168,6 @@ def _get_openapi_operation_parameters(
def get_openapi_operation_request_body(
*,
body_field: Optional[ModelField],
schema_generator: GenerateJsonSchema,
model_name_map: ModelNameMap,
field_mapping: Dict[
Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue
@ -181,7 +179,6 @@ def get_openapi_operation_request_body(
assert isinstance(body_field, ModelField)
body_schema = get_schema_from_model_field(
field=body_field,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
@ -254,7 +251,6 @@ def get_openapi_path(
*,
route: routing.APIRoute,
operation_ids: Set[str],
schema_generator: GenerateJsonSchema,
model_name_map: ModelNameMap,
field_mapping: Dict[
Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue
@ -287,7 +283,6 @@ def get_openapi_path(
security_schemes.update(security_definitions)
operation_parameters = _get_openapi_operation_parameters(
dependant=route.dependant,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
@ -309,7 +304,6 @@ def get_openapi_path(
if method in METHODS_WITH_BODY:
request_body_oai = get_openapi_operation_request_body(
body_field=route.body_field,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
@ -327,7 +321,6 @@ def get_openapi_path(
) = get_openapi_path(
route=callback,
operation_ids=operation_ids,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
@ -358,7 +351,6 @@ def get_openapi_path(
if route.response_field:
response_schema = get_schema_from_model_field(
field=route.response_field,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
@ -392,7 +384,6 @@ def get_openapi_path(
if field:
additional_field_schema = get_schema_from_model_field(
field=field,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
@ -454,7 +445,7 @@ def get_fields_from_routes(
route, routing.APIRoute
):
if route.body_field:
assert isinstance(route.body_field, ModelField), (
assert _is_model_field(route.body_field), (
"A request body must be a Pydantic Field"
)
body_fields_from_routes.append(route.body_field)
@ -510,10 +501,8 @@ def get_openapi(
operation_ids: Set[str] = set()
all_fields = get_fields_from_routes(list(routes or []) + list(webhooks or []))
model_name_map = get_compat_model_name_map(all_fields)
schema_generator = GenerateJsonSchema(ref_template=REF_TEMPLATE)
field_mapping, definitions = get_definitions(
fields=all_fields,
schema_generator=schema_generator,
model_name_map=model_name_map,
separate_input_output_schemas=separate_input_output_schemas,
)
@ -522,7 +511,6 @@ def get_openapi(
result = get_openapi_path(
route=route,
operation_ids=operation_ids,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,
@ -542,7 +530,6 @@ def get_openapi(
result = get_openapi_path(
route=webhook,
operation_ids=operation_ids,
schema_generator=schema_generator,
model_name_map=model_name_map,
field_mapping=field_mapping,
separate_input_output_schemas=separate_input_output_schemas,

53
fastapi/utils.py
View File

@ -24,6 +24,7 @@ from fastapi._compat import (
UndefinedType,
Validator,
lenient_issubclass,
v1,
)
from fastapi.datastructures import DefaultPlaceholder, DefaultType
from pydantic import BaseModel, create_model
@ -60,28 +61,33 @@ def get_path_param_names(path: str) -> Set[str]:
return set(re.findall("{(.*?)}", path))
_invalid_args_message = (
"Invalid args for response field! Hint: "
"check that {type_} is a valid Pydantic field type. "
"If you are using a return type annotation that is not a valid Pydantic "
"field (e.g. Union[Response, dict, None]) you can disable generating the "
"response model from the type annotation with the path operation decorator "
"parameter response_model=None. Read more: "
"https://fastapi.tiangolo.com/tutorial/response-model/"
)
def create_model_field(
name: str,
type_: Any,
class_validators: Optional[Dict[str, Validator]] = None,
default: Optional[Any] = Undefined,
required: Union[bool, UndefinedType] = Undefined,
model_config: Type[BaseConfig] = BaseConfig,
model_config: Type[BaseConfig] | None = None,
field_info: Optional[FieldInfo] = None,
alias: Optional[str] = None,
mode: Literal["validation", "serialization"] = "validation",
) -> ModelField:
class_validators = class_validators or {}
if PYDANTIC_V2:
field_info = field_info or FieldInfo(
annotation=type_, default=default, alias=alias
)
else:
field_info = field_info or FieldInfo()
kwargs = {"name": name, "field_info": field_info}
if PYDANTIC_V2:
kwargs.update({"mode": mode})
else:
if lenient_issubclass(type_, v1.BaseModel):
model_config = v1.BaseConfig
field_info = field_info or v1.FieldInfo()
kwargs.update(
{
"type_": type_,
@ -92,18 +98,21 @@ def create_model_field(
"alias": alias,
}
)
try:
return ModelField(**kwargs) # type: ignore[arg-type]
except (RuntimeError, PydanticSchemaGenerationError):
raise fastapi.exceptions.FastAPIError(
"Invalid args for response field! Hint: "
f"check that {type_} is a valid Pydantic field type. "
"If you are using a return type annotation that is not a valid Pydantic "
"field (e.g. Union[Response, dict, None]) you can disable generating the "
"response model from the type annotation with the path operation decorator "
"parameter response_model=None. Read more: "
"https://fastapi.tiangolo.com/tutorial/response-model/"
) from None
try:
return v1.ModelField(**kwargs) # type: ignore[arg-type]
except RuntimeError:
raise fastapi.exceptions.FastAPIError(_invalid_args_message) from None
elif PYDANTIC_V2:
from ._compat import v2
field_info = field_info or FieldInfo(
annotation=type_, default=default, alias=alias
)
kwargs.update({"mode": mode})
try:
return v2.ModelField(**kwargs) # type: ignore[arg-type]
except PydanticSchemaGenerationError:
raise fastapi.exceptions.FastAPIError(_invalid_args_message) from None
def create_cloned_field(

Write Preview
Loading…
Cancel
Save