Browse Source

🎨 [pre-commit.ci] Auto format from pre-commit.com hooks

pull/14200/head
pre-commit-ci[bot] 9 months ago
parent
commit
3c0bb4c69e
  1. 8
      fastapi/_compat/__init__.py
  2. 24
      fastapi/_compat/_v1_params.py
  3. 13
      fastapi/_compat/lazy_import.py
  4. 70
      fastapi/_compat/main.py
  5. 6
      fastapi/_compat/shared.py
  6. 120
      fastapi/_compat/v1.py
  7. 3
      fastapi/_compat/v2.py
  8. 12
      fastapi/dependencies/utils.py
  9. 5
      fastapi/encoders.py
  10. 2
      fastapi/routing.py
  11. 8
      fastapi/utils.py
  12. 141
      tests/test_pydantic_v2_first_compat.py

8
fastapi/_compat/__init__.py

@ -13,7 +13,6 @@ from typing import Any
# Import the v1 proxy module - this provides lazy loading and controlled warnings
# Don't import at module level to avoid warnings
# Import legacy compatibility symbols for backward compatibility
from .main import BaseConfig as BaseConfig
from .main import PydanticSchemaGenerationError as PydanticSchemaGenerationError
@ -71,17 +70,22 @@ CoreSchema = Any
GetJsonSchemaHandler = Any
JsonSchemaValue = dict[str, Any]
def _normalize_errors(errors):
from importlib import import_module
v1 = import_module("fastapi._compat.v1") # proxy lazy
return v1._normalize_errors(errors)
# Make v1 available as an attribute
def __getattr__(name: str):
if name == "v1":
# Import directly to avoid recursion
import importlib
return importlib.import_module("fastapi._compat.v1")
raise AttributeError(f"module 'fastapi._compat' has no attribute '{name}'")
# No __all__ defined - exports everything implicitly for backward compatibility
# No __all__ defined - exports everything implicitly for backward compatibility

24
fastapi/_compat/_v1_params.py

@ -10,32 +10,52 @@ Used internally by FastAPI for backward compatibility.
"""
from __future__ import annotations
from typing import Any
_SENTINEL = object()
def _v1() -> Any:
"""Lazy import of v1 module to avoid warnings."""
from . import v1 # lazy proxy; só avisa/erra se realmente usar v1
return v1
class _BaseParam:
"""Wrapper mínimo que delega para v1.FieldInfo sem importar v1 no import-time."""
def __init__(self, default: Any = _SENTINEL, **kwargs: Any):
v1 = _v1()
if default is _SENTINEL:
default = getattr(v1, 'Undefined', None)
default = getattr(v1, "Undefined", None)
self._fi = v1.FieldInfo(default=default, **kwargs)
def __getattr__(self, name: str) -> Any:
return getattr(self._fi, name)
# Tipos usados nos isinstance() do core
class Param(_BaseParam): ...
class Body(_BaseParam): ...
class Form(Body): ...
class File(Form): ...
class Path(Param): ...
class Query(Param): ...
class Header(Param): ...
class Cookie(Param): ...
class Cookie(Param): ...

13
fastapi/_compat/lazy_import.py

@ -5,10 +5,12 @@ Centralized lazy import helpers for v1 compatibility.
"""
from __future__ import annotations
import sys
from typing import Any, Optional, TypeVar
T = TypeVar('T')
T = TypeVar("T")
def get_v1_if_loaded() -> Optional[Any]:
"""
@ -17,20 +19,25 @@ def get_v1_if_loaded() -> Optional[Any]:
"""
if "pydantic.v1" in sys.modules:
import pydantic.v1
return pydantic.v1
return None
def with_v1_guard(func):
"""
Decorator that only executes function if pydantic.v1 is loaded.
"""
def wrapper(*args, **kwargs):
v1 = get_v1_if_loaded()
if v1 is not None:
return func(v1, *args, **kwargs)
return None
return wrapper
def v1_isinstance(obj: Any, v1_class: str) -> bool:
"""
Check isinstance with v1 class only if v1 is loaded.
@ -42,6 +49,7 @@ def v1_isinstance(obj: Any, v1_class: str) -> bool:
return isinstance(obj, cls)
return False
def v1_lenient_issubclass(cls: Any, v1_class: str) -> bool:
"""
Check lenient_issubclass with v1 class only if v1 is loaded.
@ -51,9 +59,11 @@ def v1_lenient_issubclass(cls: Any, v1_class: str) -> bool:
v1_cls = getattr(v1, v1_class, None)
if v1_cls is not None:
from .shared import lenient_issubclass
return lenient_issubclass(cls, v1_cls)
return False
def v1_call_method(method_name: str, *args, **kwargs) -> Any:
"""
Call a v1 method only if v1 is loaded.
@ -65,6 +75,7 @@ def v1_call_method(method_name: str, *args, **kwargs) -> Any:
return method(*args, **kwargs)
return None
def v1_get_attr(attr_name: str) -> Any:
"""
Get v1 attribute only if v1 is loaded.

70
fastapi/_compat/main.py

@ -9,8 +9,12 @@ from typing import (
Type,
)
from fastapi._compat.lazy_import import (
get_v1_if_loaded,
v1_isinstance,
v1_lenient_issubclass,
)
from fastapi._compat.shared import PYDANTIC_V2
from fastapi._compat.lazy_import import get_v1_if_loaded, v1_isinstance, v1_lenient_issubclass
from fastapi.types import ModelNameMap
from pydantic import BaseModel
from typing_extensions import Literal
@ -82,6 +86,7 @@ def get_cached_model_fields(model: Type[BaseModel]) -> List[ModelField]:
return v1.get_model_fields(model)
else:
from . import v2
return v2.get_model_fields(model) # type: ignore[return-value]
@ -90,6 +95,7 @@ def _is_undefined(value: object) -> bool:
return True
elif PYDANTIC_V2:
from pydantic_core import PydanticUndefined
return value is PydanticUndefined
else:
return False
@ -101,6 +107,7 @@ def _get_model_config(model: BaseModel) -> Any:
return v1._get_model_config(model)
elif PYDANTIC_V2:
from . import v2
return v2._get_model_config(model)
else:
return getattr(model, "__config__", None)
@ -114,6 +121,7 @@ def _model_dump(
return v1._model_dump(model, mode=mode, **kwargs)
if PYDANTIC_V2:
from . import v2
return v2._model_dump(model, mode=mode, **kwargs)
else:
return model.dict(**kwargs)
@ -124,6 +132,7 @@ def _is_error_wrapper(exc: Exception) -> bool:
return True
elif PYDANTIC_V2:
from . import v2
return v2._is_error_wrapper(exc)
else:
return False
@ -135,17 +144,17 @@ def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo:
return v1.copy_field_info(field_info=field_info, annotation=annotation)
else:
from . import v2
return v2.copy_field_info(field_info=field_info, annotation=annotation)
def create_body_model(
*, fields: List[ModelField], model_name: str
) -> Type[BaseModel]:
def create_body_model(*, fields: List[ModelField], model_name: str) -> Type[BaseModel]:
if fields and v1_isinstance(fields[0], "ModelField"):
v1 = get_v1_if_loaded()
return v1.create_body_model(fields=fields, model_name=model_name)
else:
from . import v2
return v2.create_body_model(fields=fields, model_name=model_name)
@ -159,6 +168,7 @@ def get_annotation_from_field_info(
)
else:
from . import v2
return v2.get_annotation_from_field_info(
annotation=annotation, field_info=field_info, field_name=field_name
)
@ -170,6 +180,7 @@ def is_bytes_field(field: ModelField) -> bool:
return v1.is_bytes_field(field)
else:
from . import v2
return v2.is_bytes_field(field)
@ -179,6 +190,7 @@ def is_bytes_sequence_field(field: ModelField) -> bool:
return v1.is_bytes_sequence_field(field)
else:
from . import v2
return v2.is_bytes_sequence_field(field)
@ -188,6 +200,7 @@ def is_scalar_field(field: ModelField) -> bool:
return v1.is_scalar_field(field)
else:
from . import v2
return v2.is_scalar_field(field)
@ -197,6 +210,7 @@ def is_scalar_sequence_field(field: ModelField) -> bool:
return v1.is_scalar_sequence_field(field)
else:
from . import v2
return v2.is_scalar_sequence_field(field)
@ -206,6 +220,7 @@ def is_sequence_field(field: ModelField) -> bool:
return v1.is_sequence_field(field)
else:
from . import v2
return v2.is_sequence_field(field)
@ -215,18 +230,30 @@ def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]:
return v1.serialize_sequence_value(field=field, value=value)
else:
from . import v2
return v2.serialize_sequence_value(field=field, value=value)
def get_compat_model_name_map(fields: List[ModelField]) -> ModelNameMap:
v1 = get_v1_if_loaded()
v1_model_fields = [field for field in fields if v1_isinstance(field, "ModelField")] if v1 else []
v1_flat_models = v1.get_flat_models_from_fields(v1_model_fields, known_models=set()) if v1 and v1_model_fields else set()
v1_model_fields = (
[field for field in fields if v1_isinstance(field, "ModelField")] if v1 else []
)
v1_flat_models = (
v1.get_flat_models_from_fields(v1_model_fields, known_models=set())
if v1 and v1_model_fields
else set()
)
all_flat_models = v1_flat_models
if PYDANTIC_V2:
from . import v2
v2_model_fields = [field for field in fields if not v1_isinstance(field, "ModelField")]
v2_flat_models = v2.get_flat_models_from_fields(v2_model_fields, known_models=set())
v2_model_fields = [
field for field in fields if not v1_isinstance(field, "ModelField")
]
v2_flat_models = v2.get_flat_models_from_fields(
v2_model_fields, known_models=set()
)
all_flat_models = v1_flat_models | v2_flat_models
model_name_map = v2.get_model_name_map(all_flat_models)
return model_name_map
@ -244,7 +271,9 @@ def get_definitions(
Dict[str, Dict[str, Any]],
]:
v1 = get_v1_if_loaded()
v1_fields = [field for field in fields if v1_isinstance(field, "ModelField")] if v1 else []
v1_fields = (
[field for field in fields if v1_isinstance(field, "ModelField")] if v1 else []
)
if v1_fields and v1:
v1_field_maps, v1_definitions = v1.get_definitions(
fields=v1_fields,
@ -252,11 +281,16 @@ def get_definitions(
separate_input_output_schemas=separate_input_output_schemas,
)
else:
v1_field_maps: Dict[Tuple[ModelField, Literal["validation", "serialization"]], Dict[str, Any]] = {}
v1_field_maps: Dict[
Tuple[ModelField, Literal["validation", "serialization"]], Dict[str, Any]
] = {}
v1_definitions: Dict[str, Dict[str, Any]] = {}
if PYDANTIC_V2:
from . import v2
v2_fields = [field for field in fields if not v1_isinstance(field, "ModelField")]
v2_fields = [
field for field in fields if not v1_isinstance(field, "ModelField")
]
v2_field_maps, v2_definitions = v2.get_definitions(
fields=v2_fields,
model_name_map=model_name_map,
@ -287,6 +321,7 @@ def get_schema_from_model_field(
)
else:
from . import v2
return v2.get_schema_from_model_field(
field=field,
model_name_map=model_name_map,
@ -300,6 +335,7 @@ def _is_model_field(value: Any) -> bool:
return True
elif PYDANTIC_V2:
from . import v2
return v2._is_model_field(value)
else:
return False
@ -310,6 +346,7 @@ def _is_model_class(value: Any) -> bool:
return True
elif PYDANTIC_V2:
from . import v2
return v2._is_model_class(value)
else:
return False
@ -321,12 +358,16 @@ def get_missing_field_error(loc: Tuple[str, ...], field: ModelField) -> Dict[str
return v1.get_missing_field_error(loc=loc, field=field)
else:
from . import v2
return v2.get_missing_field_error(loc=loc, field=field)
def evaluate_forwardref(type_: Any, globalns: Dict[str, Any], localns: Dict[str, Any]) -> Any:
def evaluate_forwardref(
type_: Any, globalns: Dict[str, Any], localns: Dict[str, Any]
) -> Any:
if PYDANTIC_V2:
from . import v2
return v2.evaluate_forwardref(type_, globalns, localns)
else:
v1 = get_v1_if_loaded()
@ -349,7 +390,9 @@ def with_info_plain_validator_function(
return pydantic_core_with_info(func)
else:
v1 = get_v1_if_loaded()
return v1.with_info_plain_validator_function(func=func, info_argname=info_argname)
return v1.with_info_plain_validator_function(
func=func, info_argname=info_argname
)
def _model_rebuild(model) -> None:
@ -358,6 +401,7 @@ def _model_rebuild(model) -> None:
v1._model_rebuild(model)
elif PYDANTIC_V2:
from . import v2
v2._model_rebuild(model)
else:
model.update_forward_refs()

6
fastapi/_compat/shared.py

@ -100,6 +100,7 @@ def _annotation_is_complex(annotation: Union[Type[Any], None]) -> bool:
if "pydantic.v1" in sys.modules:
# só agora toca v1 (já foi usado pelo app)
from fastapi._compat import v1 as _v1
types_tuple += (_v1.BaseModel,)
return (
lenient_issubclass(annotation, types_tuple)
@ -202,11 +203,14 @@ def annotation_is_pydantic_v1(annotation: Any) -> bool:
if "pydantic.v1" not in sys.modules:
return False
from fastapi._compat import v1 as _v1
if lenient_issubclass(annotation, _v1.BaseModel):
return True
origin = get_origin(annotation)
if origin in (Union, UnionType):
return any(lenient_issubclass(arg, _v1.BaseModel) for arg in get_args(annotation))
return any(
lenient_issubclass(arg, _v1.BaseModel) for arg in get_args(annotation)
)
if field_annotation_is_sequence(annotation):
return any(annotation_is_pydantic_v1(sa) for sa in get_args(annotation))
return False

120
fastapi/_compat/v1.py

@ -16,6 +16,7 @@ import sys
import warnings
from copy import copy as _copy
from typing import Any, Dict, List, Sequence, Tuple
from typing_extensions import Literal
# Never import pydantic.v1 at import-time of this file.
@ -27,6 +28,7 @@ RequiredParam = Ellipsis
_pv1 = None
_warned = False
def _load() -> Any:
global _pv1, _warned
if _pv1 is not None:
@ -41,6 +43,7 @@ def _load() -> Any:
_pv1 = importlib.import_module("pydantic.v1")
return _pv1
def __getattr__(name: str) -> Any:
if name == "RequiredParam":
return Ellipsis
@ -49,94 +52,173 @@ def __getattr__(name: str) -> Any:
if hasattr(mod, name):
return getattr(mod, name)
# tenta submódulos comuns
for sub in ("fields","schema","networks","types","color","class_validators",
"error_wrappers","errors","typing","utils"):
for sub in (
"fields",
"schema",
"networks",
"types",
"color",
"class_validators",
"error_wrappers",
"errors",
"typing",
"utils",
):
submod = getattr(mod, sub, None)
if submod and hasattr(submod, name):
return getattr(submod, name)
raise AttributeError(name)
# ---------- Wrappers usados pelo core FastAPI (mínimos) ----------
def _normalize_errors(errors: Sequence[Any]) -> List[Dict[str, Any]]:
pv1 = _load()
RequestErrorModel = pv1.create_model("Request")
out: List[Any] = []
for err in errors:
if isinstance(err, pv1.error_wrappers.ErrorWrapper):
out.extend(pv1.ValidationError(errors=[err], model=RequestErrorModel).errors())
out.extend(
pv1.ValidationError(errors=[err], model=RequestErrorModel).errors()
)
elif isinstance(err, list):
out.extend(_normalize_errors(err))
else:
out.append(err)
return out
def _regenerate_error_with_loc(*, errors: Sequence[Any], loc_prefix: Tuple[Any, ...]) -> List[Dict[str, Any]]:
return [{**e, "loc": loc_prefix + tuple(e.get("loc", ())) } for e in _normalize_errors(errors)]
def _regenerate_error_with_loc(
*, errors: Sequence[Any], loc_prefix: Tuple[Any, ...]
) -> List[Dict[str, Any]]:
return [
{**e, "loc": loc_prefix + tuple(e.get("loc", ()))}
for e in _normalize_errors(errors)
]
def _model_rebuild(model) -> None:
model.update_forward_refs()
def _model_dump(model, mode: Literal["json","python"]="json", **kwargs: Any) -> Any:
def _model_dump(model, mode: Literal["json", "python"] = "json", **kwargs: Any) -> Any:
return model.dict(**kwargs)
def _get_model_config(model) -> Any:
return getattr(model, "__config__", None)
def get_schema_from_model_field(*, field, model_name_map, field_mapping: Dict[Tuple[Any, Literal["validation","serialization"]], Dict[str, Any]], separate_input_output_schemas: bool=True) -> Dict[str, Any]:
def get_schema_from_model_field(
*,
field,
model_name_map,
field_mapping: Dict[
Tuple[Any, Literal["validation", "serialization"]], Dict[str, Any]
],
separate_input_output_schemas: bool = True,
) -> Dict[str, Any]:
schema = _load().schema
ref = "#/components/schemas"
return schema.field_schema(field, model_name_map=model_name_map, ref_prefix=ref)[0]
def get_definitions(*, fields: List[Any], model_name_map, separate_input_output_schemas: bool=True):
def get_definitions(
*, fields: List[Any], model_name_map, separate_input_output_schemas: bool = True
):
schema = _load().schema
models = schema.get_flat_models_from_fields(fields, known_models=set())
definitions: Dict[str, Dict[str, Any]] = {}
for m in models:
m_schema, m_defs, _ = schema.model_process_schema(m, model_name_map=model_name_map, ref_prefix="#/components/schemas")
m_schema, m_defs, _ = schema.model_process_schema(
m, model_name_map=model_name_map, ref_prefix="#/components/schemas"
)
definitions.update(m_defs)
definitions[model_name_map[m]] = m_schema
return {}, definitions
def get_model_fields(model) -> List[Any]:
return list(getattr(model, "__fields__", {}).values())
def is_bytes_field(field) -> bool:
return _load().utils.lenient_issubclass(field.type_, bytes)
def is_bytes_sequence_field(field) -> bool:
f = _load().fields
shapes = {f.SHAPE_LIST, f.SHAPE_SET, f.SHAPE_FROZENSET, f.SHAPE_TUPLE, f.SHAPE_SEQUENCE, f.SHAPE_TUPLE_ELLIPSIS}
return field.shape in shapes and _load().utils.lenient_issubclass(field.type_, bytes)
shapes = {
f.SHAPE_LIST,
f.SHAPE_SET,
f.SHAPE_FROZENSET,
f.SHAPE_TUPLE,
f.SHAPE_SEQUENCE,
f.SHAPE_TUPLE_ELLIPSIS,
}
return field.shape in shapes and _load().utils.lenient_issubclass(
field.type_, bytes
)
def is_scalar_field(field) -> bool:
f = _load().fields
pv1 = _load()
return (field.shape == f.SHAPE_SINGLETON
and not pv1.utils.lenient_issubclass(field.type_, pv1.BaseModel)
and not pv1.utils.lenient_issubclass(field.type_, dict))
return (
field.shape == f.SHAPE_SINGLETON
and not pv1.utils.lenient_issubclass(field.type_, pv1.BaseModel)
and not pv1.utils.lenient_issubclass(field.type_, dict)
)
def is_sequence_field(field) -> bool:
f = _load().fields
return field.shape in {f.SHAPE_LIST, f.SHAPE_SET, f.SHAPE_FROZENSET, f.SHAPE_TUPLE, f.SHAPE_SEQUENCE, f.SHAPE_TUPLE_ELLIPSIS}
return field.shape in {
f.SHAPE_LIST,
f.SHAPE_SET,
f.SHAPE_FROZENSET,
f.SHAPE_TUPLE,
f.SHAPE_SEQUENCE,
f.SHAPE_TUPLE_ELLIPSIS,
}
def is_scalar_sequence_field(field) -> bool:
f = _load().fields
pv1 = _load()
if field.shape in {f.SHAPE_LIST, f.SHAPE_SET, f.SHAPE_FROZENSET, f.SHAPE_TUPLE, f.SHAPE_SEQUENCE, f.SHAPE_TUPLE_ELLIPSIS}:
return not pv1.utils.lenient_issubclass(field.type_, pv1.BaseModel) and all(is_scalar_field(sf) for sf in (field.sub_fields or []))
if field.shape in {
f.SHAPE_LIST,
f.SHAPE_SET,
f.SHAPE_FROZENSET,
f.SHAPE_TUPLE,
f.SHAPE_SEQUENCE,
f.SHAPE_TUPLE_ELLIPSIS,
}:
return not pv1.utils.lenient_issubclass(field.type_, pv1.BaseModel) and all(
is_scalar_field(sf) for sf in (field.sub_fields or [])
)
return False
def copy_field_info(*, field_info, annotation: Any):
return _copy(field_info)
def serialize_sequence_value(*, field, value):
f = _load().fields
mapping = { f.SHAPE_LIST: list, f.SHAPE_SET: set, f.SHAPE_TUPLE: tuple, f.SHAPE_SEQUENCE: list, f.SHAPE_TUPLE_ELLIPSIS: list }
mapping = {
f.SHAPE_LIST: list,
f.SHAPE_SET: set,
f.SHAPE_TUPLE: tuple,
f.SHAPE_SEQUENCE: list,
f.SHAPE_TUPLE_ELLIPSIS: list,
}
return mapping[field.shape](value)
# Type aliases for backward compatibility
GetJsonSchemaHandler = Any
JsonSchemaValue = dict[str, Any]
CoreSchema = Any
Url = Any
Url = Any

3
fastapi/_compat/v2.py

@ -35,12 +35,15 @@ from pydantic_core import PydanticUndefined, PydanticUndefinedType
from pydantic_core import Url as Url
from typing_extensions import Annotated, Literal, get_args, get_origin
# Lazy import of v1 to avoid warnings
def _get_v1() -> Any:
"""Lazy import of v1 module to avoid warnings."""
from fastapi._compat import v1
return v1
try:
from pydantic_core.core_schema import (
with_info_plain_validator_function as with_info_plain_validator_function,

12
fastapi/dependencies/utils.py

@ -79,12 +79,15 @@ from typing_extensions import Annotated, get_args, get_origin
from .._compat import _v1_params as temp_pydantic_v1_params
# Lazy import of v1 to avoid warnings
def _get_v1():
"""Lazy import of v1 module to avoid warnings."""
from fastapi._compat import v1
return v1
if sys.version_info >= (3, 13): # pragma: no cover
from inspect import iscoroutinefunction
else: # pragma: no cover
@ -404,9 +407,9 @@ def analyze_param(
)
]
if fastapi_specific_annotations:
fastapi_annotation: Union[FieldInfo, _get_v1().FieldInfo, params.Depends, None] = (
fastapi_specific_annotations[-1]
)
fastapi_annotation: Union[
FieldInfo, _get_v1().FieldInfo, params.Depends, None
] = fastapi_specific_annotations[-1]
else:
fastapi_annotation = None
# Set default for Annotated FieldInfo
@ -531,7 +534,8 @@ def analyze_param(
type_=use_annotation_from_field_info,
default=field_info.default,
alias=alias,
required=field_info.default in (RequiredParam, _get_v1().RequiredParam, Undefined),
required=field_info.default
in (RequiredParam, _get_v1().RequiredParam, Undefined),
field_info=field_info,
)
if is_path_param:

5
fastapi/encoders.py

@ -85,12 +85,14 @@ ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = {
AnyUrl: str,
}
def _ensure_v1_encoders_registered() -> None:
"""Register V1 encoders only when needed (lazy loading)."""
# Só registra se pydantic.v1 já estiver carregado (app realmente usou v1)
if "pydantic.v1" not in sys.modules:
return
from fastapi._compat import v1 # agora sim
ENCODERS_BY_TYPE.setdefault(v1.Color, str)
ENCODERS_BY_TYPE.setdefault(v1.NameEmail, str)
ENCODERS_BY_TYPE.setdefault(v1.SecretBytes, str)
@ -217,7 +219,7 @@ def jsonable_encoder(
"""
# Ensure V1 encoders are registered if needed (lazy loading)
_ensure_v1_encoders_registered()
custom_encoder = custom_encoder or {}
if custom_encoder:
if type(obj) in custom_encoder:
@ -236,6 +238,7 @@ def jsonable_encoder(
# Check if it's a v1 model using lazy loading
if "pydantic.v1" in sys.modules:
from fastapi._compat import v1
if isinstance(obj, v1.BaseModel):
encoders = getattr(obj.__config__, "json_encoders", {}) # type: ignore[attr-defined]
if custom_encoder:

2
fastapi/routing.py

@ -25,7 +25,6 @@ from typing import (
)
from fastapi import params
from fastapi._compat import _v1_params as temp_pydantic_v1_params
from fastapi._compat import (
ModelField,
Undefined,
@ -34,6 +33,7 @@ from fastapi._compat import (
_normalize_errors,
lenient_issubclass,
)
from fastapi._compat import _v1_params as temp_pydantic_v1_params
from fastapi.datastructures import Default, DefaultPlaceholder
from fastapi.dependencies.models import Dependant
from fastapi.dependencies.utils import (

8
fastapi/utils.py

@ -27,7 +27,6 @@ from fastapi._compat import (
annotation_is_pydantic_v1,
lenient_issubclass,
)
from fastapi.datastructures import DefaultPlaceholder, DefaultType
from pydantic import BaseModel
from pydantic.fields import FieldInfo
@ -36,12 +35,15 @@ from typing_extensions import Literal
if TYPE_CHECKING: # pragma: nocover
from .routing import APIRoute
# Lazy import of v1 to avoid warnings
def _get_v1() -> Any:
"""Lazy import of v1 module to avoid warnings."""
from fastapi._compat import v1
return v1
# Cache for `create_cloned_field`
_CLONED_TYPES_CACHE: MutableMapping[Type[BaseModel], Type[BaseModel]] = (
WeakKeyDictionary()
@ -158,7 +160,9 @@ def create_cloned_field(
original_type = cast(Type[_get_v1().BaseModel], original_type)
use_type = cloned_types.get(original_type)
if use_type is None:
use_type = _get_v1().create_model(original_type.__name__, __base__=original_type)
use_type = _get_v1().create_model(
original_type.__name__, __base__=original_type
)
cloned_types[original_type] = use_type
for f in original_type.__fields__.values():
use_type.__fields__[f.name] = create_cloned_field(

141
tests/test_pydantic_v2_first_compat.py

@ -12,7 +12,6 @@ This test suite validates that:
import sys
import warnings
from typing import Any
import pytest
from fastapi import FastAPI
@ -27,60 +26,69 @@ class TestV2FirstCompatibility:
"""Test that v2-only usage doesn't trigger warnings on Python 3.14."""
if sys.version_info < (3, 14):
pytest.skip("Python 3.14+ specific test")
# Capture warnings
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter("error", DeprecationWarning)
# These should not trigger any warnings
import fastapi
import fastapi.encoders
from fastapi import FastAPI
_ = FastAPI() # Create app but don't use it
# Test jsonable_encoder with v2 model
class TestModel(BaseModel):
name: str
value: int
model = TestModel(name="test", value=42)
result = jsonable_encoder(model)
assert result == {"name": "test", "value": 42}
assert len(w) == 0, f"Unexpected warnings: {[str(warning.message) for warning in w]}"
assert len(w) == 0, (
f"Unexpected warnings: {[str(warning.message) for warning in w]}"
)
def test_proxy_classes_isinstance(self):
"""Test that proxy classes work correctly with isinstance()."""
import fastapi.temp_pydantic_v1_params as T
from fastapi import params
# Test that proxy classes are available
assert hasattr(T, 'Param')
assert hasattr(T, 'Body')
assert hasattr(T, 'Form')
assert hasattr(T, 'File')
assert hasattr(T, 'Query')
assert hasattr(T, 'Header')
assert hasattr(T, 'Cookie')
assert hasattr(T, 'Path')
assert hasattr(T, "Param")
assert hasattr(T, "Body")
assert hasattr(T, "Form")
assert hasattr(T, "File")
assert hasattr(T, "Query")
assert hasattr(T, "Header")
assert hasattr(T, "Cookie")
assert hasattr(T, "Path")
# Test isinstance() with proxy classes (expect DeprecationWarning)
with warnings.catch_warnings():
warnings.simplefilter("ignore", (DeprecationWarning, UserWarning))
param_instance = T.Param()
assert isinstance(param_instance, T.Param)
# Test that proxy classes are different from v2 params
from fastapi import params as v2_params
assert T.Param is not v2_params.Param
assert T.Body is not v2_params.Body
def test_backward_compatibility_imports(self):
"""Test that existing imports continue to work."""
# Test direct import from temp_pydantic_v1_params
from fastapi.temp_pydantic_v1_params import Body, Query, Form, File, Param, Header, Cookie, Path
from fastapi.temp_pydantic_v1_params import (
Body,
Cookie,
File,
Form,
Header,
Param,
Path,
Query,
)
# Test that classes are available and callable (expect DeprecationWarning)
with warnings.catch_warnings():
warnings.simplefilter("ignore", (DeprecationWarning, UserWarning))
@ -92,50 +100,47 @@ class TestV2FirstCompatibility:
header = Header()
cookie = Cookie()
path = Path()
# Test that they have expected attributes
assert hasattr(body, 'default')
assert hasattr(query, 'default')
assert hasattr(form, 'default')
assert hasattr(file_param, 'default')
assert hasattr(param, 'default')
assert hasattr(header, 'default')
assert hasattr(cookie, 'default')
assert hasattr(path, 'default')
assert hasattr(body, "default")
assert hasattr(query, "default")
assert hasattr(form, "default")
assert hasattr(file_param, "default")
assert hasattr(param, "default")
assert hasattr(header, "default")
assert hasattr(cookie, "default")
assert hasattr(path, "default")
def test_lazy_loading_behavior(self):
"""Test that v1 is only loaded when actually used."""
import sys
# Clear any existing pydantic.v1 from sys.modules
_ = "pydantic.v1" in sys.modules # Check but don't use
# Import FastAPI components
import fastapi
import fastapi.encoders
from fastapi import FastAPI
_ = FastAPI() # Create app but don't use it
# At this point, pydantic.v1 should not be loaded unless it was already loaded
# (we can't test the exact state because it might have been loaded by other tests)
# Test that we can still access v1 proxy
from fastapi._compat import v1
assert v1 is not None
def test_encoders_lazy_registration(self):
"""Test that v1 encoders are registered lazily."""
import sys
# Test with a v2 model (should not trigger v1 encoder registration)
class V2Model(BaseModel):
name: str
model = V2Model(name="test")
result = jsonable_encoder(model)
assert result == {"name": "test"}
# The encoders should work without importing pydantic.v1
# (unless it was already imported by other tests)
@ -145,36 +150,36 @@ class TestV2FirstCompatibility:
with warnings.catch_warnings():
warnings.simplefilter("ignore", (DeprecationWarning, UserWarning))
from fastapi._compat import v1
# Test that v1 proxy has expected attributes
assert hasattr(v1, 'BaseModel')
assert hasattr(v1, 'FieldInfo')
assert hasattr(v1, 'ValidationError')
assert hasattr(v1, "BaseModel")
assert hasattr(v1, "FieldInfo")
assert hasattr(v1, "ValidationError")
# Test that wrapper functions are available
assert hasattr(v1, '_normalize_errors')
assert hasattr(v1, '_model_dump')
assert hasattr(v1, '_model_rebuild')
assert hasattr(v1, "_normalize_errors")
assert hasattr(v1, "_model_dump")
assert hasattr(v1, "_model_rebuild")
def test_strict_mode_environment_variable(self):
"""Test FASTAPI_PYDANTIC_V1_STRICT environment variable behavior."""
import os
# Save original value
original_strict = os.environ.get("FASTAPI_PYDANTIC_V1_STRICT")
try:
# Test with strict mode enabled
os.environ["FASTAPI_PYDANTIC_V1_STRICT"] = "1"
# This should not raise an error unless we actually try to use v1
import fastapi
from fastapi import FastAPI
_ = FastAPI() # Create app but don't use it
# The strict mode only affects actual v1 usage, not imports
assert True
finally:
# Restore original value
if original_strict is not None:
@ -184,24 +189,24 @@ class TestV2FirstCompatibility:
def test_v1_params_composition(self):
"""Test that v1 params use composition instead of inheritance."""
from fastapi._compat._v1_params import Param, Body, Form
from fastapi._compat._v1_params import Body, Form, Param
# Test that they can be instantiated (expect DeprecationWarning)
with warnings.catch_warnings():
warnings.simplefilter("ignore", (DeprecationWarning, UserWarning))
param = Param()
body = Body()
form = Form()
# Test that they delegate to internal v1.FieldInfo
assert hasattr(param, '_fi')
assert hasattr(body, '_fi')
assert hasattr(form, '_fi')
assert hasattr(param, "_fi")
assert hasattr(body, "_fi")
assert hasattr(form, "_fi")
# Test that they have expected interface
assert hasattr(param, 'default')
assert hasattr(body, 'default')
assert hasattr(form, 'default')
assert hasattr(param, "default")
assert hasattr(body, "default")
assert hasattr(form, "default")
if __name__ == "__main__":

Loading…
Cancel
Save