diff --git a/fastapi/_compat/__init__.py b/fastapi/_compat/__init__.py index e938723a9..e7a92efec 100644 --- a/fastapi/_compat/__init__.py +++ b/fastapi/_compat/__init__.py @@ -104,19 +104,24 @@ CoreSchema = Any GetJsonSchemaHandler = Any JsonSchemaValue = dict[str, Any] + def _normalize_errors(errors: Any) -> Any: 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) -> Any: 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}'") + # Explicitly export all compatibility symbols __all__ = [ "PYDANTIC_V2", diff --git a/fastapi/_compat/_v1_params.py b/fastapi/_compat/_v1_params.py index da819de25..91ea2fc27 100644 --- a/fastapi/_compat/_v1_params.py +++ b/fastapi/_compat/_v1_params.py @@ -1,4 +1,3 @@ - """ Internal v1-params shim. Will be removed when v1 is dropped. @@ -13,28 +12,47 @@ from typing import Any _SENTINEL = object() + def _v1() -> Any: """Lazy import of v1 module to avoid warnings.""" from . import v1 # lazy proxy; only warns/errors if actually using v1 + return v1 + class _BaseParam: """Minimal wrapper that delegates to v1.FieldInfo without importing v1 at 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) + # Types used in isinstance() checks in 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): ... diff --git a/fastapi/_compat/_v1_params.pyi b/fastapi/_compat/_v1_params.pyi index f323975a4..2f36a8bdb 100644 --- a/fastapi/_compat/_v1_params.pyi +++ b/fastapi/_compat/_v1_params.pyi @@ -1,29 +1,44 @@ - from typing import Any class Param: - def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ... + def __init__( + self, annotation: Any = ..., default: Any = ..., **kwargs: Any + ) -> None: ... in_: Any class Body(Param): - def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ... + def __init__( + self, annotation: Any = ..., default: Any = ..., **kwargs: Any + ) -> None: ... class Form(Body): - def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ... + def __init__( + self, annotation: Any = ..., default: Any = ..., **kwargs: Any + ) -> None: ... class File(Form): - def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ... + def __init__( + self, annotation: Any = ..., default: Any = ..., **kwargs: Any + ) -> None: ... class Path(Param): - def __init__(self, annotation: Any = ..., default: Any = ..., alias: str = ..., **kwargs: Any) -> None: ... + def __init__( + self, annotation: Any = ..., default: Any = ..., alias: str = ..., **kwargs: Any + ) -> None: ... alias: str default: Any class Query(Param): - def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ... + def __init__( + self, annotation: Any = ..., default: Any = ..., **kwargs: Any + ) -> None: ... class Header(Param): - def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ... + def __init__( + self, annotation: Any = ..., default: Any = ..., **kwargs: Any + ) -> None: ... class Cookie(Param): - def __init__(self, annotation: Any = ..., default: Any = ..., **kwargs: Any) -> None: ... + def __init__( + self, annotation: Any = ..., default: Any = ..., **kwargs: Any + ) -> None: ... diff --git a/fastapi/_compat/lazy_import.py b/fastapi/_compat/lazy_import.py index f0b66883f..08c508dc9 100644 --- a/fastapi/_compat/lazy_import.py +++ b/fastapi/_compat/lazy_import.py @@ -8,7 +8,8 @@ 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 +18,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: Any) -> Any: """ Decorator that only executes function if pydantic.v1 is loaded. """ + def wrapper(*args: Any, **kwargs: Any) -> Any: 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 +48,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 +58,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: Any, **kwargs: Any) -> Any: """ Call a v1 method only if v1 is loaded. @@ -65,6 +74,7 @@ def v1_call_method(method_name: str, *args: Any, **kwargs: Any) -> 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 383822ec6..74ed03795 100644 --- a/fastapi/_compat/main.py +++ b/fastapi/_compat/main.py @@ -34,6 +34,7 @@ def get_cached_model_fields(model: Type[BaseModel]) -> Any: return v1.get_model_fields(model) from . import v2 + return v2.get_model_fields(model) @@ -42,6 +43,7 @@ def _is_undefined(value: object) -> bool: return True elif PYDANTIC_V2: from pydantic_core import PydanticUndefined + return value is PydanticUndefined else: return False @@ -56,6 +58,7 @@ def _get_model_config(model: BaseModel) -> Any: if PYDANTIC_V2: from . import v2 + return v2._get_model_config(model) return getattr(model, "__config__", None) @@ -72,6 +75,7 @@ def _model_dump( if PYDANTIC_V2: from . import v2 + return v2._model_dump(model, mode=mode, **kwargs) return model.dict(**kwargs) @@ -96,12 +100,11 @@ def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo: return v1.copy_field_info(field_info=field_info, annotation=annotation) from . import v2 + return v2.copy_field_info(field_info=field_info, annotation=annotation) -def create_body_model( - *, fields: List[ModelField], model_name: str -) -> Any: +def create_body_model(*, fields: List[ModelField], model_name: str) -> Any: if fields and v1_isinstance(fields[0], "ModelField"): v1 = get_v1_if_loaded() if v1 is None: @@ -109,6 +112,7 @@ def create_body_model( return v1.create_body_model(fields=fields, model_name=model_name) from . import v2 + return v2.create_body_model(fields=fields, model_name=model_name) @@ -124,6 +128,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 ) @@ -137,6 +142,7 @@ def is_bytes_field(field: ModelField) -> Any: return v1.is_bytes_field(field) from . import v2 + return v2.is_bytes_field(field) @@ -148,6 +154,7 @@ def is_bytes_sequence_field(field: ModelField) -> Any: return v1.is_bytes_sequence_field(field) from . import v2 + return v2.is_bytes_sequence_field(field) @@ -159,6 +166,7 @@ def is_scalar_field(field: ModelField) -> Any: return v1.is_scalar_field(field) from . import v2 + return v2.is_scalar_field(field) @@ -170,6 +178,7 @@ def is_scalar_sequence_field(field: ModelField) -> Any: return v1.is_scalar_sequence_field(field) from . import v2 + return v2.is_scalar_sequence_field(field) @@ -181,6 +190,7 @@ def is_sequence_field(field: ModelField) -> Any: return v1.is_sequence_field(field) from . import v2 + return v2.is_sequence_field(field) @@ -192,18 +202,30 @@ def serialize_sequence_value(*, field: ModelField, value: Any) -> Any: return v1.serialize_sequence_value(field=field, value=value) from . import v2 + return v2.serialize_sequence_value(field=field, value=value) def get_compat_model_name_map(fields: List[ModelField]) -> Any: 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 @@ -221,7 +243,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, @@ -233,7 +257,10 @@ def get_definitions( v1_definitions = {} 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, @@ -266,6 +293,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, @@ -279,6 +307,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 @@ -289,6 +318,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 @@ -302,12 +332,16 @@ def get_missing_field_error(loc: Tuple[str, ...], field: ModelField) -> Any: 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() @@ -334,7 +368,9 @@ def with_info_plain_validator_function( v1 = get_v1_if_loaded() if v1 is None: return func - 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: Any) -> None: @@ -344,6 +380,7 @@ def _model_rebuild(model: Any) -> 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 4d906c691..023b5bba2 100644 --- a/fastapi/_compat/shared.py +++ b/fastapi/_compat/shared.py @@ -205,7 +205,9 @@ def annotation_is_pydantic_v1(annotation: Any) -> bool: 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 985d42075..17f6c7f96 100644 --- a/fastapi/_compat/v1.py +++ b/fastapi/_compat/v1.py @@ -1,4 +1,3 @@ - """ Pydantic V1 compatibility module with lazy loading and controlled warnings. @@ -26,6 +25,7 @@ RequiredParam = Ellipsis _pv1 = None _warned = False + def _load() -> Any: global _pv1, _warned if _pv1 is not None: @@ -42,6 +42,7 @@ def _load() -> Any: _pv1 = importlib.import_module("pydantic.v1") return _pv1 + def __getattr__(name: str) -> Any: if name == "RequiredParam": return Ellipsis @@ -50,92 +51,176 @@ def __getattr__(name: str) -> Any: if hasattr(mod, name): return getattr(mod, name) # try common submodules - 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 used by FastAPI core (minimal) ---------- + 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: Any) -> None: model.update_forward_refs() -def _model_dump(model: Any, mode: Literal["json","python"]="json", **kwargs: Any) -> Any: + +def _model_dump( + model: Any, mode: Literal["json", "python"] = "json", **kwargs: Any +) -> Any: return model.dict(**kwargs) + def _get_model_config(model: Any) -> Any: return getattr(model, "__config__", None) -def get_schema_from_model_field(*, field: Any, model_name_map: Any, 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: Any, + model_name_map: Any, + 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: Any, separate_input_output_schemas: bool=True) -> Any: + +def get_definitions( + *, + fields: List[Any], + model_name_map: Any, + separate_input_output_schemas: bool = True, +) -> Any: 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: Any) -> List[Any]: return list(getattr(model, "__fields__", {}).values()) + def is_bytes_field(field: Any) -> bool: return _load().utils.lenient_issubclass(field.type_, bytes) + def is_bytes_sequence_field(field: Any) -> 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: Any) -> 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: Any) -> 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: Any) -> 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: Any, annotation: Any) -> Any: return _copy(field_info) + def serialize_sequence_value(*, field: Any, value: Any) -> Any: 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] diff --git a/fastapi/_compat/v1.pyi b/fastapi/_compat/v1.pyi index 42dccbf46..2532fe390 100644 --- a/fastapi/_compat/v1.pyi +++ b/fastapi/_compat/v1.pyi @@ -16,12 +16,24 @@ class Validator: ... def create_model(__name__: str, **kwargs: Any) -> Any: ... def _model_rebuild(model: Any) -> None: ... -def _model_dump(model: Any, mode: Literal["json","python"]="json", **kwargs: Any) -> Any: ... +def _model_dump( + model: Any, mode: Literal["json", "python"] = "json", **kwargs: Any +) -> Any: ... def _get_model_config(model: Any) -> Any: ... def _normalize_errors(errors: Any) -> Any: ... -def _regenerate_error_with_loc(*, errors: Any, loc_prefix: tuple[Any, ...]) -> list[dict[str, Any]]: ... -def get_schema_from_model_field(*, field: Any, model_name_map: Any, field_mapping: Any, separate_input_output_schemas: bool=True) -> dict[str, Any]: ... -def get_definitions(*, fields: Any, model_name_map: Any, separate_input_output_schemas: bool=True) -> tuple[Any, dict[str, dict[str, Any]]]: ... +def _regenerate_error_with_loc( + *, errors: Any, loc_prefix: tuple[Any, ...] +) -> list[dict[str, Any]]: ... +def get_schema_from_model_field( + *, + field: Any, + model_name_map: Any, + field_mapping: Any, + separate_input_output_schemas: bool = True, +) -> dict[str, Any]: ... +def get_definitions( + *, fields: Any, model_name_map: Any, separate_input_output_schemas: bool = True +) -> tuple[Any, dict[str, dict[str, Any]]]: ... def get_model_fields(model: Any) -> list[Any]: ... def is_bytes_field(field: Any) -> bool: ... def is_bytes_sequence_field(field: Any) -> bool: ... @@ -30,8 +42,12 @@ def is_scalar_sequence_field(field: Any) -> bool: ... def is_sequence_field(field: Any) -> bool: ... def copy_field_info(field_info: Any, annotation: Any, **kwargs: Any) -> Any: ... def create_body_model(fields: Any, model_name: str) -> Any: ... -def evaluate_forwardref(type_: Any, globalns: dict[str, Any], localns: dict[str, Any]) -> Any: ... -def get_annotation_from_field_info(annotation: Any, field_info: Any, field_name: str) -> Any: ... +def evaluate_forwardref( + type_: Any, globalns: dict[str, Any], localns: dict[str, Any] +) -> Any: ... +def get_annotation_from_field_info( + annotation: Any, field_info: Any, field_name: str +) -> Any: ... def get_missing_field_error(loc: tuple[str, ...], field: Any) -> dict[str, Any]: ... def serialize_sequence_value(*, field: Any, value: Any) -> Any: ... def with_info_plain_validator_function(func: Any) -> Any: ... diff --git a/fastapi/_compat/v2.py b/fastapi/_compat/v2.py index 313bd536e..bf81aff1d 100644 --- a/fastapi/_compat/v2.py +++ b/fastapi/_compat/v2.py @@ -40,8 +40,10 @@ from typing_extensions import Annotated, Literal, get_args, get_origin 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/_compat/v2.pyi b/fastapi/_compat/v2.pyi index afcf1eb46..153cbcb10 100644 --- a/fastapi/_compat/v2.pyi +++ b/fastapi/_compat/v2.pyi @@ -12,15 +12,27 @@ class Validator: ... def _is_model_class(value: Any) -> bool: ... def _model_rebuild(model: Any) -> None: ... -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: ... def _get_model_config(model: Any) -> Any: ... def _model_dump(model: Any, **kwargs: Any) -> Any: ... def copy_field_info(field_info: Any, annotation: Any, **kwargs: Any) -> Any: ... def create_body_model(fields: Any, model_name: str) -> Any: ... -def get_annotation_from_field_info(annotation: Any, field_info: Any, field_name: str) -> Any: ... -def get_definitions(*, fields: Any, model_name_map: Any, separate_input_output_schemas: bool=True) -> tuple[Any, dict[str, dict[str, Any]]]: ... +def get_annotation_from_field_info( + annotation: Any, field_info: Any, field_name: str +) -> Any: ... +def get_definitions( + *, fields: Any, model_name_map: Any, separate_input_output_schemas: bool = True +) -> tuple[Any, dict[str, dict[str, Any]]]: ... def get_missing_field_error(loc: tuple[str, ...], field: Any) -> dict[str, Any]: ... -def get_schema_from_model_field(*, field: Any, model_name_map: Any, field_mapping: Any, separate_input_output_schemas: bool=True) -> dict[str, Any]: ... +def get_schema_from_model_field( + *, + field: Any, + model_name_map: Any, + field_mapping: Any, + separate_input_output_schemas: bool = True, +) -> dict[str, Any]: ... def is_bytes_field(field: Any) -> bool: ... def is_bytes_sequence_field(field: Any) -> bool: ... def is_scalar_field(field: Any) -> bool: ... diff --git a/fastapi/datastructures.py b/fastapi/datastructures.py index db2ee9d3d..d096abbb4 100644 --- a/fastapi/datastructures.py +++ b/fastapi/datastructures.py @@ -168,6 +168,7 @@ class UploadFile(StarletteUploadFile): cls, source: Type[Any], handler: Callable[[Any], CoreSchema] ) -> CoreSchema: from fastapi._compat.v2 import with_info_plain_validator_function # noqa: I001 + return with_info_plain_validator_function(cls._validate) diff --git a/fastapi/dependencies/utils.py b/fastapi/dependencies/utils.py index 3053fb511..6941f5a5d 100644 --- a/fastapi/dependencies/utils.py +++ b/fastapi/dependencies/utils.py @@ -83,8 +83,10 @@ from .._compat import _v1_params as temp_pydantic_v1_params def _get_v1() -> Any: """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 @@ -531,7 +533,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, # type: ignore[arg-type] ) if is_path_param: diff --git a/fastapi/encoders.py b/fastapi/encoders.py index 4b48155f8..dbe374420 100644 --- a/fastapi/encoders.py +++ b/fastapi/encoders.py @@ -86,6 +86,7 @@ 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).""" # Only register if pydantic.v1 is already loaded (app actually used v1) @@ -236,7 +237,7 @@ def jsonable_encoder( # Check if it's a v1 model using lazy loading if "pydantic.v1" in sys.modules: if isinstance(obj, v1.BaseModel): - encoders = getattr(obj.__config__, "json_encoders", {}) # type: ignore[attr-defined] + encoders = getattr(obj.__config__, "json_encoders", {}) # type: ignore[attr-defined] if custom_encoder: encoders = {**encoders, **custom_encoder} obj_dict = _model_dump( diff --git a/fastapi/utils.py b/fastapi/utils.py index d158681e6..1b4f419b3 100644 --- a/fastapi/utils.py +++ b/fastapi/utils.py @@ -33,12 +33,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() @@ -146,7 +149,9 @@ def create_cloned_field( original_type = cast(Type[Any], 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 b09d87c1f..bbe71365e 100644 --- a/tests/test_pydantic_v2_first_compat.py +++ b/tests/test_pydantic_v2_first_compat.py @@ -45,21 +45,23 @@ class TestV2FirstCompatibility: 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 # 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(): @@ -69,6 +71,7 @@ class TestV2FirstCompatibility: # 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 @@ -99,14 +102,14 @@ class TestV2FirstCompatibility: 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.""" @@ -124,6 +127,7 @@ class TestV2FirstCompatibility: # Test that we can still access v1 proxy from fastapi._compat import v1 + assert v1 is not None def test_encoders_lazy_registration(self): @@ -148,14 +152,14 @@ class TestV2FirstCompatibility: 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.""" @@ -170,6 +174,7 @@ class TestV2FirstCompatibility: # This should not raise an error unless we actually try to use v1 from fastapi import FastAPI + _ = FastAPI() # Create app but don't use it # The strict mode only affects actual v1 usage, not imports @@ -194,14 +199,14 @@ class TestV2FirstCompatibility: 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__":