Browse Source

♻️ Simplify and unify compat layer

pull/14609/head
Sebastián Ramírez 7 months ago
parent
commit
420ed20e83
  1. 30
      fastapi/_compat/__init__.py
  2. 59
      fastapi/_compat/main.py
  3. 17
      fastapi/_compat/v2.py

30
fastapi/_compat/__init__.py

@ -1,18 +1,3 @@
from .main import BaseConfig as BaseConfig
from .main import PydanticSchemaGenerationError as PydanticSchemaGenerationError
from .main import RequiredParam as RequiredParam
from .main import Undefined as Undefined
from .main import UndefinedType as UndefinedType
from .main import Url as Url
from .main import Validator as Validator
from .main import evaluate_forwardref as evaluate_forwardref
from .main import get_cached_model_fields as get_cached_model_fields
from .main import get_compat_model_name_map as get_compat_model_name_map
from .main import get_missing_field_error as get_missing_field_error
from .main import get_schema_from_model_field as get_schema_from_model_field
from .main import (
with_info_plain_validator_function as with_info_plain_validator_function,
)
from .shared import PYDANTIC_V2 as PYDANTIC_V2 from .shared import PYDANTIC_V2 as PYDANTIC_V2
from .shared import PYDANTIC_VERSION_MINOR_TUPLE as PYDANTIC_VERSION_MINOR_TUPLE from .shared import PYDANTIC_VERSION_MINOR_TUPLE as PYDANTIC_VERSION_MINOR_TUPLE
from .shared import annotation_is_pydantic_v1 as annotation_is_pydantic_v1 from .shared import annotation_is_pydantic_v1 as annotation_is_pydantic_v1
@ -28,14 +13,29 @@ from .shared import (
from .shared import lenient_issubclass as lenient_issubclass from .shared import lenient_issubclass as lenient_issubclass
from .shared import sequence_types as sequence_types from .shared import sequence_types as sequence_types
from .shared import value_is_sequence as value_is_sequence from .shared import value_is_sequence as value_is_sequence
from .v2 import BaseConfig as BaseConfig
from .v2 import ModelField as ModelField from .v2 import ModelField as ModelField
from .v2 import PydanticSchemaGenerationError as PydanticSchemaGenerationError
from .v2 import RequiredParam as RequiredParam
from .v2 import Undefined as Undefined
from .v2 import UndefinedType as UndefinedType
from .v2 import Url as Url
from .v2 import Validator as Validator
from .v2 import _regenerate_error_with_loc as _regenerate_error_with_loc from .v2 import _regenerate_error_with_loc as _regenerate_error_with_loc
from .v2 import copy_field_info as copy_field_info from .v2 import copy_field_info as copy_field_info
from .v2 import create_body_model as create_body_model from .v2 import create_body_model as create_body_model
from .v2 import evaluate_forwardref as evaluate_forwardref
from .v2 import get_cached_model_fields as get_cached_model_fields
from .v2 import get_compat_model_name_map as get_compat_model_name_map
from .v2 import get_definitions as get_definitions 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_field as is_bytes_field
from .v2 import is_bytes_sequence_field as is_bytes_sequence_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_field as is_scalar_field
from .v2 import is_scalar_sequence_field as is_scalar_sequence_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 is_sequence_field as is_sequence_field
from .v2 import serialize_sequence_value as serialize_sequence_value from .v2 import serialize_sequence_value as serialize_sequence_value
from .v2 import (
with_info_plain_validator_function as with_info_plain_validator_function,
)

59
fastapi/_compat/main.py

@ -1,59 +0,0 @@
from functools import lru_cache
from typing import (
Any,
)
from fastapi.types import ModelNameMap
from pydantic import BaseModel
from typing_extensions import Literal
from . import v2
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
from .v2 import UndefinedType as UndefinedType
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,
)
@lru_cache
def get_cached_model_fields(model: type[BaseModel]) -> list[ModelField]:
return get_model_fields(model) # type: ignore[return-value]
def get_compat_model_name_map(fields: list[ModelField]) -> ModelNameMap:
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())
all_flat_models = all_flat_models.union(v2_flat_models) # type: ignore[arg-type]
model_name_map = v2.get_model_name_map(all_flat_models) # type: ignore[arg-type]
return model_name_map
def get_schema_from_model_field(
*,
field: ModelField,
model_name_map: ModelNameMap,
field_mapping: dict[
tuple[ModelField, Literal["validation", "serialization"]],
dict[str, Any],
],
separate_input_output_schemas: bool = True,
) -> dict[str, Any]:
return v2.get_schema_from_model_field(
field=field, # type: ignore[arg-type]
model_name_map=model_name_map,
field_mapping=field_mapping, # type: ignore[arg-type]
separate_input_output_schemas=separate_input_output_schemas,
)

17
fastapi/_compat/v2.py

@ -4,6 +4,7 @@ from collections.abc import Sequence
from copy import copy, deepcopy from copy import copy, deepcopy
from dataclasses import dataclass, is_dataclass from dataclasses import dataclass, is_dataclass
from enum import Enum from enum import Enum
from functools import lru_cache
from typing import ( from typing import (
Annotated, Annotated,
Any, Any,
@ -474,6 +475,11 @@ def get_model_fields(model: type[BaseModel]) -> list[ModelField]:
return model_fields return model_fields
@lru_cache
def get_cached_model_fields(model: type[BaseModel]) -> list[ModelField]:
return get_model_fields(model) # type: ignore[return-value]
# Duplicate of several schema functions from Pydantic v1 to make them compatible with # Duplicate of several schema functions from Pydantic v1 to make them compatible with
# Pydantic v2 and allow mixing the models # Pydantic v2 and allow mixing the models
@ -493,6 +499,17 @@ def get_model_name_map(unique_models: TypeModelSet) -> dict[TypeModelOrEnum, str
return {v: k for k, v in name_model_map.items()} return {v: k for k, v in name_model_map.items()}
def get_compat_model_name_map(fields: list[ModelField]) -> ModelNameMap:
all_flat_models = set()
v2_model_fields = [field for field in fields if isinstance(field, ModelField)]
v2_flat_models = get_flat_models_from_fields(v2_model_fields, known_models=set())
all_flat_models = all_flat_models.union(v2_flat_models) # type: ignore[arg-type]
model_name_map = get_model_name_map(all_flat_models) # type: ignore[arg-type]
return model_name_map
def get_flat_models_from_model( def get_flat_models_from_model(
model: type["BaseModel"], known_models: Union[TypeModelSet, None] = None model: type["BaseModel"], known_models: Union[TypeModelSet, None] = None
) -> TypeModelSet: ) -> TypeModelSet:

Loading…
Cancel
Save