🔥 Remove more unnecessary Pydantic v1 specific code

fastapi/_compat/__init__.py

@@ -5,25 +5,11 @@ from .main import Undefined as Undefined
 from .main import UndefinedType as UndefinedType
 from .main import Url as Url
 from .main import Validator as Validator
-from .main import _get_model_config as _get_model_config
-from .main import _is_model_class as _is_model_class
-from .main import _is_model_field as _is_model_field
-from .main import _is_undefined as _is_undefined
-from .main import _model_dump as _model_dump
-from .main import copy_field_info as copy_field_info
-from .main import create_body_model as create_body_model
 from .main import evaluate_forwardref as evaluate_forwardref
-from .main import get_annotation_from_field_info as get_annotation_from_field_info
 from .main import get_cached_model_fields as get_cached_model_fields
 from .main import get_compat_model_name_map as get_compat_model_name_map
 from .main import get_missing_field_error as get_missing_field_error
 from .main import get_schema_from_model_field as get_schema_from_model_field
-from .main import is_bytes_field as is_bytes_field
-from .main import is_bytes_sequence_field as is_bytes_sequence_field
-from .main import is_scalar_field as is_scalar_field
-from .main import is_scalar_sequence_field as is_scalar_sequence_field
-from .main import is_sequence_field as is_sequence_field
-from .main import serialize_sequence_value as serialize_sequence_value
 from .main import (
     with_info_plain_validator_function as with_info_plain_validator_function,
 )
@@ -43,4 +29,13 @@ from .shared import lenient_issubclass as lenient_issubclass
 from .shared import sequence_types as sequence_types
 from .shared import value_is_sequence as value_is_sequence
 from .v2 import ModelField as ModelField
+from .v2 import _regenerate_error_with_loc as _regenerate_error_with_loc
+from .v2 import copy_field_info as copy_field_info
+from .v2 import create_body_model as create_body_model
 from .v2 import get_definitions as get_definitions
+from .v2 import is_bytes_field as is_bytes_field
+from .v2 import is_bytes_sequence_field as is_bytes_sequence_field
+from .v2 import is_scalar_field as is_scalar_field
+from .v2 import is_scalar_sequence_field as is_scalar_sequence_field
+from .v2 import is_sequence_field as is_sequence_field
+from .v2 import serialize_sequence_value as serialize_sequence_value

fastapi/_compat/main.py

@@ -1,10 +1,8 @@
-from collections.abc import Sequence
 from functools import lru_cache
 from typing import (
     Any,
 )
 
-from fastapi._compat.shared import lenient_issubclass
 from fastapi.types import ModelNameMap
 from pydantic import BaseModel
 from typing_extensions import Literal
@@ -32,62 +30,6 @@ def get_cached_model_fields(model: type[BaseModel]) -> list[ModelField]:
     return get_model_fields(model)  # type: ignore[return-value]
 
 
-def _is_undefined(value: object) -> bool:
-    return isinstance(value, v2.UndefinedType)
-
-
-def _get_model_config(model: BaseModel) -> Any:
-    return v2._get_model_config(model)
-
-
-def _model_dump(
-    model: BaseModel, mode: Literal["json", "python"] = "json", **kwargs: Any
-) -> Any:
-    return v2._model_dump(model, mode=mode, **kwargs)
-
-
-def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo:
-    return v2.copy_field_info(field_info=field_info, annotation=annotation)
-
-
-def create_body_model(
-    *, fields: Sequence[ModelField], model_name: str
-) -> type[BaseModel]:
-    return v2.create_body_model(fields=fields, model_name=model_name)  # type: ignore[arg-type]
-
-
-def get_annotation_from_field_info(
-    annotation: Any, field_info: FieldInfo, field_name: str
-) -> Any:
-    return v2.get_annotation_from_field_info(
-        annotation=annotation, field_info=field_info, field_name=field_name
-    )
-
-
-def is_bytes_field(field: ModelField) -> bool:
-    return v2.is_bytes_field(field)  # type: ignore[arg-type]
-
-
-def is_bytes_sequence_field(field: ModelField) -> bool:
-    return v2.is_bytes_sequence_field(field)  # type: ignore[arg-type]
-
-
-def is_scalar_field(field: ModelField) -> bool:
-    return v2.is_scalar_field(field)  # type: ignore[arg-type]
-
-
-def is_scalar_sequence_field(field: ModelField) -> bool:
-    return v2.is_scalar_sequence_field(field)  # type: ignore[arg-type]
-
-
-def is_sequence_field(field: ModelField) -> bool:
-    return v2.is_sequence_field(field)  # type: ignore[arg-type]
-
-
-def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]:
-    return v2.serialize_sequence_value(field=field, value=value)  # type: ignore[arg-type]
-
-
 def get_compat_model_name_map(fields: list[ModelField]) -> ModelNameMap:
     all_flat_models = set()
 
@@ -115,11 +57,3 @@ def get_schema_from_model_field(
         field_mapping=field_mapping,  # type: ignore[arg-type]
         separate_input_output_schemas=separate_input_output_schemas,
     )
-
-
-def _is_model_field(value: Any) -> bool:
-    return isinstance(value, v2.ModelField)
-
-
-def _is_model_class(value: Any) -> bool:
-    return lenient_issubclass(value, v2.BaseModel)  # type: ignore[attr-defined]

fastapi/_compat/may_v1.py

@@ -1,12 +0,0 @@
-from collections.abc import Sequence
-from typing import Any, Union
-
-
-def _regenerate_error_with_loc(
-    *, errors: Sequence[Any], loc_prefix: tuple[Union[str, int], ...]
-) -> list[dict[str, Any]]:
-    updated_loc_errors: list[Any] = [
-        {**err, "loc": loc_prefix + err.get("loc", ())} for err in errors
-    ]
-
-    return updated_loc_errors

fastapi/_compat/v2.py

@@ -11,7 +11,7 @@ from typing import (
     cast,
 )
 
-from fastapi._compat import may_v1, shared
+from fastapi._compat import shared
 from fastapi.openapi.constants import REF_TEMPLATE
 from fastapi.types import IncEx, ModelNameMap, UnionType
 from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, create_model
@@ -175,7 +175,7 @@ class ModelField:
                 None,
             )
         except ValidationError as exc:
-            return None, may_v1._regenerate_error_with_loc(
+            return None, _regenerate_error_with_loc(
                 errors=exc.errors(include_url=False), loc_prefix=loc
             )
 
@@ -210,22 +210,6 @@ class ModelField:
         return id(self)
 
 
-def get_annotation_from_field_info(
-    annotation: Any, field_info: FieldInfo, field_name: str
-) -> Any:
-    return annotation
-
-
-def _model_dump(
-    model: BaseModel, mode: Literal["json", "python"] = "json", **kwargs: Any
-) -> Any:
-    return model.model_dump(mode=mode, **kwargs)
-
-
-def _get_model_config(model: BaseModel) -> Any:
-    return model.model_config
-
-
 def _has_computed_fields(field: ModelField) -> bool:
     computed_fields = field._type_adapter.core_schema.get("schema", {}).get(
         "computed_fields", []
@@ -555,3 +539,13 @@ def get_flat_models_from_fields(
     for field in fields:
         get_flat_models_from_field(field, known_models=known_models)
     return known_models
+
+
+def _regenerate_error_with_loc(
+    *, errors: Sequence[Any], loc_prefix: tuple[Union[str, int], ...]
+) -> list[dict[str, Any]]:
+    updated_loc_errors: list[Any] = [
+        {**err, "loc": loc_prefix + err.get("loc", ())} for err in errors
+    ]
+
+    return updated_loc_errors

fastapi/dependencies/utils.py

@@ -21,12 +21,11 @@ from fastapi._compat import (
     ModelField,
     RequiredParam,
     Undefined,
-    _is_model_class,
+    _regenerate_error_with_loc,
     copy_field_info,
     create_body_model,
     evaluate_forwardref,
     field_annotation_is_scalar,
-    get_annotation_from_field_info,
     get_cached_model_fields,
     get_missing_field_error,
     is_bytes_field,
@@ -37,7 +36,6 @@ from fastapi._compat import (
     is_uploadfile_or_nonable_uploadfile_annotation,
     is_uploadfile_sequence_annotation,
     lenient_issubclass,
-    may_v1,
     sequence_types,
     serialize_sequence_value,
     value_is_sequence,
@@ -184,7 +182,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 _is_model_class(first_field.type_):
+    if len(fields) == 1 and lenient_issubclass(first_field.type_, BaseModel):
         fields_to_extract = get_cached_model_fields(first_field.type_)
         return fields_to_extract
     return fields
@@ -492,11 +490,7 @@ def analyze_param(
             and getattr(field_info, "in_", None) is None
         ):
             field_info.in_ = params.ParamTypes.query
-        use_annotation_from_field_info = get_annotation_from_field_info(
-            use_annotation,
-            field_info,
-            param_name,
-        )
+        use_annotation_from_field_info = use_annotation
         if isinstance(field_info, params.Form):
             ensure_multipart_is_installed()
         if not field_info.alias and getattr(field_info, "convert_underscores", None):
@@ -521,7 +515,7 @@ def analyze_param(
                 is_scalar_field(field)
                 or is_scalar_sequence_field(field)
                 or (
-                    _is_model_class(field.type_)
+                    lenient_issubclass(field.type_, BaseModel)
                     # For Pydantic v1
                     and getattr(field, "shape", 1) == 1
                 )
@@ -716,7 +710,7 @@ def _validate_value_with_model_field(
             return deepcopy(field.default), []
     v_, errors_ = field.validate(value, values, loc=loc)
     if isinstance(errors_, list):
-        new_errors = may_v1._regenerate_error_with_loc(errors=errors_, loc_prefix=())
+        new_errors = _regenerate_error_with_loc(errors=errors_, loc_prefix=())
         return None, new_errors
     else:
         return v_, []
@@ -842,7 +836,7 @@ def is_union_of_base_models(field_type: Any) -> bool:
     union_args = get_args(field_type)
 
     for arg in union_args:
-        if not _is_model_class(arg):
+        if not lenient_issubclass(arg, BaseModel):
             return False
 
     return True
@@ -865,7 +859,7 @@ def _should_embed_body_fields(fields: list[ModelField]) -> bool:
     # otherwise it has to be embedded, so that the key value pair can be extracted
     if (
         isinstance(first_field.field_info, params.Form)
-        and not _is_model_class(first_field.type_)
+        and not lenient_issubclass(first_field.type_, BaseModel)
         and not is_union_of_base_models(first_field.type_)
     ):
         return True
@@ -935,7 +929,7 @@ async def request_body_to_args(
 
     if (
         single_not_embedded_field
-        and _is_model_class(first_field.type_)
+        and lenient_issubclass(first_field.type_, BaseModel)
         and isinstance(received_body, FormData)
     ):
         fields_to_extract = get_cached_model_fields(first_field.type_)

fastapi/encoders.py

@@ -24,11 +24,10 @@ from pydantic import BaseModel
 from pydantic.color import Color
 from pydantic.networks import AnyUrl, NameEmail
 from pydantic.types import SecretBytes, SecretStr
+from pydantic_core import PydanticUndefinedType
 
 from ._compat import (
     Url,
-    _is_undefined,
-    _model_dump,
     is_pydantic_v1_model_instance,
 )
 
@@ -224,8 +223,7 @@ def jsonable_encoder(
     if exclude is not None and not isinstance(exclude, (set, dict)):
         exclude = set(exclude)
     if isinstance(obj, BaseModel):
-        obj_dict = _model_dump(
-            obj,  # type: ignore[arg-type]
+        obj_dict = obj.model_dump(
             mode="json",
             include=include,
             exclude=exclude,
@@ -260,7 +258,7 @@ def jsonable_encoder(
         return str(obj)
     if isinstance(obj, (str, int, float, type(None))):
         return obj
-    if _is_undefined(obj):
+    if isinstance(obj, PydanticUndefinedType):
         return None
     if isinstance(obj, dict):
         encoded_dict = {}

fastapi/openapi/utils.py

@@ -38,8 +38,6 @@ 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",
@@ -187,7 +185,7 @@ def get_openapi_operation_request_body(
 ) -> Optional[dict[str, Any]]:
     if not body_field:
         return None
-    assert _is_model_field(body_field)
+    assert isinstance(body_field, ModelField)
     body_schema = get_schema_from_model_field(
         field=body_field,
         model_name_map=model_name_map,
@@ -456,7 +454,7 @@ def get_fields_from_routes(
             route, routing.APIRoute
         ):
             if route.body_field:
-                assert _is_model_field(route.body_field), (
+                assert isinstance(route.body_field, ModelField), (
                     "A request body must be a Pydantic Field"
                 )
                 body_fields_from_routes.append(route.body_field)

tests/test_compat.py

@@ -3,7 +3,6 @@ from typing import Union
 from fastapi import FastAPI, UploadFile
 from fastapi._compat import (
     Undefined,
-    _get_model_config,
     is_uploadfile_sequence_annotation,
 )
 from fastapi._compat.shared import is_bytes_sequence_annotation
@@ -23,23 +22,6 @@ def test_model_field_default_required():
     assert field.default is Undefined
 
 
-def test_is_model_field():
-    # For coverage
-    from fastapi._compat import _is_model_field
-
-    assert not _is_model_field(str)
-
-
-def test_get_model_config():
-    # For coverage in Pydantic v2
-    class Foo(BaseModel):
-        model_config = ConfigDict(from_attributes=True)
-
-    foo = Foo()
-    config = _get_model_config(foo)
-    assert config == {"from_attributes": True}
-
-
 def test_complex():
     app = FastAPI()