diff --git a/fastapi/_compat/__init__.py b/fastapi/_compat/__init__.py index 76779268d..423036ca6 100644 --- a/fastapi/_compat/__init__.py +++ b/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 newline at end of file + +# No __all__ defined - exports everything implicitly for backward compatibility diff --git a/fastapi/_compat/_v1_params.py b/fastapi/_compat/_v1_params.py index 8fb6b8aa3..0fa5bb675 100644 --- a/fastapi/_compat/_v1_params.py +++ b/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): ... \ No newline at end of file + + +class Cookie(Param): ... diff --git a/fastapi/_compat/lazy_import.py b/fastapi/_compat/lazy_import.py index 270017799..42034539a 100644 --- a/fastapi/_compat/lazy_import.py +++ b/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. diff --git a/fastapi/_compat/main.py b/fastapi/_compat/main.py index f0ffec0bb..001f761be 100644 --- a/fastapi/_compat/main.py +++ b/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() diff --git a/fastapi/_compat/shared.py b/fastapi/_compat/shared.py index d1d47cf99..46daf1f60 100644 --- a/fastapi/_compat/shared.py +++ b/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 diff --git a/fastapi/_compat/v1.py b/fastapi/_compat/v1.py index abdc9a553..1841edbfd 100644 --- a/fastapi/_compat/v1.py +++ b/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 \ No newline at end of file +Url = Any diff --git a/fastapi/_compat/v2.py b/fastapi/_compat/v2.py index 946c4a3e0..bc3ba2df9 100644 --- a/fastapi/_compat/v2.py +++ b/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, diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py index 1b9b12a90..b472cecb2 100644 --- a/fastapi/dependencies/utils.py +++ b/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: diff --git a/fastapi/encoders.py b/fastapi/encoders.py index 0177cceaa..9efb4edbb 100644 --- a/fastapi/encoders.py +++ b/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: diff --git a/fastapi/routing.py b/fastapi/routing.py index 8b8952931..609b890f4 100644 --- a/fastapi/routing.py +++ b/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 ( diff --git a/fastapi/utils.py b/fastapi/utils.py index 1212b60c2..5f249c2d2 100644 --- a/fastapi/utils.py +++ b/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( diff --git a/tests/test_pydantic_v2_first_compat.py b/tests/test_pydantic_v2_first_compat.py index a9810001b..bbe71365e 100644 --- a/tests/test_pydantic_v2_first_compat.py +++ b/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__":