v2.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480
  1. import re
  2. import warnings
  3. from collections.abc import Sequence
  4. from copy import copy
  5. from dataclasses import dataclass, is_dataclass
  6. from enum import Enum
  7. from functools import lru_cache
  8. from typing import (
  9. Annotated,
  10. Any,
  11. Literal,
  12. Union,
  13. cast,
  14. get_args,
  15. get_origin,
  16. )
  17. from fastapi._compat import lenient_issubclass, shared
  18. from fastapi.openapi.constants import REF_TEMPLATE
  19. from fastapi.types import IncEx, ModelNameMap, UnionType
  20. from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, create_model
  21. from pydantic import PydanticSchemaGenerationError as PydanticSchemaGenerationError
  22. from pydantic import PydanticUndefinedAnnotation as PydanticUndefinedAnnotation
  23. from pydantic import ValidationError as ValidationError
  24. from pydantic._internal._schema_generation_shared import ( # type: ignore[attr-defined] # ty: ignore[unused-ignore-comment]
  25. GetJsonSchemaHandler as GetJsonSchemaHandler,
  26. )
  27. from pydantic._internal._typing_extra import eval_type_lenient # ty: ignore[deprecated]
  28. from pydantic.fields import FieldInfo as FieldInfo
  29. from pydantic.json_schema import GenerateJsonSchema as _GenerateJsonSchema
  30. from pydantic.json_schema import JsonSchemaValue as JsonSchemaValue
  31. from pydantic_core import CoreSchema as CoreSchema
  32. from pydantic_core import PydanticUndefined
  33. from pydantic_core import Url as Url
  34. from pydantic_core.core_schema import (
  35. with_info_plain_validator_function as with_info_plain_validator_function,
  36. )
  37. RequiredParam = PydanticUndefined
  38. Undefined = PydanticUndefined
  39. evaluate_forwardref = eval_type_lenient # ty: ignore[deprecated]
  40. class GenerateJsonSchema(_GenerateJsonSchema):
  41. # TODO: remove when this is merged (or equivalent): https://github.com/pydantic/pydantic/pull/12841
  42. # and dropping support for any version of Pydantic before that one (so, in a very long time)
  43. def bytes_schema(self, schema: CoreSchema) -> JsonSchemaValue:
  44. json_schema = {"type": "string", "contentMediaType": "application/octet-stream"}
  45. bytes_mode = (
  46. self._config.ser_json_bytes
  47. if self.mode == "serialization"
  48. else self._config.val_json_bytes
  49. )
  50. if bytes_mode == "base64":
  51. json_schema["contentEncoding"] = "base64"
  52. self.update_with_validations(json_schema, schema, self.ValidationsMapping.bytes)
  53. return json_schema
  54. # TODO: remove when dropping support for Pydantic < v2.12.3
  55. _Attrs = {
  56. "default": ...,
  57. "default_factory": None,
  58. "alias": None,
  59. "alias_priority": None,
  60. "validation_alias": None,
  61. "serialization_alias": None,
  62. "title": None,
  63. "field_title_generator": None,
  64. "description": None,
  65. "examples": None,
  66. "exclude": None,
  67. "exclude_if": None,
  68. "discriminator": None,
  69. "deprecated": None,
  70. "json_schema_extra": None,
  71. "frozen": None,
  72. "validate_default": None,
  73. "repr": True,
  74. "init": None,
  75. "init_var": None,
  76. "kw_only": None,
  77. }
  78. # TODO: remove when dropping support for Pydantic < v2.12.3
  79. def asdict(field_info: FieldInfo) -> dict[str, Any]:
  80. attributes = {}
  81. for attr in _Attrs:
  82. value = getattr(field_info, attr, Undefined)
  83. if value is not Undefined:
  84. attributes[attr] = value
  85. return {
  86. "annotation": field_info.annotation,
  87. "metadata": field_info.metadata,
  88. "attributes": attributes,
  89. }
  90. @dataclass
  91. class ModelField:
  92. field_info: FieldInfo
  93. name: str
  94. mode: Literal["validation", "serialization"] = "validation"
  95. config: ConfigDict | None = None
  96. @property
  97. def alias(self) -> str:
  98. a = self.field_info.alias
  99. return a if a is not None else self.name
  100. @property
  101. def validation_alias(self) -> str | None:
  102. va = self.field_info.validation_alias
  103. if isinstance(va, str) and va:
  104. return va
  105. return None
  106. @property
  107. def serialization_alias(self) -> str | None:
  108. sa = self.field_info.serialization_alias
  109. return sa or None
  110. @property
  111. def default(self) -> Any:
  112. return self.get_default()
  113. def __post_init__(self) -> None:
  114. with warnings.catch_warnings():
  115. # Pydantic >= 2.12.0 warns about field specific metadata that is unused
  116. # (e.g. `TypeAdapter(Annotated[int, Field(alias='b')])`). In some cases, we
  117. # end up building the type adapter from a model field annotation so we
  118. # need to ignore the warning:
  119. if shared.PYDANTIC_VERSION_MINOR_TUPLE >= (2, 12):
  120. from pydantic.warnings import UnsupportedFieldAttributeWarning
  121. warnings.simplefilter(
  122. "ignore", category=UnsupportedFieldAttributeWarning
  123. )
  124. # TODO: remove after setting the min Pydantic to v2.12.3
  125. # that adds asdict(), and use self.field_info.asdict() instead
  126. field_dict = asdict(self.field_info)
  127. annotated_args = (
  128. field_dict["annotation"],
  129. *field_dict["metadata"],
  130. # this FieldInfo needs to be created again so that it doesn't include
  131. # the old field info metadata and only the rest of the attributes
  132. Field(**field_dict["attributes"]),
  133. )
  134. self._type_adapter: TypeAdapter[Any] = TypeAdapter(
  135. Annotated[annotated_args], # ty: ignore[invalid-type-form]
  136. config=self.config,
  137. )
  138. def get_default(self) -> Any:
  139. if self.field_info.is_required():
  140. return Undefined
  141. return self.field_info.get_default(call_default_factory=True)
  142. def validate(
  143. self,
  144. value: Any,
  145. values: dict[str, Any] = {}, # noqa: B006
  146. *,
  147. loc: tuple[int | str, ...] = (),
  148. ) -> tuple[Any, list[dict[str, Any]]]:
  149. try:
  150. return (
  151. self._type_adapter.validate_python(value, from_attributes=True),
  152. [],
  153. )
  154. except ValidationError as exc:
  155. return None, _regenerate_error_with_loc(
  156. errors=exc.errors(include_url=False), loc_prefix=loc
  157. )
  158. def serialize(
  159. self,
  160. value: Any,
  161. *,
  162. mode: Literal["json", "python"] = "json",
  163. include: IncEx | None = None,
  164. exclude: IncEx | None = None,
  165. by_alias: bool = True,
  166. exclude_unset: bool = False,
  167. exclude_defaults: bool = False,
  168. exclude_none: bool = False,
  169. ) -> Any:
  170. # What calls this code passes a value that already called
  171. # self._type_adapter.validate_python(value)
  172. return self._type_adapter.dump_python(
  173. value,
  174. mode=mode,
  175. include=include,
  176. exclude=exclude,
  177. by_alias=by_alias,
  178. exclude_unset=exclude_unset,
  179. exclude_defaults=exclude_defaults,
  180. exclude_none=exclude_none,
  181. )
  182. def serialize_json(
  183. self,
  184. value: Any,
  185. *,
  186. include: IncEx | None = None,
  187. exclude: IncEx | None = None,
  188. by_alias: bool = True,
  189. exclude_unset: bool = False,
  190. exclude_defaults: bool = False,
  191. exclude_none: bool = False,
  192. ) -> bytes:
  193. # What calls this code passes a value that already called
  194. # self._type_adapter.validate_python(value)
  195. # This uses Pydantic's dump_json() which serializes directly to JSON
  196. # bytes in one pass (via Rust), avoiding the intermediate Python dict
  197. # step of dump_python(mode="json") + json.dumps().
  198. return self._type_adapter.dump_json(
  199. value,
  200. include=include,
  201. exclude=exclude,
  202. by_alias=by_alias,
  203. exclude_unset=exclude_unset,
  204. exclude_defaults=exclude_defaults,
  205. exclude_none=exclude_none,
  206. )
  207. def __hash__(self) -> int:
  208. # Each ModelField is unique for our purposes, to allow making a dict from
  209. # ModelField to its JSON Schema.
  210. return id(self)
  211. def _has_computed_fields(field: ModelField) -> bool:
  212. computed_fields = field._type_adapter.core_schema.get("schema", {}).get(
  213. "computed_fields", []
  214. )
  215. return len(computed_fields) > 0
  216. def get_schema_from_model_field(
  217. *,
  218. field: ModelField,
  219. model_name_map: ModelNameMap,
  220. field_mapping: dict[
  221. tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue
  222. ],
  223. separate_input_output_schemas: bool = True,
  224. ) -> dict[str, Any]:
  225. override_mode: Literal["validation"] | None = (
  226. None
  227. if (separate_input_output_schemas or _has_computed_fields(field))
  228. else "validation"
  229. )
  230. field_alias = (
  231. (field.validation_alias or field.alias)
  232. if field.mode == "validation"
  233. else (field.serialization_alias or field.alias)
  234. )
  235. # This expects that GenerateJsonSchema was already used to generate the definitions
  236. json_schema = field_mapping[(field, override_mode or field.mode)]
  237. if "$ref" not in json_schema:
  238. # TODO remove when deprecating Pydantic v1
  239. # Ref: https://github.com/pydantic/pydantic/blob/d61792cc42c80b13b23e3ffa74bc37ec7c77f7d1/pydantic/schema.py#L207
  240. json_schema["title"] = field.field_info.title or field_alias.title().replace(
  241. "_", " "
  242. )
  243. return json_schema
  244. def get_definitions(
  245. *,
  246. fields: Sequence[ModelField],
  247. model_name_map: ModelNameMap,
  248. separate_input_output_schemas: bool = True,
  249. ) -> tuple[
  250. dict[tuple[ModelField, Literal["validation", "serialization"]], JsonSchemaValue],
  251. dict[str, dict[str, Any]],
  252. ]:
  253. schema_generator = GenerateJsonSchema(ref_template=REF_TEMPLATE)
  254. validation_fields = [field for field in fields if field.mode == "validation"]
  255. serialization_fields = [field for field in fields if field.mode == "serialization"]
  256. flat_validation_models = get_flat_models_from_fields(
  257. validation_fields, known_models=set()
  258. )
  259. flat_serialization_models = get_flat_models_from_fields(
  260. serialization_fields, known_models=set()
  261. )
  262. flat_validation_model_fields = [
  263. ModelField(
  264. field_info=FieldInfo(annotation=model),
  265. name=model.__name__,
  266. mode="validation",
  267. )
  268. for model in flat_validation_models
  269. ]
  270. flat_serialization_model_fields = [
  271. ModelField(
  272. field_info=FieldInfo(annotation=model),
  273. name=model.__name__,
  274. mode="serialization",
  275. )
  276. for model in flat_serialization_models
  277. ]
  278. flat_model_fields = flat_validation_model_fields + flat_serialization_model_fields
  279. input_types = {f.field_info.annotation for f in fields}
  280. unique_flat_model_fields = {
  281. f for f in flat_model_fields if f.field_info.annotation not in input_types
  282. }
  283. inputs = [
  284. (
  285. field,
  286. (
  287. field.mode
  288. if (separate_input_output_schemas or _has_computed_fields(field))
  289. else "validation"
  290. ),
  291. field._type_adapter.core_schema,
  292. )
  293. for field in list(fields) + list(unique_flat_model_fields)
  294. ]
  295. field_mapping, definitions = schema_generator.generate_definitions(inputs=inputs)
  296. for item_def in cast(dict[str, dict[str, Any]], definitions).values():
  297. if "description" in item_def:
  298. item_description = cast(str, item_def["description"]).split("\f")[0]
  299. item_def["description"] = item_description
  300. # definitions: dict[DefsRef, dict[str, Any]]
  301. # but mypy complains about general str in other places that are not declared as
  302. # DefsRef, although DefsRef is just str:
  303. # DefsRef = NewType('DefsRef', str)
  304. # So, a cast to simplify the types here
  305. return field_mapping, cast(dict[str, dict[str, Any]], definitions)
  306. def is_scalar_field(field: ModelField) -> bool:
  307. from fastapi import params
  308. return shared.field_annotation_is_scalar(
  309. field.field_info.annotation
  310. ) and not isinstance(field.field_info, params.Body)
  311. def copy_field_info(*, field_info: FieldInfo, annotation: Any) -> FieldInfo:
  312. cls = type(field_info)
  313. merged_field_info = cls.from_annotation(annotation)
  314. new_field_info = copy(field_info)
  315. new_field_info.metadata = merged_field_info.metadata
  316. new_field_info.annotation = merged_field_info.annotation
  317. return new_field_info
  318. def serialize_sequence_value(*, field: ModelField, value: Any) -> Sequence[Any]:
  319. origin_type = get_origin(field.field_info.annotation) or field.field_info.annotation
  320. if origin_type is Union or origin_type is UnionType: # Handle optional sequences
  321. union_args = get_args(field.field_info.annotation)
  322. for union_arg in union_args:
  323. if union_arg is type(None):
  324. continue
  325. origin_type = get_origin(union_arg) or union_arg
  326. break
  327. assert issubclass(origin_type, shared.sequence_types) # type: ignore[arg-type]
  328. return shared.sequence_annotation_to_type[origin_type](value) # type: ignore[no-any-return,index]
  329. def get_missing_field_error(loc: tuple[int | str, ...]) -> dict[str, Any]:
  330. error = ValidationError.from_exception_data(
  331. "Field required", [{"type": "missing", "loc": loc, "input": {}}]
  332. ).errors(include_url=False)[0]
  333. error["input"] = None
  334. return error # type: ignore[return-value]
  335. def create_body_model(
  336. *, fields: Sequence[ModelField], model_name: str
  337. ) -> type[BaseModel]:
  338. field_params = {f.name: (f.field_info.annotation, f.field_info) for f in fields}
  339. BodyModel: type[BaseModel] = create_model(model_name, **field_params) # type: ignore[call-overload]
  340. return BodyModel
  341. def get_model_fields(model: type[BaseModel]) -> list[ModelField]:
  342. model_fields: list[ModelField] = []
  343. for name, field_info in model.model_fields.items():
  344. type_ = field_info.annotation
  345. if lenient_issubclass(type_, (BaseModel, dict)) or is_dataclass(type_):
  346. model_config = None
  347. else:
  348. model_config = model.model_config
  349. model_fields.append(
  350. ModelField(
  351. field_info=field_info,
  352. name=name,
  353. config=model_config,
  354. )
  355. )
  356. return model_fields
  357. @lru_cache
  358. def get_cached_model_fields(model: type[BaseModel]) -> list[ModelField]:
  359. return get_model_fields(model)
  360. # Duplicate of several schema functions from Pydantic v1 to make them compatible with
  361. # Pydantic v2 and allow mixing the models
  362. TypeModelOrEnum = type["BaseModel"] | type[Enum]
  363. TypeModelSet = set[TypeModelOrEnum]
  364. def normalize_name(name: str) -> str:
  365. return re.sub(r"[^a-zA-Z0-9.\-_]", "_", name)
  366. def get_model_name_map(unique_models: TypeModelSet) -> dict[TypeModelOrEnum, str]:
  367. name_model_map = {}
  368. for model in unique_models:
  369. model_name = normalize_name(model.__name__)
  370. name_model_map[model_name] = model
  371. return {v: k for k, v in name_model_map.items()}
  372. def get_flat_models_from_model(
  373. model: type["BaseModel"], known_models: TypeModelSet | None = None
  374. ) -> TypeModelSet:
  375. known_models = known_models or set()
  376. fields = get_model_fields(model)
  377. get_flat_models_from_fields(fields, known_models=known_models)
  378. return known_models
  379. def get_flat_models_from_annotation(
  380. annotation: Any, known_models: TypeModelSet
  381. ) -> TypeModelSet:
  382. origin = get_origin(annotation)
  383. if origin is not None:
  384. for arg in get_args(annotation):
  385. if lenient_issubclass(arg, (BaseModel, Enum)):
  386. if arg not in known_models:
  387. known_models.add(arg) # type: ignore[arg-type] # ty: ignore[unused-ignore-comment]
  388. if lenient_issubclass(arg, BaseModel):
  389. get_flat_models_from_model(arg, known_models=known_models)
  390. else:
  391. get_flat_models_from_annotation(arg, known_models=known_models)
  392. return known_models
  393. def get_flat_models_from_field(
  394. field: ModelField, known_models: TypeModelSet
  395. ) -> TypeModelSet:
  396. field_type = field.field_info.annotation
  397. if lenient_issubclass(field_type, BaseModel):
  398. if field_type in known_models:
  399. return known_models
  400. known_models.add(field_type)
  401. get_flat_models_from_model(field_type, known_models=known_models)
  402. elif lenient_issubclass(field_type, Enum):
  403. known_models.add(field_type)
  404. else:
  405. get_flat_models_from_annotation(field_type, known_models=known_models)
  406. return known_models
  407. def get_flat_models_from_fields(
  408. fields: Sequence[ModelField], known_models: TypeModelSet
  409. ) -> TypeModelSet:
  410. for field in fields:
  411. get_flat_models_from_field(field, known_models=known_models)
  412. return known_models
  413. def _regenerate_error_with_loc(
  414. *, errors: Sequence[Any], loc_prefix: tuple[str | int, ...]
  415. ) -> list[dict[str, Any]]:
  416. updated_loc_errors: list[Any] = [
  417. {**err, "loc": loc_prefix + err.get("loc", ())} for err in errors
  418. ]
  419. return updated_loc_errors