encoders.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347
  1. import dataclasses
  2. import datetime
  3. from collections import defaultdict, deque
  4. from collections.abc import Callable
  5. from decimal import Decimal
  6. from enum import Enum
  7. from ipaddress import (
  8. IPv4Address,
  9. IPv4Interface,
  10. IPv4Network,
  11. IPv6Address,
  12. IPv6Interface,
  13. IPv6Network,
  14. )
  15. from pathlib import Path, PurePath
  16. from re import Pattern
  17. from types import GeneratorType
  18. from typing import Annotated, Any
  19. from uuid import UUID
  20. from annotated_doc import Doc
  21. from fastapi.exceptions import PydanticV1NotSupportedError
  22. from fastapi.types import IncEx
  23. from pydantic import BaseModel
  24. from pydantic.color import Color # ty: ignore[deprecated]
  25. from pydantic.networks import AnyUrl, NameEmail
  26. from pydantic.types import SecretBytes, SecretStr
  27. from pydantic_core import PydanticUndefinedType
  28. from ._compat import (
  29. Url,
  30. is_pydantic_v1_model_instance,
  31. )
  32. # Taken from Pydantic v1 as is
  33. def isoformat(o: datetime.date | datetime.time) -> str:
  34. return o.isoformat()
  35. # Adapted from Pydantic v1
  36. # TODO: pv2 should this return strings instead?
  37. def decimal_encoder(dec_value: Decimal) -> int | float:
  38. """
  39. Encodes a Decimal as int if there's no exponent, otherwise float
  40. This is useful when we use ConstrainedDecimal to represent Numeric(x,0)
  41. where an integer (but not int typed) is used. Encoding this as a float
  42. results in failed round-tripping between encode and parse.
  43. Our Id type is a prime example of this.
  44. >>> decimal_encoder(Decimal("1.0"))
  45. 1.0
  46. >>> decimal_encoder(Decimal("1"))
  47. 1
  48. >>> decimal_encoder(Decimal("NaN"))
  49. nan
  50. """
  51. exponent = dec_value.as_tuple().exponent
  52. if isinstance(exponent, int) and exponent >= 0:
  53. return int(dec_value)
  54. else:
  55. return float(dec_value)
  56. ENCODERS_BY_TYPE: dict[type[Any], Callable[[Any], Any]] = {
  57. bytes: lambda o: o.decode(),
  58. Color: str, # ty: ignore[deprecated]
  59. datetime.date: isoformat,
  60. datetime.datetime: isoformat,
  61. datetime.time: isoformat,
  62. datetime.timedelta: lambda td: td.total_seconds(),
  63. Decimal: decimal_encoder,
  64. Enum: lambda o: o.value,
  65. frozenset: list,
  66. deque: list,
  67. GeneratorType: list,
  68. IPv4Address: str,
  69. IPv4Interface: str,
  70. IPv4Network: str,
  71. IPv6Address: str,
  72. IPv6Interface: str,
  73. IPv6Network: str,
  74. NameEmail: str,
  75. Path: str,
  76. Pattern: lambda o: o.pattern,
  77. SecretBytes: str,
  78. SecretStr: str,
  79. set: list,
  80. UUID: str,
  81. Url: str,
  82. AnyUrl: str,
  83. }
  84. def generate_encoders_by_class_tuples(
  85. type_encoder_map: dict[Any, Callable[[Any], Any]],
  86. ) -> dict[Callable[[Any], Any], tuple[Any, ...]]:
  87. encoders_by_class_tuples: dict[Callable[[Any], Any], tuple[Any, ...]] = defaultdict(
  88. tuple
  89. )
  90. for type_, encoder in type_encoder_map.items():
  91. encoders_by_class_tuples[encoder] += (type_,)
  92. return encoders_by_class_tuples
  93. encoders_by_class_tuples = generate_encoders_by_class_tuples(ENCODERS_BY_TYPE)
  94. def jsonable_encoder(
  95. obj: Annotated[
  96. Any,
  97. Doc(
  98. """
  99. The input object to convert to JSON.
  100. """
  101. ),
  102. ],
  103. include: Annotated[
  104. IncEx | None,
  105. Doc(
  106. """
  107. Pydantic's `include` parameter, passed to Pydantic models to set the
  108. fields to include.
  109. """
  110. ),
  111. ] = None,
  112. exclude: Annotated[
  113. IncEx | None,
  114. Doc(
  115. """
  116. Pydantic's `exclude` parameter, passed to Pydantic models to set the
  117. fields to exclude.
  118. """
  119. ),
  120. ] = None,
  121. by_alias: Annotated[
  122. bool,
  123. Doc(
  124. """
  125. Pydantic's `by_alias` parameter, passed to Pydantic models to define if
  126. the output should use the alias names (when provided) or the Python
  127. attribute names. In an API, if you set an alias, it's probably because you
  128. want to use it in the result, so you probably want to leave this set to
  129. `True`.
  130. """
  131. ),
  132. ] = True,
  133. exclude_unset: Annotated[
  134. bool,
  135. Doc(
  136. """
  137. Pydantic's `exclude_unset` parameter, passed to Pydantic models to define
  138. if it should exclude from the output the fields that were not explicitly
  139. set (and that only had their default values).
  140. """
  141. ),
  142. ] = False,
  143. exclude_defaults: Annotated[
  144. bool,
  145. Doc(
  146. """
  147. Pydantic's `exclude_defaults` parameter, passed to Pydantic models to define
  148. if it should exclude from the output the fields that had the same default
  149. value, even when they were explicitly set.
  150. """
  151. ),
  152. ] = False,
  153. exclude_none: Annotated[
  154. bool,
  155. Doc(
  156. """
  157. Pydantic's `exclude_none` parameter, passed to Pydantic models to define
  158. if it should exclude from the output any fields that have a `None` value.
  159. """
  160. ),
  161. ] = False,
  162. custom_encoder: Annotated[
  163. dict[Any, Callable[[Any], Any]] | None,
  164. Doc(
  165. """
  166. Pydantic's `custom_encoder` parameter, passed to Pydantic models to define
  167. a custom encoder.
  168. """
  169. ),
  170. ] = None,
  171. sqlalchemy_safe: Annotated[
  172. bool,
  173. Doc(
  174. """
  175. Exclude from the output any fields that start with the name `_sa`.
  176. This is mainly a hack for compatibility with SQLAlchemy objects, they
  177. store internal SQLAlchemy-specific state in attributes named with `_sa`,
  178. and those objects can't (and shouldn't be) serialized to JSON.
  179. """
  180. ),
  181. ] = True,
  182. ) -> Any:
  183. """
  184. Convert any object to something that can be encoded in JSON.
  185. This is used internally by FastAPI to make sure anything you return can be
  186. encoded as JSON before it is sent to the client.
  187. You can also use it yourself, for example to convert objects before saving them
  188. in a database that supports only JSON.
  189. Read more about it in the
  190. [FastAPI docs for JSON Compatible Encoder](https://fastapi.tiangolo.com/tutorial/encoder/).
  191. """
  192. custom_encoder = custom_encoder or {}
  193. if custom_encoder:
  194. if type(obj) in custom_encoder:
  195. return custom_encoder[type(obj)](obj)
  196. else:
  197. for encoder_type, encoder_instance in custom_encoder.items():
  198. if isinstance(obj, encoder_type):
  199. return encoder_instance(obj)
  200. if include is not None and not isinstance(include, (set, dict)):
  201. include = set(include) # type: ignore[assignment] # ty: ignore[unused-ignore-comment]
  202. if exclude is not None and not isinstance(exclude, (set, dict)):
  203. exclude = set(exclude) # type: ignore[assignment] # ty: ignore[unused-ignore-comment]
  204. if isinstance(obj, BaseModel):
  205. obj_dict = obj.model_dump(
  206. mode="json",
  207. include=include,
  208. exclude=exclude,
  209. by_alias=by_alias,
  210. exclude_unset=exclude_unset,
  211. exclude_none=exclude_none,
  212. exclude_defaults=exclude_defaults,
  213. )
  214. return jsonable_encoder(
  215. obj_dict,
  216. exclude_none=exclude_none,
  217. exclude_defaults=exclude_defaults,
  218. sqlalchemy_safe=sqlalchemy_safe,
  219. )
  220. if dataclasses.is_dataclass(obj):
  221. assert not isinstance(obj, type)
  222. obj_dict = dataclasses.asdict(obj)
  223. return jsonable_encoder(
  224. obj_dict,
  225. include=include,
  226. exclude=exclude,
  227. by_alias=by_alias,
  228. exclude_unset=exclude_unset,
  229. exclude_defaults=exclude_defaults,
  230. exclude_none=exclude_none,
  231. custom_encoder=custom_encoder,
  232. sqlalchemy_safe=sqlalchemy_safe,
  233. )
  234. if isinstance(obj, Enum):
  235. return obj.value
  236. if isinstance(obj, PurePath):
  237. return str(obj)
  238. if isinstance(obj, (str, int, float, type(None))):
  239. return obj
  240. if isinstance(obj, PydanticUndefinedType):
  241. return None
  242. if isinstance(obj, dict):
  243. encoded_dict = {}
  244. allowed_keys = set(obj.keys())
  245. if include is not None:
  246. allowed_keys &= set(include)
  247. if exclude is not None:
  248. allowed_keys -= set(exclude)
  249. for key, value in obj.items():
  250. if (
  251. (
  252. not sqlalchemy_safe
  253. or (not isinstance(key, str))
  254. or (not key.startswith("_sa"))
  255. )
  256. and (value is not None or not exclude_none)
  257. and key in allowed_keys
  258. ):
  259. encoded_key = jsonable_encoder(
  260. key,
  261. by_alias=by_alias,
  262. exclude_unset=exclude_unset,
  263. exclude_none=exclude_none,
  264. custom_encoder=custom_encoder,
  265. sqlalchemy_safe=sqlalchemy_safe,
  266. )
  267. encoded_value = jsonable_encoder(
  268. value,
  269. by_alias=by_alias,
  270. exclude_unset=exclude_unset,
  271. exclude_none=exclude_none,
  272. custom_encoder=custom_encoder,
  273. sqlalchemy_safe=sqlalchemy_safe,
  274. )
  275. encoded_dict[encoded_key] = encoded_value
  276. return encoded_dict
  277. if isinstance(obj, (list, set, frozenset, GeneratorType, tuple, deque)):
  278. encoded_list = []
  279. for item in obj:
  280. encoded_list.append(
  281. jsonable_encoder(
  282. item,
  283. include=include,
  284. exclude=exclude,
  285. by_alias=by_alias,
  286. exclude_unset=exclude_unset,
  287. exclude_defaults=exclude_defaults,
  288. exclude_none=exclude_none,
  289. custom_encoder=custom_encoder,
  290. sqlalchemy_safe=sqlalchemy_safe,
  291. )
  292. )
  293. return encoded_list
  294. if type(obj) in ENCODERS_BY_TYPE:
  295. return ENCODERS_BY_TYPE[type(obj)](obj)
  296. for encoder, classes_tuple in encoders_by_class_tuples.items():
  297. if isinstance(obj, classes_tuple):
  298. return encoder(obj)
  299. if is_pydantic_v1_model_instance(obj):
  300. raise PydanticV1NotSupportedError(
  301. "pydantic.v1 models are no longer supported by FastAPI."
  302. f" Please update the model {obj!r}."
  303. )
  304. try:
  305. data = dict(obj)
  306. except Exception as e:
  307. errors: list[Exception] = []
  308. errors.append(e)
  309. try:
  310. data = vars(obj)
  311. except Exception as e:
  312. errors.append(e)
  313. raise ValueError(errors) from e
  314. return jsonable_encoder(
  315. data,
  316. include=include,
  317. exclude=exclude,
  318. by_alias=by_alias,
  319. exclude_unset=exclude_unset,
  320. exclude_defaults=exclude_defaults,
  321. exclude_none=exclude_none,
  322. custom_encoder=custom_encoder,
  323. sqlalchemy_safe=sqlalchemy_safe,
  324. )