From 9bcbafa55b06e6f4c691a83be39e91027d0a85d8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Roberto=20Bert=C3=B3?= <463349+robertoberto@users.noreply.github.com> Date: Sat, 18 Oct 2025 00:51:02 +0000 Subject: [PATCH] compat: v2-first lazy v1 proxy; py3.14-safe; on-demand v1 encoders MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _compat/v1.py: proxy PEP-562 (lazy), STRICT env, wrappers mínimos - encoders.py: remove v1.* do dict global; registra v1 on-demand - _compat/shared.py: guards com sys.modules; sem import v1 no topo - _compat/__init__.py: __getattr__('v1') + _normalize_errors forward on-demand - _compat/_v1_params.py: shim privado por composição (lazy) - temp_pydantic_v1_params.py: proxy compat p/ zero breaking change - _compat/main.py: helpers v1 centralizados via lazy_import - remove scripts auxiliares e helper duplicado ✅ Smoke tests: v2 (3.14) sem warnings; v1 (3.14) warning único / STRICT→erro --- fastapi/_compat/__init__.py | 45 +- fastapi/_compat/_v1_params.py | 40 ++ fastapi/_compat/lazy_import.py | 74 +++ fastapi/_compat/main.py | 265 +++++---- fastapi/_compat/shared.py | 23 +- fastapi/_compat/v1.py | 432 ++++---------- fastapi/_compat/v2.py | 10 +- fastapi/dependencies/utils.py | 26 +- fastapi/encoders.py | 37 +- fastapi/routing.py | 3 +- fastapi/temp_pydantic_v1_params.py | 759 ++----------------------- fastapi/utils.py | 23 +- tests/test_pydantic_v2_first_compat.py | 208 +++++++ 13 files changed, 754 insertions(+), 1191 deletions(-) create mode 100644 fastapi/_compat/_v1_params.py create mode 100644 fastapi/_compat/lazy_import.py create mode 100644 tests/test_pydantic_v2_first_compat.py diff --git a/fastapi/_compat/__init__.py b/fastapi/_compat/__init__.py index b2ae5adc7..76779268d 100644 --- a/fastapi/_compat/__init__.py +++ b/fastapi/_compat/__init__.py @@ -1,3 +1,20 @@ +# path: fastapi/_compat/__init__.py + +""" +FastAPI Pydantic compatibility layer. + +This module provides a v2-first compatibility layer that automatically +detects the Pydantic version and provides the appropriate symbols. +""" + +from __future__ import annotations + +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 from .main import RequiredParam as RequiredParam @@ -44,7 +61,27 @@ from .shared import ( 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 .v1 import CoreSchema as CoreSchema -from .v1 import GetJsonSchemaHandler as GetJsonSchemaHandler -from .v1 import JsonSchemaValue as JsonSchemaValue -from .v1 import _normalize_errors as _normalize_errors + +# V1 symbols are available via the v1 proxy module +# Access them directly: from fastapi._compat import v1; v1.CoreSchema +# This avoids import-time access and warnings + +# Export V1 symbols as Any to avoid import-time access +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 newline at end of file diff --git a/fastapi/_compat/_v1_params.py b/fastapi/_compat/_v1_params.py new file mode 100644 index 000000000..92b889356 --- /dev/null +++ b/fastapi/_compat/_v1_params.py @@ -0,0 +1,40 @@ +# path: fastapi/_compat/_v1_params.py + +""" +Internal v1-params shim. Will be removed when v1 is dropped. + +This module provides compatibility shims for Pydantic v1 FieldInfo/Param types +to support isinstance() checks without importing pydantic.v1 at import-time. +Used internally by FastAPI for backward compatibility. +""" + +from __future__ import annotations +from typing import Any + +_SENTINEL = object() + +def _v1(): + """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) + 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): ... \ No newline at end of file diff --git a/fastapi/_compat/lazy_import.py b/fastapi/_compat/lazy_import.py new file mode 100644 index 000000000..cde1a1d9a --- /dev/null +++ b/fastapi/_compat/lazy_import.py @@ -0,0 +1,74 @@ +# path: fastapi/_compat/import.py +""" +Centralized lazy import helpers for v1 compatibility. +""" + +from __future__ import annotations +import sys +from typing import Any, Optional, TypeVar + +T = TypeVar('T') + +def get_v1_if_loaded() -> Optional[Any]: + """ + Get pydantic.v1 module only if it's already loaded. + Returns None if not loaded, avoiding any warnings. + """ + 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. + """ + v1 = get_v1_if_loaded() + if v1 is not None: + cls = getattr(v1, v1_class, None) + if cls is not None: + 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. + """ + v1 = get_v1_if_loaded() + if v1 is not None: + 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. + """ + v1 = get_v1_if_loaded() + if v1 is not None: + method = getattr(v1, method_name, None) + if method is not None: + return method(*args, **kwargs) + return None + +def v1_get_attr(attr_name: str) -> Any: + """ + Get v1 attribute only if v1 is loaded. + """ + v1 = get_v1_if_loaded() + if v1 is not None: + return getattr(v1, attr_name, None) + return None diff --git a/fastapi/_compat/main.py b/fastapi/_compat/main.py index 3f758f072..04a96da95 100644 --- a/fastapi/_compat/main.py +++ b/fastapi/_compat/main.py @@ -8,8 +8,8 @@ from typing import ( Type, ) -from fastapi._compat import v1 from fastapi._compat.shared import PYDANTIC_V2, lenient_issubclass +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 @@ -25,204 +25,211 @@ if PYDANTIC_V2: from .v2 import UndefinedType as UndefinedType from .v2 import Url as Url from .v2 import Validator as Validator + from .v2 import _get_model_config as _get_model_config + from .v2 import _model_dump as _model_dump + from .v2 import _model_rebuild as _model_rebuild + from .v2 import copy_field_info as copy_field_info + from .v2 import create_body_model as create_body_model from .v2 import evaluate_forwardref as evaluate_forwardref + from .v2 import get_annotation_from_field_info as get_annotation_from_field_info + from .v2 import get_definitions as get_definitions from .v2 import get_missing_field_error as get_missing_field_error + from .v2 import get_schema_from_model_field as get_schema_from_model_field + 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 from .v2 import ( with_info_plain_validator_function as with_info_plain_validator_function, ) else: - from .v1 import BaseConfig as BaseConfig # type: ignore[assignment] + from .v1 import BaseConfig as BaseConfig from .v1 import FieldInfo as FieldInfo - from .v1 import ( # type: ignore[assignment] - PydanticSchemaGenerationError as PydanticSchemaGenerationError, - ) + from .v1 import PydanticSchemaGenerationError as PydanticSchemaGenerationError from .v1 import RequiredParam as RequiredParam from .v1 import Undefined as Undefined from .v1 import UndefinedType as UndefinedType - from .v1 import Url as Url # type: ignore[assignment] + from .v1 import Url as Url from .v1 import Validator as Validator + from .v1 import _get_model_config as _get_model_config + from .v1 import _model_dump as _model_dump + from .v1 import _model_rebuild as _model_rebuild + from .v1 import copy_field_info as copy_field_info + from .v1 import create_body_model as create_body_model from .v1 import evaluate_forwardref as evaluate_forwardref + from .v1 import get_annotation_from_field_info as get_annotation_from_field_info + from .v1 import get_definitions as get_definitions from .v1 import get_missing_field_error as get_missing_field_error - from .v1 import ( # type: ignore[assignment] + from .v1 import get_schema_from_model_field as get_schema_from_model_field + from .v1 import is_bytes_field as is_bytes_field + from .v1 import is_bytes_sequence_field as is_bytes_sequence_field + from .v1 import is_scalar_field as is_scalar_field + from .v1 import is_scalar_sequence_field as is_scalar_sequence_field + from .v1 import is_sequence_field as is_sequence_field + from .v1 import serialize_sequence_value as serialize_sequence_value + from .v1 import ( with_info_plain_validator_function as with_info_plain_validator_function, ) @lru_cache def get_cached_model_fields(model: Type[BaseModel]) -> List[ModelField]: - if lenient_issubclass(model, v1.BaseModel): + if v1_lenient_issubclass(model, "BaseModel"): + v1 = get_v1_if_loaded() return v1.get_model_fields(model) else: from . import v2 - return v2.get_model_fields(model) # type: ignore[return-value] def _is_undefined(value: object) -> bool: - if isinstance(value, v1.UndefinedType): + if v1_isinstance(value, "UndefinedType"): return True elif PYDANTIC_V2: - from . import v2 - - return isinstance(value, v2.UndefinedType) - return False + from pydantic_core import PydanticUndefined + return value is PydanticUndefined + else: + return False def _get_model_config(model: BaseModel) -> Any: - if isinstance(model, v1.BaseModel): + if v1_isinstance(model, "BaseModel"): + v1 = get_v1_if_loaded() return v1._get_model_config(model) elif PYDANTIC_V2: from . import v2 - return v2._get_model_config(model) + else: + return getattr(model, "__config__", None) def _model_dump( model: BaseModel, mode: Literal["json", "python"] = "json", **kwargs: Any ) -> Any: - if isinstance(model, v1.BaseModel): + if v1_isinstance(model, "BaseModel"): + v1 = get_v1_if_loaded() return v1._model_dump(model, mode=mode, **kwargs) - elif PYDANTIC_V2: + if PYDANTIC_V2: from . import v2 - return v2._model_dump(model, mode=mode, **kwargs) + else: + return model.dict(**kwargs) def _is_error_wrapper(exc: Exception) -> bool: - if isinstance(exc, v1.ErrorWrapper): + if v1_isinstance(exc, "ErrorWrapper"): return True elif PYDANTIC_V2: from . import v2 - - return isinstance(exc, v2.ErrorWrapper) - return False + return v2._is_error_wrapper(exc) + else: + return False def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo: - if isinstance(field_info, v1.FieldInfo): + if v1_isinstance(field_info, "FieldInfo"): + v1 = get_v1_if_loaded() return v1.copy_field_info(field_info=field_info, annotation=annotation) else: - assert PYDANTIC_V2 from . import v2 - return v2.copy_field_info(field_info=field_info, annotation=annotation) def create_body_model( - *, fields: Sequence[ModelField], model_name: str + *, fields: List[ModelField], model_name: str ) -> Type[BaseModel]: - if fields and isinstance(fields[0], v1.ModelField): + 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: - assert PYDANTIC_V2 from . import v2 - - return v2.create_body_model(fields=fields, model_name=model_name) # type: ignore[arg-type] + return v2.create_body_model(fields=fields, model_name=model_name) def get_annotation_from_field_info( annotation: Any, field_info: FieldInfo, field_name: str ) -> Any: - if isinstance(field_info, v1.FieldInfo): + if v1_isinstance(field_info, "FieldInfo"): + v1 = get_v1_if_loaded() return v1.get_annotation_from_field_info( annotation=annotation, field_info=field_info, field_name=field_name ) else: - assert PYDANTIC_V2 from . import v2 - return v2.get_annotation_from_field_info( annotation=annotation, field_info=field_info, field_name=field_name ) def is_bytes_field(field: ModelField) -> bool: - if isinstance(field, v1.ModelField): + if v1_isinstance(field, "ModelField"): + v1 = get_v1_if_loaded() return v1.is_bytes_field(field) else: - assert PYDANTIC_V2 from . import v2 - - return v2.is_bytes_field(field) # type: ignore[arg-type] + return v2.is_bytes_field(field) def is_bytes_sequence_field(field: ModelField) -> bool: - if isinstance(field, v1.ModelField): + if v1_isinstance(field, "ModelField"): + v1 = get_v1_if_loaded() return v1.is_bytes_sequence_field(field) else: - assert PYDANTIC_V2 from . import v2 - - return v2.is_bytes_sequence_field(field) # type: ignore[arg-type] + return v2.is_bytes_sequence_field(field) def is_scalar_field(field: ModelField) -> bool: - if isinstance(field, v1.ModelField): + if v1_isinstance(field, "ModelField"): + v1 = get_v1_if_loaded() return v1.is_scalar_field(field) else: - assert PYDANTIC_V2 from . import v2 - - return v2.is_scalar_field(field) # type: ignore[arg-type] + return v2.is_scalar_field(field) def is_scalar_sequence_field(field: ModelField) -> bool: - if isinstance(field, v1.ModelField): + if v1_isinstance(field, "ModelField"): + v1 = get_v1_if_loaded() return v1.is_scalar_sequence_field(field) else: - assert PYDANTIC_V2 from . import v2 - - return v2.is_scalar_sequence_field(field) # type: ignore[arg-type] + return v2.is_scalar_sequence_field(field) def is_sequence_field(field: ModelField) -> bool: - if isinstance(field, v1.ModelField): + if v1_isinstance(field, "ModelField"): + v1 = get_v1_if_loaded() return v1.is_sequence_field(field) else: - assert PYDANTIC_V2 from . import v2 - - return v2.is_sequence_field(field) # type: ignore[arg-type] + return v2.is_sequence_field(field) def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]: - if isinstance(field, v1.ModelField): + if v1_isinstance(field, "ModelField"): + v1 = get_v1_if_loaded() return v1.serialize_sequence_value(field=field, value=value) else: - assert PYDANTIC_V2 - from . import v2 - - return v2.serialize_sequence_value(field=field, value=value) # type: ignore[arg-type] - - -def _model_rebuild(model: Type[BaseModel]) -> None: - if lenient_issubclass(model, v1.BaseModel): - v1._model_rebuild(model) - elif PYDANTIC_V2: from . import v2 - - v2._model_rebuild(model) + return v2.serialize_sequence_value(field=field, value=value) def get_compat_model_name_map(fields: List[ModelField]) -> ModelNameMap: - v1_model_fields = [field for field in fields if isinstance(field, v1.ModelField)] - v1_flat_models = v1.get_flat_models_from_fields(v1_model_fields, known_models=set()) # type: ignore[attr-defined] + 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() all_flat_models = v1_flat_models if PYDANTIC_V2: from . import v2 - - v2_model_fields = [ - field for field in fields if isinstance(field, v2.ModelField) - ] - v2_flat_models = v2.get_flat_models_from_fields( - v2_model_fields, known_models=set() - ) - all_flat_models = all_flat_models.union(v2_flat_models) - + 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 - model_name_map = v1.get_model_name_map(all_flat_models) + model_name_map = v1.get_model_name_map(all_flat_models) if v1 else {} return model_name_map @@ -232,29 +239,32 @@ def get_definitions( model_name_map: ModelNameMap, separate_input_output_schemas: bool = True, ) -> Tuple[ - Dict[Tuple[ModelField, Literal["validation", "serialization"]], v1.JsonSchemaValue], + Dict[Tuple[ModelField, Literal["validation", "serialization"]], Dict[str, Any]], Dict[str, Dict[str, Any]], ]: - v1_fields = [field for field in fields if isinstance(field, v1.ModelField)] - v1_field_maps, v1_definitions = v1.get_definitions( - fields=v1_fields, - model_name_map=model_name_map, - separate_input_output_schemas=separate_input_output_schemas, - ) - if not PYDANTIC_V2: - return v1_field_maps, v1_definitions + v1 = get_v1_if_loaded() + 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, + model_name_map=model_name_map, + separate_input_output_schemas=separate_input_output_schemas, + ) else: + 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 isinstance(field, v2.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, separate_input_output_schemas=separate_input_output_schemas, ) - all_definitions = {**v1_definitions, **v2_definitions} all_field_maps = {**v1_field_maps, **v2_field_maps} + all_definitions = {**v1_definitions, **v2_definitions} return all_field_maps, all_definitions + return v1_field_maps, v1_definitions def get_schema_from_model_field( @@ -262,11 +272,12 @@ def get_schema_from_model_field( field: ModelField, model_name_map: ModelNameMap, field_mapping: Dict[ - Tuple[ModelField, Literal["validation", "serialization"]], v1.JsonSchemaValue + Tuple[ModelField, Literal["validation", "serialization"]], Dict[str, Any] ], separate_input_output_schemas: bool = True, ) -> Dict[str, Any]: - if isinstance(field, v1.ModelField): + if v1_isinstance(field, "ModelField"): + v1 = get_v1_if_loaded() return v1.get_schema_from_model_field( field=field, model_name_map=model_name_map, @@ -274,32 +285,78 @@ def get_schema_from_model_field( separate_input_output_schemas=separate_input_output_schemas, ) else: - assert PYDANTIC_V2 from . import v2 - return v2.get_schema_from_model_field( - field=field, # type: ignore[arg-type] + field=field, model_name_map=model_name_map, - field_mapping=field_mapping, # type: ignore[arg-type] + field_mapping=field_mapping, separate_input_output_schemas=separate_input_output_schemas, ) def _is_model_field(value: Any) -> bool: - if isinstance(value, v1.ModelField): + if v1_isinstance(value, "ModelField"): return True elif PYDANTIC_V2: from . import v2 - - return isinstance(value, v2.ModelField) - return False + return v2._is_model_field(value) + else: + return False def _is_model_class(value: Any) -> bool: - if lenient_issubclass(value, v1.BaseModel): + if v1_lenient_issubclass(value, "BaseModel"): return True elif PYDANTIC_V2: from . import v2 + return v2._is_model_class(value) + else: + return False + + +def get_missing_field_error(loc: Tuple[str, ...], field: ModelField) -> Dict[str, Any]: + if v1_isinstance(field, "ModelField"): + v1 = get_v1_if_loaded() + return v1.get_missing_field_error(loc=loc, field=field) + else: + from . import v2 + return v2.get_missing_field_error(loc=loc, field=field) - return lenient_issubclass(value, v2.BaseModel) # type: ignore[attr-defined] - return False + +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_=type_, globalns=globalns, localns=localns) + else: + v1 = get_v1_if_loaded() + return v1.evaluate_forwardref(type_=type_, globalns=globalns, localns=localns) + + +def with_info_plain_validator_function( + func: Any, + info_argname: str = "info", +) -> Any: + if PYDANTIC_V2: + try: + from pydantic_core.core_schema import ( + with_info_plain_validator_function as pydantic_core_with_info, + ) + except ImportError: # pragma: no cover + from pydantic_core.core_schema import ( + general_plain_validator_function as pydantic_core_with_info, + ) + 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) + + +def _model_rebuild(model) -> None: + if v1_lenient_issubclass(model, "BaseModel"): + v1 = get_v1_if_loaded() + v1._model_rebuild(model) + elif PYDANTIC_V2: + from . import v2 + v2._model_rebuild(model) + else: + model.update_forward_refs() \ No newline at end of file diff --git a/fastapi/_compat/shared.py b/fastapi/_compat/shared.py index 495d5c5f7..d1d47cf99 100644 --- a/fastapi/_compat/shared.py +++ b/fastapi/_compat/shared.py @@ -16,7 +16,6 @@ from typing import ( Union, ) -from fastapi._compat import v1 from fastapi.types import UnionType from pydantic import BaseModel from pydantic.version import VERSION as PYDANTIC_VERSION @@ -97,8 +96,13 @@ def value_is_sequence(value: Any) -> bool: def _annotation_is_complex(annotation: Union[Type[Any], None]) -> bool: + types_tuple = (BaseModel, Mapping, UploadFile) + 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, (BaseModel, v1.BaseModel, Mapping, UploadFile)) + lenient_issubclass(annotation, types_tuple) or _annotation_is_sequence(annotation) or is_dataclass(annotation) ) @@ -195,15 +199,14 @@ def is_uploadfile_sequence_annotation(annotation: Any) -> bool: def annotation_is_pydantic_v1(annotation: Any) -> bool: - if lenient_issubclass(annotation, v1.BaseModel): + 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 is Union or origin is UnionType: - for arg in get_args(annotation): - if lenient_issubclass(arg, v1.BaseModel): - return True + if origin in (Union, UnionType): + return any(lenient_issubclass(arg, _v1.BaseModel) for arg in get_args(annotation)) if field_annotation_is_sequence(annotation): - for sub_annotation in get_args(annotation): - if annotation_is_pydantic_v1(sub_annotation): - return True + return any(annotation_is_pydantic_v1(sa) for sa in get_args(annotation)) return False diff --git a/fastapi/_compat/v1.py b/fastapi/_compat/v1.py index f0ac51634..ce6a35310 100644 --- a/fastapi/_compat/v1.py +++ b/fastapi/_compat/v1.py @@ -1,334 +1,136 @@ -from copy import copy -from dataclasses import dataclass, is_dataclass -from enum import Enum -from typing import ( - Any, - Callable, - Dict, - List, - Sequence, - Set, - Tuple, - Type, - Union, -) +# path: fastapi/_compat/v1.py -from fastapi._compat import shared -from fastapi.openapi.constants import REF_PREFIX as REF_PREFIX -from fastapi.types import ModelNameMap -from pydantic.version import VERSION as PYDANTIC_VERSION -from typing_extensions import Literal - -PYDANTIC_VERSION_MINOR_TUPLE = tuple(int(x) for x in PYDANTIC_VERSION.split(".")[:2]) -PYDANTIC_V2 = PYDANTIC_VERSION_MINOR_TUPLE[0] == 2 -# Keeping old "Required" functionality from Pydantic V1, without -# shadowing typing.Required. -RequiredParam: Any = Ellipsis - -if not PYDANTIC_V2: - from pydantic import BaseConfig as BaseConfig - from pydantic import BaseModel as BaseModel - from pydantic import ValidationError as ValidationError - from pydantic import create_model as create_model - from pydantic.class_validators import Validator as Validator - from pydantic.color import Color as Color - from pydantic.error_wrappers import ErrorWrapper as ErrorWrapper - from pydantic.errors import MissingError - from pydantic.fields import ( # type: ignore[attr-defined] - SHAPE_FROZENSET, - SHAPE_LIST, - SHAPE_SEQUENCE, - SHAPE_SET, - SHAPE_SINGLETON, - SHAPE_TUPLE, - SHAPE_TUPLE_ELLIPSIS, - ) - from pydantic.fields import FieldInfo as FieldInfo - from pydantic.fields import ModelField as ModelField # type: ignore[attr-defined] - from pydantic.fields import Undefined as Undefined # type: ignore[attr-defined] - from pydantic.fields import ( # type: ignore[attr-defined] - UndefinedType as UndefinedType, - ) - from pydantic.networks import AnyUrl as AnyUrl - from pydantic.networks import NameEmail as NameEmail - from pydantic.schema import TypeModelSet as TypeModelSet - from pydantic.schema import ( - field_schema, - get_flat_models_from_fields, - model_process_schema, - ) - from pydantic.schema import ( - get_annotation_from_field_info as get_annotation_from_field_info, - ) - from pydantic.schema import get_flat_models_from_field as get_flat_models_from_field - from pydantic.schema import get_model_name_map as get_model_name_map - from pydantic.types import SecretBytes as SecretBytes - from pydantic.types import SecretStr as SecretStr - from pydantic.typing import evaluate_forwardref as evaluate_forwardref - from pydantic.utils import lenient_issubclass as lenient_issubclass - - -else: - from pydantic.v1 import BaseConfig as BaseConfig # type: ignore[assignment] - from pydantic.v1 import BaseModel as BaseModel # type: ignore[assignment] - from pydantic.v1 import ( # type: ignore[assignment] - ValidationError as ValidationError, - ) - from pydantic.v1 import create_model as create_model # type: ignore[no-redef] - from pydantic.v1.class_validators import Validator as Validator - from pydantic.v1.color import Color as Color # type: ignore[assignment] - from pydantic.v1.error_wrappers import ErrorWrapper as ErrorWrapper - from pydantic.v1.errors import MissingError - from pydantic.v1.fields import ( - SHAPE_FROZENSET, - SHAPE_LIST, - SHAPE_SEQUENCE, - SHAPE_SET, - SHAPE_SINGLETON, - SHAPE_TUPLE, - SHAPE_TUPLE_ELLIPSIS, - ) - from pydantic.v1.fields import FieldInfo as FieldInfo # type: ignore[assignment] - from pydantic.v1.fields import ModelField as ModelField - from pydantic.v1.fields import Undefined as Undefined - from pydantic.v1.fields import UndefinedType as UndefinedType - from pydantic.v1.networks import AnyUrl as AnyUrl - from pydantic.v1.networks import ( # type: ignore[assignment] - NameEmail as NameEmail, - ) - from pydantic.v1.schema import TypeModelSet as TypeModelSet - from pydantic.v1.schema import ( - field_schema, - get_flat_models_from_fields, - model_process_schema, - ) - from pydantic.v1.schema import ( - get_annotation_from_field_info as get_annotation_from_field_info, - ) - from pydantic.v1.schema import ( - get_flat_models_from_field as get_flat_models_from_field, - ) - from pydantic.v1.schema import get_model_name_map as get_model_name_map - from pydantic.v1.types import ( # type: ignore[assignment] - SecretBytes as SecretBytes, - ) - from pydantic.v1.types import ( # type: ignore[assignment] - SecretStr as SecretStr, - ) - from pydantic.v1.typing import evaluate_forwardref as evaluate_forwardref - from pydantic.v1.utils import lenient_issubclass as lenient_issubclass - - -GetJsonSchemaHandler = Any -JsonSchemaValue = Dict[str, Any] -CoreSchema = Any -Url = AnyUrl - -sequence_shapes = { - SHAPE_LIST, - SHAPE_SET, - SHAPE_FROZENSET, - SHAPE_TUPLE, - SHAPE_SEQUENCE, - SHAPE_TUPLE_ELLIPSIS, -} -sequence_shape_to_type = { - SHAPE_LIST: list, - SHAPE_SET: set, - SHAPE_TUPLE: tuple, - SHAPE_SEQUENCE: list, - SHAPE_TUPLE_ELLIPSIS: list, -} - - -@dataclass -class GenerateJsonSchema: - ref_template: str - - -class PydanticSchemaGenerationError(Exception): - pass +""" +Pydantic V1 compatibility module with lazy loading and controlled warnings. +This module acts as a transparent proxy to pydantic.v1, loading it only when needed +and providing controlled warnings for Python 3.14+ usage. +""" -RequestErrorModel: Type[BaseModel] = create_model("Request") +from __future__ import annotations +import importlib +import os +import sys +import warnings +from typing import Any, Dict, List, Sequence, Tuple +from typing_extensions import Literal -def with_info_plain_validator_function( - function: Callable[..., Any], - *, - ref: Union[str, None] = None, - metadata: Any = None, - serialization: Any = None, -) -> Any: - return {} - - -def get_model_definitions( - *, - flat_models: Set[Union[Type[BaseModel], Type[Enum]]], - model_name_map: Dict[Union[Type[BaseModel], Type[Enum]], str], -) -> Dict[str, Any]: - definitions: Dict[str, Dict[str, Any]] = {} - for model in flat_models: - m_schema, m_definitions, m_nested_models = model_process_schema( - model, model_name_map=model_name_map, ref_prefix=REF_PREFIX - ) - definitions.update(m_definitions) - model_name = model_name_map[model] - definitions[model_name] = m_schema - for m_schema in definitions.values(): - if "description" in m_schema: - m_schema["description"] = m_schema["description"].split("\f")[0] - return definitions - - -def is_pv1_scalar_field(field: ModelField) -> bool: - from fastapi import params - - field_info = field.field_info - if not ( - field.shape == SHAPE_SINGLETON - and not lenient_issubclass(field.type_, BaseModel) - and not lenient_issubclass(field.type_, dict) - and not shared.field_annotation_is_sequence(field.type_) - and not is_dataclass(field.type_) - and not isinstance(field_info, params.Body) - ): - return False - if field.sub_fields: - if not all(is_pv1_scalar_field(f) for f in field.sub_fields): - return False - return True - - -def is_pv1_scalar_sequence_field(field: ModelField) -> bool: - if (field.shape in sequence_shapes) and not lenient_issubclass( - field.type_, BaseModel - ): - if field.sub_fields is not None: - for sub_field in field.sub_fields: - if not is_pv1_scalar_field(sub_field): - return False - return True - if shared._annotation_is_sequence(field.type_): - return True - return False - +# Never import pydantic.v1 at import-time of this file. +# Load on demand in __getattr__ (PEP 562). + +_pv1 = None +_warned = False + +def _load(): + global _pv1, _warned + if _pv1 is not None: + return _pv1 + if sys.version_info >= (3, 14): + msg = "Pydantic v1 em Python 3.14+ é desaconselhado/deprecado. Migre para v2." + if os.getenv("FASTAPI_PYDANTIC_V1_STRICT") == "1": + raise RuntimeError(msg) + if not _warned: + warnings.warn(msg, DeprecationWarning, stacklevel=3) + _warned = True + _pv1 = importlib.import_module("pydantic.v1") + return _pv1 + +def __getattr__(name: str) -> Any: + mod = _load() + # tenta direto no módulo principal + 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"): + 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]]: - use_errors: List[Any] = [] - for error in errors: - if isinstance(error, ErrorWrapper): - new_errors = ValidationError( # type: ignore[call-arg] - errors=[error], model=RequestErrorModel - ).errors() - use_errors.extend(new_errors) - elif isinstance(error, list): - use_errors.extend(_normalize_errors(error)) + 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()) + elif isinstance(err, list): + out.extend(_normalize_errors(err)) else: - use_errors.append(error) - return use_errors - - -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 _normalize_errors(errors) - ] - - return updated_loc_errors + 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 _model_rebuild(model: Type[BaseModel]) -> None: +def _model_rebuild(model) -> None: model.update_forward_refs() - -def _model_dump( - model: BaseModel, 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_model_config(model: BaseModel) -> Any: - return model.__config__ # type: ignore[attr-defined] - - -def get_schema_from_model_field( - *, - field: ModelField, - model_name_map: ModelNameMap, - field_mapping: Dict[ - Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue - ], - separate_input_output_schemas: bool = True, -) -> Dict[str, Any]: - return field_schema( # type: ignore[no-any-return] - field, model_name_map=model_name_map, ref_prefix=REF_PREFIX - )[0] - - -# def get_compat_model_name_map(fields: List[ModelField]) -> ModelNameMap: -# models = get_flat_models_from_fields(fields, known_models=set()) -# return get_model_name_map(models) # type: ignore[no-any-return] - - -def get_definitions( - *, - fields: List[ModelField], - model_name_map: ModelNameMap, - separate_input_output_schemas: bool = True, -) -> Tuple[ - Dict[Tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue], - Dict[str, Dict[str, Any]], -]: - models = get_flat_models_from_fields(fields, known_models=set()) - return {}, get_model_definitions(flat_models=models, model_name_map=model_name_map) - - -def is_scalar_field(field: ModelField) -> bool: - return is_pv1_scalar_field(field) - - -def is_sequence_field(field: ModelField) -> bool: - return field.shape in sequence_shapes or shared._annotation_is_sequence(field.type_) - - -def is_scalar_sequence_field(field: ModelField) -> bool: - return is_pv1_scalar_sequence_field(field) - - -def is_bytes_field(field: ModelField) -> bool: - return lenient_issubclass(field.type_, bytes) # type: ignore[no-any-return] - - -def is_bytes_sequence_field(field: ModelField) -> bool: - return field.shape in sequence_shapes and lenient_issubclass(field.type_, bytes) - - -def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo: - return copy(field_info) - - -def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]: - return sequence_shape_to_type[field.shape](value) # type: ignore[no-any-return] - - -def get_missing_field_error(loc: Tuple[str, ...]) -> Dict[str, Any]: - missing_field_error = ErrorWrapper(MissingError(), loc=loc) - new_error = ValidationError([missing_field_error], RequestErrorModel) - return new_error.errors()[0] # type: ignore[return-value] +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): + 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") + 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) + +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)) + +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} + +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 [])) + return False -def create_body_model( - *, fields: Sequence[ModelField], model_name: str -) -> Type[BaseModel]: - BodyModel = create_model(model_name) - for f in fields: - BodyModel.__fields__[f.name] = f # type: ignore[index] - return BodyModel +from copy import copy as _copy +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 } + return mapping[field.shape](value) -def get_model_fields(model: Type[BaseModel]) -> List[ModelField]: - return list(model.__fields__.values()) # type: ignore[attr-defined] +# Type aliases for backward compatibility +GetJsonSchemaHandler = Any +JsonSchemaValue = dict[str, Any] +CoreSchema = Any +Url = Any \ No newline at end of file diff --git a/fastapi/_compat/v2.py b/fastapi/_compat/v2.py index 29606b9f3..dd2893370 100644 --- a/fastapi/_compat/v2.py +++ b/fastapi/_compat/v2.py @@ -15,7 +15,13 @@ from typing import ( cast, ) -from fastapi._compat import shared, v1 +from fastapi._compat import shared + +# 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 from fastapi.openapi.constants import REF_TEMPLATE from fastapi.types import IncEx, ModelNameMap from pydantic import BaseModel, TypeAdapter, create_model @@ -116,7 +122,7 @@ class ModelField: None, ) except ValidationError as exc: - return None, v1._regenerate_error_with_loc( + return None, _get_v1()._regenerate_error_with_loc( errors=exc.errors(include_url=False), loc_prefix=loc ) diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py index 675ad6faf..a82c0ca06 100644 --- a/fastapi/dependencies/utils.py +++ b/fastapi/dependencies/utils.py @@ -45,10 +45,16 @@ from fastapi._compat import ( lenient_issubclass, sequence_types, serialize_sequence_value, - v1, + # v1, # Lazy import to avoid warnings value_is_sequence, ) from fastapi._compat.shared import annotation_is_pydantic_v1 + +# 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 from fastapi.background import BackgroundTasks from fastapi.concurrency import ( asynccontextmanager, @@ -76,7 +82,7 @@ from starlette.responses import Response from starlette.websockets import WebSocket from typing_extensions import Annotated, get_args, get_origin -from .. import temp_pydantic_v1_params +from .._compat import _v1_params as temp_pydantic_v1_params if sys.version_info >= (3, 13): # pragma: no cover from inspect import iscoroutinefunction @@ -380,7 +386,7 @@ def analyze_param( fastapi_annotations = [ arg for arg in annotated_args[1:] - if isinstance(arg, (FieldInfo, v1.FieldInfo, params.Depends)) + if isinstance(arg, (FieldInfo, _get_v1().FieldInfo, params.Depends)) ] fastapi_specific_annotations = [ arg @@ -397,21 +403,21 @@ def analyze_param( ) ] if fastapi_specific_annotations: - fastapi_annotation: Union[FieldInfo, v1.FieldInfo, params.Depends, None] = ( + fastapi_annotation: Union[FieldInfo, _get_v1().FieldInfo, params.Depends, None] = ( fastapi_specific_annotations[-1] ) else: fastapi_annotation = None # Set default for Annotated FieldInfo - if isinstance(fastapi_annotation, (FieldInfo, v1.FieldInfo)): + if isinstance(fastapi_annotation, (FieldInfo, _get_v1().FieldInfo)): # Copy `field_info` because we mutate `field_info.default` below. field_info = copy_field_info( field_info=fastapi_annotation, annotation=use_annotation ) assert field_info.default in { Undefined, - v1.Undefined, - } or field_info.default in {RequiredParam, v1.RequiredParam}, ( + _get_v1().Undefined, + } or field_info.default in {RequiredParam, _get_v1().RequiredParam}, ( f"`{field_info.__class__.__name__}` default value cannot be set in" f" `Annotated` for {param_name!r}. Set the default value with `=` instead." ) @@ -435,7 +441,7 @@ def analyze_param( ) depends = value # Get FieldInfo from default value - elif isinstance(value, (FieldInfo, v1.FieldInfo)): + elif isinstance(value, (FieldInfo, _get_v1().FieldInfo)): assert field_info is None, ( "Cannot specify FastAPI annotations in `Annotated` and default value" f" together for {param_name!r}" @@ -524,7 +530,7 @@ def analyze_param( type_=use_annotation_from_field_info, default=field_info.default, alias=alias, - required=field_info.default in (RequiredParam, v1.RequiredParam, Undefined), + required=field_info.default in (RequiredParam, _get_v1().RequiredParam, Undefined), field_info=field_info, ) if is_path_param: @@ -741,7 +747,7 @@ def _validate_value_with_model_field( if _is_error_wrapper(errors_): # type: ignore[arg-type] return None, [errors_] elif isinstance(errors_, list): - new_errors = v1._regenerate_error_with_loc(errors=errors_, loc_prefix=()) + new_errors = _get_v1()._regenerate_error_with_loc(errors=errors_, loc_prefix=()) return None, new_errors else: return v_, [] diff --git a/fastapi/encoders.py b/fastapi/encoders.py index 8ff7d58dd..0177cceaa 100644 --- a/fastapi/encoders.py +++ b/fastapi/encoders.py @@ -1,5 +1,6 @@ import dataclasses import datetime +import sys from collections import defaultdict, deque from decimal import Decimal from enum import Enum @@ -17,7 +18,6 @@ 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 @@ -59,7 +59,6 @@ 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, @@ -76,21 +75,29 @@ 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, } +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) + ENCODERS_BY_TYPE.setdefault(v1.SecretStr, str) + ENCODERS_BY_TYPE.setdefault(getattr(v1.networks, "Url", v1.AnyUrl), str) + ENCODERS_BY_TYPE.setdefault(v1.AnyUrl, str) + def generate_encoders_by_class_tuples( type_encoder_map: Dict[Any, Callable[[Any], Any]], @@ -208,6 +215,9 @@ def jsonable_encoder( Read more about it in the [FastAPI docs for JSON Compatible Encoder](https://fastapi.tiangolo.com/tutorial/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: @@ -220,13 +230,16 @@ def jsonable_encoder( include = set(include) if exclude is not None and not isinstance(exclude, (set, dict)): exclude = set(exclude) - if isinstance(obj, (BaseModel, v1.BaseModel)): + if isinstance(obj, BaseModel): # TODO: remove when deprecating Pydantic v1 encoders: Dict[Any, Any] = {} - if isinstance(obj, v1.BaseModel): - encoders = getattr(obj.__config__, "json_encoders", {}) # type: ignore[attr-defined] - if custom_encoder: - encoders = {**encoders, **custom_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: + encoders = {**encoders, **custom_encoder} obj_dict = _model_dump( obj, mode="json", diff --git a/fastapi/routing.py b/fastapi/routing.py index fe25d7dec..8b8952931 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -24,7 +24,8 @@ from typing import ( Union, ) -from fastapi import params, temp_pydantic_v1_params +from fastapi import params +from fastapi._compat import _v1_params as temp_pydantic_v1_params from fastapi._compat import ( ModelField, Undefined, diff --git a/fastapi/temp_pydantic_v1_params.py b/fastapi/temp_pydantic_v1_params.py index 0535ee727..80399d1f7 100644 --- a/fastapi/temp_pydantic_v1_params.py +++ b/fastapi/temp_pydantic_v1_params.py @@ -1,724 +1,35 @@ -import warnings -from typing import Any, Callable, Dict, List, Optional, Union - -from fastapi.openapi.models import Example -from fastapi.params import ParamTypes -from typing_extensions import Annotated, deprecated - -from ._compat.shared import PYDANTIC_VERSION_MINOR_TUPLE -from ._compat.v1 import FieldInfo, Undefined - -_Unset: Any = Undefined - - -class Param(FieldInfo): # type: ignore[misc] - in_: ParamTypes - - def __init__( - self, - default: Any = Undefined, - *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - # TODO: update when deprecating Pydantic v1, import these types - # validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - regex: Annotated[ - Optional[str], - deprecated( - "Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead." - ), - ] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - example: Annotated[ - Optional[Any], - deprecated( - "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " - "although still supported. Use examples instead." - ), - ] = _Unset, - openapi_examples: Optional[Dict[str, Example]] = None, - deprecated: Union[deprecated, str, bool, None] = None, - include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, - **extra: Any, - ): - if example is not _Unset: - warnings.warn( - "`example` has been deprecated, please use `examples` instead", - category=DeprecationWarning, - stacklevel=4, - ) - self.example = example - self.include_in_schema = include_in_schema - self.openapi_examples = openapi_examples - kwargs = dict( - default=default, - default_factory=default_factory, - alias=alias, - title=title, - description=description, - gt=gt, - ge=ge, - lt=lt, - le=le, - min_length=min_length, - max_length=max_length, - discriminator=discriminator, - multiple_of=multiple_of, - allow_inf_nan=allow_inf_nan, - max_digits=max_digits, - decimal_places=decimal_places, - **extra, - ) - if examples is not None: - kwargs["examples"] = examples - if regex is not None: - warnings.warn( - "`regex` has been deprecated, please use `pattern` instead", - category=DeprecationWarning, - stacklevel=4, - ) - current_json_schema_extra = json_schema_extra or extra - if PYDANTIC_VERSION_MINOR_TUPLE < (2, 7): - self.deprecated = deprecated - else: - kwargs["deprecated"] = deprecated - kwargs["regex"] = pattern or regex - kwargs.update(**current_json_schema_extra) - use_kwargs = {k: v for k, v in kwargs.items() if v is not _Unset} - - super().__init__(**use_kwargs) - - def __repr__(self) -> str: - return f"{self.__class__.__name__}({self.default})" - - -class Path(Param): # type: ignore[misc] - in_ = ParamTypes.path - - def __init__( - self, - default: Any = ..., - *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - # TODO: update when deprecating Pydantic v1, import these types - # validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - regex: Annotated[ - Optional[str], - deprecated( - "Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead." - ), - ] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - example: Annotated[ - Optional[Any], - deprecated( - "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " - "although still supported. Use examples instead." - ), - ] = _Unset, - openapi_examples: Optional[Dict[str, Example]] = None, - deprecated: Union[deprecated, str, bool, None] = None, - include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, - **extra: Any, - ): - assert default is ..., "Path parameters cannot have a default value" - self.in_ = self.in_ - super().__init__( - default=default, - default_factory=default_factory, - annotation=annotation, - alias=alias, - alias_priority=alias_priority, - validation_alias=validation_alias, - serialization_alias=serialization_alias, - title=title, - description=description, - gt=gt, - ge=ge, - lt=lt, - le=le, - min_length=min_length, - max_length=max_length, - pattern=pattern, - regex=regex, - discriminator=discriminator, - strict=strict, - multiple_of=multiple_of, - allow_inf_nan=allow_inf_nan, - max_digits=max_digits, - decimal_places=decimal_places, - deprecated=deprecated, - example=example, - examples=examples, - openapi_examples=openapi_examples, - include_in_schema=include_in_schema, - json_schema_extra=json_schema_extra, - **extra, - ) - - -class Query(Param): # type: ignore[misc] - in_ = ParamTypes.query - - def __init__( - self, - default: Any = Undefined, - *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - # TODO: update when deprecating Pydantic v1, import these types - # validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - regex: Annotated[ - Optional[str], - deprecated( - "Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead." - ), - ] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - example: Annotated[ - Optional[Any], - deprecated( - "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " - "although still supported. Use examples instead." - ), - ] = _Unset, - openapi_examples: Optional[Dict[str, Example]] = None, - deprecated: Union[deprecated, str, bool, None] = None, - include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, - **extra: Any, - ): - super().__init__( - default=default, - default_factory=default_factory, - annotation=annotation, - alias=alias, - alias_priority=alias_priority, - validation_alias=validation_alias, - serialization_alias=serialization_alias, - title=title, - description=description, - gt=gt, - ge=ge, - lt=lt, - le=le, - min_length=min_length, - max_length=max_length, - pattern=pattern, - regex=regex, - discriminator=discriminator, - strict=strict, - multiple_of=multiple_of, - allow_inf_nan=allow_inf_nan, - max_digits=max_digits, - decimal_places=decimal_places, - deprecated=deprecated, - example=example, - examples=examples, - openapi_examples=openapi_examples, - include_in_schema=include_in_schema, - json_schema_extra=json_schema_extra, - **extra, - ) - - -class Header(Param): # type: ignore[misc] - in_ = ParamTypes.header - - def __init__( - self, - default: Any = Undefined, - *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - # TODO: update when deprecating Pydantic v1, import these types - # validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - convert_underscores: bool = True, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - regex: Annotated[ - Optional[str], - deprecated( - "Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead." - ), - ] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - example: Annotated[ - Optional[Any], - deprecated( - "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " - "although still supported. Use examples instead." - ), - ] = _Unset, - openapi_examples: Optional[Dict[str, Example]] = None, - deprecated: Union[deprecated, str, bool, None] = None, - include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, - **extra: Any, - ): - self.convert_underscores = convert_underscores - super().__init__( - default=default, - default_factory=default_factory, - annotation=annotation, - alias=alias, - alias_priority=alias_priority, - validation_alias=validation_alias, - serialization_alias=serialization_alias, - title=title, - description=description, - gt=gt, - ge=ge, - lt=lt, - le=le, - min_length=min_length, - max_length=max_length, - pattern=pattern, - regex=regex, - discriminator=discriminator, - strict=strict, - multiple_of=multiple_of, - allow_inf_nan=allow_inf_nan, - max_digits=max_digits, - decimal_places=decimal_places, - deprecated=deprecated, - example=example, - examples=examples, - openapi_examples=openapi_examples, - include_in_schema=include_in_schema, - json_schema_extra=json_schema_extra, - **extra, - ) - - -class Cookie(Param): # type: ignore[misc] - in_ = ParamTypes.cookie - - def __init__( - self, - default: Any = Undefined, - *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - # TODO: update when deprecating Pydantic v1, import these types - # validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - regex: Annotated[ - Optional[str], - deprecated( - "Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead." - ), - ] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - example: Annotated[ - Optional[Any], - deprecated( - "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " - "although still supported. Use examples instead." - ), - ] = _Unset, - openapi_examples: Optional[Dict[str, Example]] = None, - deprecated: Union[deprecated, str, bool, None] = None, - include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, - **extra: Any, - ): - super().__init__( - default=default, - default_factory=default_factory, - annotation=annotation, - alias=alias, - alias_priority=alias_priority, - validation_alias=validation_alias, - serialization_alias=serialization_alias, - title=title, - description=description, - gt=gt, - ge=ge, - lt=lt, - le=le, - min_length=min_length, - max_length=max_length, - pattern=pattern, - regex=regex, - discriminator=discriminator, - strict=strict, - multiple_of=multiple_of, - allow_inf_nan=allow_inf_nan, - max_digits=max_digits, - decimal_places=decimal_places, - deprecated=deprecated, - example=example, - examples=examples, - openapi_examples=openapi_examples, - include_in_schema=include_in_schema, - json_schema_extra=json_schema_extra, - **extra, - ) - - -class Body(FieldInfo): # type: ignore[misc] - def __init__( - self, - default: Any = Undefined, - *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - embed: Union[bool, None] = None, - media_type: str = "application/json", - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - # TODO: update when deprecating Pydantic v1, import these types - # validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - regex: Annotated[ - Optional[str], - deprecated( - "Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead." - ), - ] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - example: Annotated[ - Optional[Any], - deprecated( - "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " - "although still supported. Use examples instead." - ), - ] = _Unset, - openapi_examples: Optional[Dict[str, Example]] = None, - deprecated: Union[deprecated, str, bool, None] = None, - include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, - **extra: Any, - ): - self.embed = embed - self.media_type = media_type - if example is not _Unset: - warnings.warn( - "`example` has been deprecated, please use `examples` instead", - category=DeprecationWarning, - stacklevel=4, - ) - self.example = example - self.include_in_schema = include_in_schema - self.openapi_examples = openapi_examples - kwargs = dict( - default=default, - default_factory=default_factory, - alias=alias, - title=title, - description=description, - gt=gt, - ge=ge, - lt=lt, - le=le, - min_length=min_length, - max_length=max_length, - discriminator=discriminator, - multiple_of=multiple_of, - allow_inf_nan=allow_inf_nan, - max_digits=max_digits, - decimal_places=decimal_places, - **extra, - ) - if examples is not None: - kwargs["examples"] = examples - if regex is not None: - warnings.warn( - "`regex` has been deprecated, please use `pattern` instead", - category=DeprecationWarning, - stacklevel=4, - ) - current_json_schema_extra = json_schema_extra or extra - if PYDANTIC_VERSION_MINOR_TUPLE < (2, 7): - self.deprecated = deprecated - else: - kwargs["deprecated"] = deprecated - kwargs["regex"] = pattern or regex - kwargs.update(**current_json_schema_extra) - - use_kwargs = {k: v for k, v in kwargs.items() if v is not _Unset} - - super().__init__(**use_kwargs) - - def __repr__(self) -> str: - return f"{self.__class__.__name__}({self.default})" - - -class Form(Body): # type: ignore[misc] - def __init__( - self, - default: Any = Undefined, - *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - media_type: str = "application/x-www-form-urlencoded", - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - # TODO: update when deprecating Pydantic v1, import these types - # validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - regex: Annotated[ - Optional[str], - deprecated( - "Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead." - ), - ] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - example: Annotated[ - Optional[Any], - deprecated( - "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " - "although still supported. Use examples instead." - ), - ] = _Unset, - openapi_examples: Optional[Dict[str, Example]] = None, - deprecated: Union[deprecated, str, bool, None] = None, - include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, - **extra: Any, - ): - super().__init__( - default=default, - default_factory=default_factory, - annotation=annotation, - media_type=media_type, - alias=alias, - alias_priority=alias_priority, - validation_alias=validation_alias, - serialization_alias=serialization_alias, - title=title, - description=description, - gt=gt, - ge=ge, - lt=lt, - le=le, - min_length=min_length, - max_length=max_length, - pattern=pattern, - regex=regex, - discriminator=discriminator, - strict=strict, - multiple_of=multiple_of, - allow_inf_nan=allow_inf_nan, - max_digits=max_digits, - decimal_places=decimal_places, - deprecated=deprecated, - example=example, - examples=examples, - openapi_examples=openapi_examples, - include_in_schema=include_in_schema, - json_schema_extra=json_schema_extra, - **extra, - ) - - -class File(Form): # type: ignore[misc] - def __init__( - self, - default: Any = Undefined, - *, - default_factory: Union[Callable[[], Any], None] = _Unset, - annotation: Optional[Any] = None, - media_type: str = "multipart/form-data", - alias: Optional[str] = None, - alias_priority: Union[int, None] = _Unset, - # TODO: update when deprecating Pydantic v1, import these types - # validation_alias: str | AliasPath | AliasChoices | None - validation_alias: Union[str, None] = None, - serialization_alias: Union[str, None] = None, - title: Optional[str] = None, - description: Optional[str] = None, - gt: Optional[float] = None, - ge: Optional[float] = None, - lt: Optional[float] = None, - le: Optional[float] = None, - min_length: Optional[int] = None, - max_length: Optional[int] = None, - pattern: Optional[str] = None, - regex: Annotated[ - Optional[str], - deprecated( - "Deprecated in FastAPI 0.100.0 and Pydantic v2, use `pattern` instead." - ), - ] = None, - discriminator: Union[str, None] = None, - strict: Union[bool, None] = _Unset, - multiple_of: Union[float, None] = _Unset, - allow_inf_nan: Union[bool, None] = _Unset, - max_digits: Union[int, None] = _Unset, - decimal_places: Union[int, None] = _Unset, - examples: Optional[List[Any]] = None, - example: Annotated[ - Optional[Any], - deprecated( - "Deprecated in OpenAPI 3.1.0 that now uses JSON Schema 2020-12, " - "although still supported. Use examples instead." - ), - ] = _Unset, - openapi_examples: Optional[Dict[str, Example]] = None, - deprecated: Union[deprecated, str, bool, None] = None, - include_in_schema: bool = True, - json_schema_extra: Union[Dict[str, Any], None] = None, - **extra: Any, - ): - super().__init__( - default=default, - default_factory=default_factory, - annotation=annotation, - media_type=media_type, - alias=alias, - alias_priority=alias_priority, - validation_alias=validation_alias, - serialization_alias=serialization_alias, - title=title, - description=description, - gt=gt, - ge=ge, - lt=lt, - le=le, - min_length=min_length, - max_length=max_length, - pattern=pattern, - regex=regex, - discriminator=discriminator, - strict=strict, - multiple_of=multiple_of, - allow_inf_nan=allow_inf_nan, - max_digits=max_digits, - decimal_places=decimal_places, - deprecated=deprecated, - example=example, - examples=examples, - openapi_examples=openapi_examples, - include_in_schema=include_in_schema, - json_schema_extra=json_schema_extra, - **extra, - ) +# path: fastapi/temp_pydantic_v1_params.py +""" +Compatibility proxy for Pydantic v1 params. + +NOTE: This module re-exports the private shim from fastapi._compat._v1_params. +It exists to keep backward compatibility for imports in tests/docs: + from fastapi.temp_pydantic_v1_params import Body, Query, Form, ... + +Internally, _v1_params is lazy and will only touch pydantic.v1 if v1 is actually used. +""" + +from __future__ import annotations + +# Re-export shim classes so isinstance(...) keeps working +from ._compat._v1_params import ( # noqa: F401 + Body, + Cookie, + File, + Form, + Header, + Param, + Path, + Query, +) + +__all__ = [ + "Param", + "Body", + "Form", + "File", + "Query", + "Header", + "Cookie", + "Path", +] diff --git a/fastapi/utils.py b/fastapi/utils.py index 3ea9271b1..653733f05 100644 --- a/fastapi/utils.py +++ b/fastapi/utils.py @@ -25,8 +25,13 @@ from fastapi._compat import ( Validator, annotation_is_pydantic_v1, lenient_issubclass, - v1, ) + +# 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 from fastapi.datastructures import DefaultPlaceholder, DefaultType from pydantic import BaseModel from pydantic.fields import FieldInfo @@ -87,8 +92,8 @@ def create_model_field( ) -> ModelField: class_validators = class_validators or {} - v1_model_config = v1.BaseConfig - v1_field_info = field_info or v1.FieldInfo() + v1_model_config = _get_v1().BaseConfig + v1_field_info = field_info or _get_v1().FieldInfo() v1_kwargs = { "name": name, "field_info": v1_field_info, @@ -102,11 +107,11 @@ def create_model_field( if ( annotation_is_pydantic_v1(type_) - or isinstance(field_info, v1.FieldInfo) + or isinstance(field_info, _get_v1().FieldInfo) or version == "1" ): try: - return v1.ModelField(**v1_kwargs) # type: ignore[no-any-return] + return _get_v1().ModelField(**v1_kwargs) # type: ignore[no-any-return] except RuntimeError: raise fastapi.exceptions.FastAPIError(_invalid_args_message) from None elif PYDANTIC_V2: @@ -123,7 +128,7 @@ def create_model_field( # Pydantic v2 is not installed, but it's not a Pydantic v1 ModelField, it could be # a Pydantic v1 type, like a constrained int try: - return v1.ModelField(**v1_kwargs) # type: ignore[no-any-return] + return _get_v1().ModelField(**v1_kwargs) # type: ignore[no-any-return] except RuntimeError: raise fastapi.exceptions.FastAPIError(_invalid_args_message) from None @@ -147,11 +152,11 @@ def create_cloned_field( if is_dataclass(original_type) and hasattr(original_type, "__pydantic_model__"): original_type = original_type.__pydantic_model__ use_type = original_type - if lenient_issubclass(original_type, v1.BaseModel): - original_type = cast(Type[v1.BaseModel], original_type) + if lenient_issubclass(original_type, _get_v1().BaseModel): + original_type = cast(Type[_get_v1().BaseModel], original_type) use_type = cloned_types.get(original_type) if use_type is None: - use_type = 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( diff --git a/tests/test_pydantic_v2_first_compat.py b/tests/test_pydantic_v2_first_compat.py new file mode 100644 index 000000000..be85f39a2 --- /dev/null +++ b/tests/test_pydantic_v2_first_compat.py @@ -0,0 +1,208 @@ +# path: tests/test_pydantic_v2_first_compat.py + +""" +Tests for v2-first compatibility layer with lazy v1 loading. + +This test suite validates that: +1. Python 3.14 + Pydantic v2 runs without warnings +2. Python 3.14 + Pydantic v1 shows controlled warnings/errors +3. isinstance() checks work with proxy classes +4. Zero breaking changes for existing imports +""" + +import sys +import warnings +from typing import Any + +import pytest +from fastapi import FastAPI +from fastapi.encoders import jsonable_encoder +from pydantic import BaseModel + + +class TestV2FirstCompatibility: + """Test v2-first compatibility layer.""" + + def test_v2_imports_no_warnings(self): + """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 + + app = FastAPI() + + # 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]}" + + 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') + + # 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 + + # Test that classes are available and callable (expect DeprecationWarning) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", (DeprecationWarning, UserWarning)) + body = Body() + query = Query() + form = Form() + file_param = File() + param = Param() + 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') + + 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 + v1_was_loaded = "pydantic.v1" in sys.modules + + # Import FastAPI components + import fastapi + import fastapi.encoders + from fastapi import FastAPI + + app = FastAPI() + + # 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) + + def test_compat_module_structure(self): + """Test that the compat module has expected structure.""" + # Test that v1 proxy has expected attributes (expect DeprecationWarning) + 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') + + # Test that wrapper functions are available + 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 + app = FastAPI() + + # The strict mode only affects actual v1 usage, not imports + assert True + + finally: + # Restore original value + if original_strict is not None: + os.environ["FASTAPI_PYDANTIC_V1_STRICT"] = original_strict + else: + os.environ.pop("FASTAPI_PYDANTIC_V1_STRICT", None) + + def test_v1_params_composition(self): + """Test that v1 params use composition instead of inheritance.""" + from fastapi._compat._v1_params import Param, Body, Form + + # 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') + + # Test that they have expected interface + assert hasattr(param, 'default') + assert hasattr(body, 'default') + assert hasattr(form, 'default') + + +if __name__ == "__main__": + pytest.main([__file__])