From 3c8db03107d0a21a1e2d8073d07edce09edc6375 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Sebasti=C3=A1n=20Ram=C3=ADrez?= Date: Fri, 26 Dec 2025 20:09:36 +0100 Subject: [PATCH] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Refactor=20=5Fcompat,=20re?= =?UTF-8?q?move=20Pydantic=20v1,=20first=20pass?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- fastapi/_compat/__init__.py | 6 +- fastapi/_compat/main.py | 116 ++--------------- fastapi/_compat/may_v1.py | 106 +--------------- fastapi/_compat/model_field.py | 50 -------- fastapi/_compat/shared.py | 4 +- fastapi/_compat/v1.py | 222 --------------------------------- fastapi/_compat/v2.py | 16 +-- 7 files changed, 15 insertions(+), 505 deletions(-) delete mode 100644 fastapi/_compat/model_field.py delete mode 100644 fastapi/_compat/v1.py diff --git a/fastapi/_compat/__init__.py b/fastapi/_compat/__init__.py index fd1df8c6a..43f3e7d7f 100644 --- a/fastapi/_compat/__init__.py +++ b/fastapi/_compat/__init__.py @@ -6,7 +6,6 @@ from .main import UndefinedType as UndefinedType from .main import Url as Url from .main import Validator as Validator from .main import _get_model_config as _get_model_config -from .main import _is_error_wrapper as _is_error_wrapper from .main import _is_model_class as _is_model_class from .main import _is_model_field as _is_model_field from .main import _is_undefined as _is_undefined @@ -29,11 +28,7 @@ from .main import serialize_sequence_value as serialize_sequence_value from .main import ( with_info_plain_validator_function as with_info_plain_validator_function, ) -from .may_v1 import CoreSchema as CoreSchema -from .may_v1 import GetJsonSchemaHandler as GetJsonSchemaHandler -from .may_v1 import JsonSchemaValue as JsonSchemaValue from .may_v1 import _normalize_errors as _normalize_errors -from .model_field import ModelField as ModelField from .shared import PYDANTIC_V2 as PYDANTIC_V2 from .shared import PYDANTIC_VERSION_MINOR_TUPLE as PYDANTIC_VERSION_MINOR_TUPLE from .shared import annotation_is_pydantic_v1 as annotation_is_pydantic_v1 @@ -47,3 +42,4 @@ 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 .v2 import ModelField as ModelField diff --git a/fastapi/_compat/main.py b/fastapi/_compat/main.py index 95053a237..1b3c9367c 100644 --- a/fastapi/_compat/main.py +++ b/fastapi/_compat/main.py @@ -12,9 +12,9 @@ from pydantic import BaseModel from typing_extensions import Literal from . import v2 -from .model_field import ModelField from .v2 import BaseConfig as BaseConfig from .v2 import FieldInfo as FieldInfo +from .v2 import ModelField from .v2 import PydanticSchemaGenerationError as PydanticSchemaGenerationError from .v2 import RequiredParam as RequiredParam from .v2 import Undefined as Undefined @@ -23,6 +23,7 @@ from .v2 import Url as Url from .v2 import Validator as Validator from .v2 import evaluate_forwardref as evaluate_forwardref from .v2 import get_missing_field_error as get_missing_field_error +from .v2 import get_model_fields as get_model_fields from .v2 import ( with_info_plain_validator_function as with_info_plain_validator_function, ) @@ -30,109 +31,50 @@ from .v2 import ( @lru_cache def get_cached_model_fields(model: type[BaseModel]) -> list[ModelField]: - if lenient_issubclass(model, may_v1.BaseModel): - from fastapi._compat import v1 - - return v1.get_model_fields(model) # type: ignore[arg-type,return-value] - else: - from . import v2 - - return v2.get_model_fields(model) # type: ignore[return-value] + return get_model_fields(model) # type: ignore[return-value] def _is_undefined(value: object) -> bool: - if isinstance(value, may_v1.UndefinedType): - return True - return isinstance(value, v2.UndefinedType) def _get_model_config(model: BaseModel) -> Any: - if isinstance(model, may_v1.BaseModel): - from fastapi._compat import v1 - - return v1._get_model_config(model) - return v2._get_model_config(model) def _model_dump( model: BaseModel, mode: Literal["json", "python"] = "json", **kwargs: Any ) -> Any: - if isinstance(model, may_v1.BaseModel): - from fastapi._compat import v1 - - return v1._model_dump(model, mode=mode, **kwargs) - return v2._model_dump(model, mode=mode, **kwargs) -def _is_error_wrapper(exc: Exception) -> bool: - if isinstance(exc, may_v1.ErrorWrapper): - return True - - return isinstance(exc, v2.ErrorWrapper) - - def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo: - if isinstance(field_info, may_v1.FieldInfo): - from fastapi._compat import v1 - - return v1.copy_field_info(field_info=field_info, annotation=annotation) - return v2.copy_field_info(field_info=field_info, annotation=annotation) def create_body_model( *, fields: Sequence[ModelField], model_name: str ) -> type[BaseModel]: - if fields and isinstance(fields[0], may_v1.ModelField): - from fastapi._compat import v1 - - return v1.create_body_model(fields=fields, model_name=model_name) - return v2.create_body_model(fields=fields, model_name=model_name) # type: ignore[arg-type] def get_annotation_from_field_info( annotation: Any, field_info: FieldInfo, field_name: str ) -> Any: - if isinstance(field_info, may_v1.FieldInfo): - from fastapi._compat import v1 - - return v1.get_annotation_from_field_info( - annotation=annotation, field_info=field_info, field_name=field_name - ) - 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, may_v1.ModelField): - from fastapi._compat import v1 - - return v1.is_bytes_field(field) - return v2.is_bytes_field(field) # type: ignore[arg-type] def is_bytes_sequence_field(field: ModelField) -> bool: - if isinstance(field, may_v1.ModelField): - from fastapi._compat import v1 - - return v1.is_bytes_sequence_field(field) - return v2.is_bytes_sequence_field(field) # type: ignore[arg-type] def is_scalar_field(field: ModelField) -> bool: - if isinstance(field, may_v1.ModelField): - from fastapi._compat import v1 - - return v1.is_scalar_field(field) - return v2.is_scalar_field(field) # type: ignore[arg-type] @@ -141,37 +83,15 @@ def is_scalar_sequence_field(field: ModelField) -> bool: def is_sequence_field(field: ModelField) -> bool: - if isinstance(field, may_v1.ModelField): - from fastapi._compat import v1 - - return v1.is_sequence_field(field) - return v2.is_sequence_field(field) # type: ignore[arg-type] def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]: - if isinstance(field, may_v1.ModelField): - from fastapi._compat import v1 - - return v1.serialize_sequence_value(field=field, value=value) - return v2.serialize_sequence_value(field=field, value=value) # type: ignore[arg-type] def get_compat_model_name_map(fields: list[ModelField]) -> ModelNameMap: - v1_model_fields = [ - field for field in fields if isinstance(field, may_v1.ModelField) - ] - if v1_model_fields: - from fastapi._compat import v1 - - v1_flat_models = v1.get_flat_models_from_fields( - v1_model_fields, # type: ignore[arg-type] - known_models=set(), - ) - all_flat_models = v1_flat_models - else: - all_flat_models = set() + all_flat_models = set() 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()) @@ -187,28 +107,18 @@ def get_definitions( model_name_map: ModelNameMap, separate_input_output_schemas: bool = True, ) -> tuple[ - dict[ - tuple[ModelField, Literal["validation", "serialization"]], - may_v1.JsonSchemaValue, - ], + dict[tuple[ModelField, Literal["validation", "serialization"]], dict[str, Any]], dict[str, dict[str, Any]], ]: if sys.version_info < (3, 14): - v1_fields = [field for field in fields if isinstance(field, may_v1.ModelField)] - v1_field_maps, v1_definitions = may_v1.get_definitions( - fields=v1_fields, # type: ignore[arg-type] - model_name_map=model_name_map, - separate_input_output_schemas=separate_input_output_schemas, - ) - v2_fields = [field for field in fields if isinstance(field, v2.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} # type: ignore[misc] + all_definitions = {**v2_definitions} + all_field_maps = {**v2_field_maps} # type: ignore[misc] return all_field_maps, all_definitions # Pydantic v1 is not supported since Python 3.14 @@ -228,20 +138,10 @@ def get_schema_from_model_field( model_name_map: ModelNameMap, field_mapping: dict[ tuple[ModelField, Literal["validation", "serialization"]], - may_v1.JsonSchemaValue, + dict[str, Any], ], separate_input_output_schemas: bool = True, ) -> dict[str, Any]: - if isinstance(field, may_v1.ModelField): - from fastapi._compat import v1 - - return v1.get_schema_from_model_field( - field=field, - model_name_map=model_name_map, - field_mapping=field_mapping, - separate_input_output_schemas=separate_input_output_schemas, - ) - return v2.get_schema_from_model_field( field=field, # type: ignore[arg-type] model_name_map=model_name_map, diff --git a/fastapi/_compat/may_v1.py b/fastapi/_compat/may_v1.py index 3ac86aa98..6e3b224a2 100644 --- a/fastapi/_compat/may_v1.py +++ b/fastapi/_compat/may_v1.py @@ -1,112 +1,14 @@ -import sys from collections.abc import Sequence -from typing import Any, Literal, Union +from typing import Any, Union -from fastapi.types import ModelNameMap - -if sys.version_info >= (3, 14): - - class AnyUrl: - pass - - class BaseConfig: - pass - - class BaseModel: - pass - - class Color: - pass - - class CoreSchema: - pass - - class ErrorWrapper: - pass - - class FieldInfo: - pass - - class GetJsonSchemaHandler: - pass - - class JsonSchemaValue: - pass - - class ModelField: - pass - - class NameEmail: - pass - - class RequiredParam: - pass - - class SecretBytes: - pass - - class SecretStr: - pass - - class Undefined: - pass - - class UndefinedType: - pass - - class Url: - pass - - from .v2 import ValidationError, create_model - - 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]], - ]: - return {}, {} # pragma: no cover - - -else: - from .v1 import AnyUrl as AnyUrl - from .v1 import BaseConfig as BaseConfig - from .v1 import BaseModel as BaseModel - from .v1 import Color as Color - from .v1 import CoreSchema as CoreSchema - from .v1 import ErrorWrapper as ErrorWrapper - from .v1 import FieldInfo as FieldInfo - from .v1 import GetJsonSchemaHandler as GetJsonSchemaHandler - from .v1 import JsonSchemaValue as JsonSchemaValue - from .v1 import ModelField as ModelField - from .v1 import NameEmail as NameEmail - from .v1 import RequiredParam as RequiredParam - from .v1 import SecretBytes as SecretBytes - from .v1 import SecretStr as SecretStr - from .v1 import Undefined as Undefined - from .v1 import UndefinedType as UndefinedType - from .v1 import Url as Url - from .v1 import ValidationError, create_model - from .v1 import get_definitions as get_definitions - - -RequestErrorModel: type[BaseModel] = create_model("Request") +from pydantic.v1 import BaseModel as BaseModel +from pydantic.v1.fields import ModelField as ModelField 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( - errors=[error], model=RequestErrorModel - ).errors() - use_errors.extend(new_errors) - elif isinstance(error, list): + if isinstance(error, list): use_errors.extend(_normalize_errors(error)) else: use_errors.append(error) diff --git a/fastapi/_compat/model_field.py b/fastapi/_compat/model_field.py deleted file mode 100644 index 47d05cb94..000000000 --- a/fastapi/_compat/model_field.py +++ /dev/null @@ -1,50 +0,0 @@ -from typing import ( - Any, - Union, -) - -from fastapi.types import IncEx -from pydantic.fields import FieldInfo -from typing_extensions import Literal, Protocol - - -class ModelField(Protocol): - field_info: "FieldInfo" - name: str - mode: Literal["validation", "serialization"] = "validation" - _version: Literal["v1", "v2"] = "v1" - - @property - def alias(self) -> str: ... - - @property - def required(self) -> bool: ... - - @property - def default(self) -> Any: ... - - @property - def type_(self) -> Any: ... - - def get_default(self) -> Any: ... - - def validate( - self, - value: Any, - values: dict[str, Any] = {}, # noqa: B006 - *, - loc: tuple[Union[int, str], ...] = (), - ) -> tuple[Any, Union[list[dict[str, Any]], None]]: ... - - def serialize( - self, - value: Any, - *, - mode: Literal["json", "python"] = "json", - include: Union[IncEx, None] = None, - exclude: Union[IncEx, None] = None, - by_alias: bool = True, - exclude_unset: bool = False, - exclude_defaults: bool = False, - exclude_none: bool = False, - ) -> Any: ... diff --git a/fastapi/_compat/shared.py b/fastapi/_compat/shared.py index 3a11e88ac..2e0b5fd7b 100644 --- a/fastapi/_compat/shared.py +++ b/fastapi/_compat/shared.py @@ -81,9 +81,7 @@ def value_is_sequence(value: Any) -> bool: def _annotation_is_complex(annotation: Union[type[Any], None]) -> bool: return ( - lenient_issubclass( - annotation, (BaseModel, may_v1.BaseModel, Mapping, UploadFile) - ) + lenient_issubclass(annotation, (BaseModel, Mapping, UploadFile)) or _annotation_is_sequence(annotation) or is_dataclass(annotation) ) diff --git a/fastapi/_compat/v1.py b/fastapi/_compat/v1.py deleted file mode 100644 index b0a9dd35f..000000000 --- a/fastapi/_compat/v1.py +++ /dev/null @@ -1,222 +0,0 @@ -from collections.abc import Sequence -from copy import copy -from dataclasses import dataclass, is_dataclass -from enum import Enum -from typing import ( - Any, - Callable, - Union, -) - -from fastapi._compat import shared -from fastapi.openapi.constants import REF_PREFIX as REF_PREFIX -from fastapi.types import ModelNameMap -from pydantic.v1 import BaseConfig as BaseConfig -from pydantic.v1 import BaseModel as BaseModel -from pydantic.v1 import ValidationError as ValidationError -from pydantic.v1 import create_model as create_model -from pydantic.v1.class_validators import Validator as Validator -from pydantic.v1.color import Color as Color -from pydantic.v1.error_wrappers import ErrorWrapper as ErrorWrapper -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 -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 NameEmail as NameEmail -from pydantic.v1.schema import TypeModelSet as TypeModelSet -from pydantic.v1.schema import field_schema, 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_flat_models_from_fields as get_flat_models_from_fields, -) -from pydantic.v1.schema import get_model_name_map as get_model_name_map -from pydantic.v1.types import SecretBytes as SecretBytes -from pydantic.v1.types import SecretStr as SecretStr -from pydantic.v1.typing import evaluate_forwardref as evaluate_forwardref -from pydantic.v1.utils import lenient_issubclass as lenient_issubclass -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 - - -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 - - -RequestErrorModel: type[BaseModel] = create_model("Request") - - -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 _model_dump( - model: BaseModel, mode: Literal["json", "python"] = "json", **kwargs: Any -) -> Any: - return model.dict(**kwargs) - - -def _get_model_config(model: BaseModel) -> Any: - return model.__config__ - - -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( - field, - model_name_map=model_name_map, # type: ignore[arg-type] - 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) # type: ignore[arg-type] - - -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_bytes_field(field: ModelField) -> bool: - return lenient_issubclass(field.type_, bytes) - - -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 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 - return BodyModel - - -def get_model_fields(model: type[BaseModel]) -> list[ModelField]: - return list(model.__fields__.values()) diff --git a/fastapi/_compat/v2.py b/fastapi/_compat/v2.py index cbcb98e1a..df0f461a5 100644 --- a/fastapi/_compat/v2.py +++ b/fastapi/_compat/v2.py @@ -503,19 +503,9 @@ def normalize_name(name: str) -> str: def get_model_name_map(unique_models: TypeModelSet) -> dict[TypeModelOrEnum, str]: name_model_map = {} - conflicting_names: set[str] = set() for model in unique_models: model_name = normalize_name(model.__name__) - if model_name in conflicting_names: - model_name = get_long_model_name(model) - name_model_map[model_name] = model - elif model_name in name_model_map: - conflicting_names.add(model_name) - conflicting_model = name_model_map.pop(model_name) - name_model_map[get_long_model_name(conflicting_model)] = conflicting_model - name_model_map[get_long_model_name(model)] = model - else: - name_model_map[model_name] = model + name_model_map[model_name] = model return {v: k for k, v in name_model_map.items()} @@ -565,7 +555,3 @@ def get_flat_models_from_fields( for field in fields: get_flat_models_from_field(field, known_models=known_models) return known_models - - -def get_long_model_name(model: TypeModelOrEnum) -> str: - return f"{model.__module__}__{model.__qualname__}".replace(".", "__")