mirror of
https://github.com/rjNemo/fastapi
synced 2026-06-10 04:26:43 +00:00
ab2b86fe2c
* Make compatible with pydantic v1 * Remove unused import * Remove unused ignores * Update pydantic version * Fix minor formatting issue * ⏪ Revert removing iterate_in_threadpool * ✨ Add backwards compatibility with Pydantic 0.32.2 with deprecation warnings * ✅ Update tests to not break when using Pydantic < 1.0.0 * 📝 Update docs for Pydantic version 1.0.0 * 📌 Update Pydantic range version to support from 0.32.2 * 🎨 Format test imports * ✨ Add support for Pydantic < 1.2 for populate_validators * ✨ Add backwards compatibility for Pydantic < 1.2.0 with required fields * 📌 Relax requirement for Pydantic to < 2.0.0 🎉 🚀 * 💚 Update pragma coverage for older Pydantic versions
131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
from enum import Enum
|
|
from types import GeneratorType
|
|
from typing import Any, Dict, List, Set, Union
|
|
|
|
from fastapi.utils import PYDANTIC_1, logger
|
|
from pydantic import BaseModel
|
|
from pydantic.json import ENCODERS_BY_TYPE
|
|
|
|
SetIntStr = Set[Union[int, str]]
|
|
DictIntStrAny = Dict[Union[int, str], Any]
|
|
|
|
|
|
def jsonable_encoder(
|
|
obj: Any,
|
|
include: Union[SetIntStr, DictIntStrAny] = None,
|
|
exclude: Union[SetIntStr, DictIntStrAny] = set(),
|
|
by_alias: bool = True,
|
|
skip_defaults: bool = None,
|
|
exclude_unset: bool = False,
|
|
include_none: bool = True,
|
|
custom_encoder: dict = {},
|
|
sqlalchemy_safe: bool = True,
|
|
) -> Any:
|
|
if skip_defaults is not None:
|
|
logger.warning( # pragma: nocover
|
|
"skip_defaults in jsonable_encoder has been deprecated in \
|
|
favor of exclude_unset to keep in line with Pydantic v1, support for it \
|
|
will be removed soon."
|
|
)
|
|
if include is not None and not isinstance(include, set):
|
|
include = set(include)
|
|
if exclude is not None and not isinstance(exclude, set):
|
|
exclude = set(exclude)
|
|
if isinstance(obj, BaseModel):
|
|
encoder = getattr(obj.Config, "json_encoders", custom_encoder)
|
|
if PYDANTIC_1:
|
|
obj_dict = obj.dict(
|
|
include=include,
|
|
exclude=exclude,
|
|
by_alias=by_alias,
|
|
exclude_unset=bool(exclude_unset or skip_defaults),
|
|
)
|
|
else: # pragma: nocover
|
|
obj_dict = obj.dict(
|
|
include=include,
|
|
exclude=exclude,
|
|
by_alias=by_alias,
|
|
skip_defaults=bool(exclude_unset or skip_defaults),
|
|
)
|
|
return jsonable_encoder(
|
|
obj_dict,
|
|
include_none=include_none,
|
|
custom_encoder=encoder,
|
|
sqlalchemy_safe=sqlalchemy_safe,
|
|
)
|
|
if isinstance(obj, Enum):
|
|
return obj.value
|
|
if isinstance(obj, (str, int, float, type(None))):
|
|
return obj
|
|
if isinstance(obj, dict):
|
|
encoded_dict = {}
|
|
for key, value in obj.items():
|
|
if (
|
|
(
|
|
not sqlalchemy_safe
|
|
or (not isinstance(key, str))
|
|
or (not key.startswith("_sa"))
|
|
)
|
|
and (value is not None or include_none)
|
|
and ((include and key in include) or key not in exclude)
|
|
):
|
|
encoded_key = jsonable_encoder(
|
|
key,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
include_none=include_none,
|
|
custom_encoder=custom_encoder,
|
|
sqlalchemy_safe=sqlalchemy_safe,
|
|
)
|
|
encoded_value = jsonable_encoder(
|
|
value,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
include_none=include_none,
|
|
custom_encoder=custom_encoder,
|
|
sqlalchemy_safe=sqlalchemy_safe,
|
|
)
|
|
encoded_dict[encoded_key] = encoded_value
|
|
return encoded_dict
|
|
if isinstance(obj, (list, set, frozenset, GeneratorType, tuple)):
|
|
encoded_list = []
|
|
for item in obj:
|
|
encoded_list.append(
|
|
jsonable_encoder(
|
|
item,
|
|
include=include,
|
|
exclude=exclude,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
include_none=include_none,
|
|
custom_encoder=custom_encoder,
|
|
sqlalchemy_safe=sqlalchemy_safe,
|
|
)
|
|
)
|
|
return encoded_list
|
|
errors: List[Exception] = []
|
|
try:
|
|
if custom_encoder and type(obj) in custom_encoder:
|
|
encoder = custom_encoder[type(obj)]
|
|
else:
|
|
encoder = ENCODERS_BY_TYPE[type(obj)]
|
|
return encoder(obj)
|
|
except KeyError as e:
|
|
errors.append(e)
|
|
try:
|
|
data = dict(obj)
|
|
except Exception as e:
|
|
errors.append(e)
|
|
try:
|
|
data = vars(obj)
|
|
except Exception as e:
|
|
errors.append(e)
|
|
raise ValueError(errors)
|
|
return jsonable_encoder(
|
|
data,
|
|
by_alias=by_alias,
|
|
exclude_unset=exclude_unset,
|
|
include_none=include_none,
|
|
custom_encoder=custom_encoder,
|
|
sqlalchemy_safe=sqlalchemy_safe,
|
|
)
|