Spaces:
Runtime error
Runtime error
| import dataclasses | |
| import datetime | |
| from collections import defaultdict, deque | |
| from decimal import Decimal | |
| from enum import Enum | |
| from ipaddress import ( | |
| IPv4Address, | |
| IPv4Interface, | |
| IPv4Network, | |
| IPv6Address, | |
| IPv6Interface, | |
| IPv6Network, | |
| ) | |
| from pathlib import Path, PurePath | |
| from re import Pattern | |
| from types import GeneratorType | |
| from typing import Any, Callable, Dict, List, Optional, Tuple, Type, Union | |
| from uuid import UUID | |
| from fastapi.types import IncEx | |
| from pydantic import BaseModel | |
| from pydantic.color import Color | |
| from pydantic.networks import AnyUrl, NameEmail | |
| from pydantic.types import SecretBytes, SecretStr | |
| from ._compat import PYDANTIC_V2, Url, _model_dump | |
| # Taken from Pydantic v1 as is | |
| def isoformat(o: Union[datetime.date, datetime.time]) -> str: | |
| return o.isoformat() | |
| # Taken from Pydantic v1 as is | |
| # TODO: pv2 should this return strings instead? | |
| def decimal_encoder(dec_value: Decimal) -> Union[int, float]: | |
| """ | |
| Encodes a Decimal as int of there's no exponent, otherwise float | |
| This is useful when we use ConstrainedDecimal to represent Numeric(x,0) | |
| where a integer (but not int typed) is used. Encoding this as a float | |
| results in failed round-tripping between encode and parse. | |
| Our Id type is a prime example of this. | |
| >>> decimal_encoder(Decimal("1.0")) | |
| 1.0 | |
| >>> decimal_encoder(Decimal("1")) | |
| 1 | |
| """ | |
| if dec_value.as_tuple().exponent >= 0: # type: ignore[operator] | |
| return int(dec_value) | |
| else: | |
| return float(dec_value) | |
| ENCODERS_BY_TYPE: Dict[Type[Any], Callable[[Any], Any]] = { | |
| bytes: lambda o: o.decode(), | |
| Color: str, | |
| datetime.date: isoformat, | |
| datetime.datetime: isoformat, | |
| datetime.time: isoformat, | |
| datetime.timedelta: lambda td: td.total_seconds(), | |
| Decimal: decimal_encoder, | |
| Enum: lambda o: o.value, | |
| frozenset: list, | |
| deque: list, | |
| GeneratorType: list, | |
| IPv4Address: str, | |
| IPv4Interface: str, | |
| IPv4Network: str, | |
| IPv6Address: str, | |
| IPv6Interface: str, | |
| IPv6Network: str, | |
| NameEmail: str, | |
| Path: str, | |
| Pattern: lambda o: o.pattern, | |
| SecretBytes: str, | |
| SecretStr: str, | |
| set: list, | |
| UUID: str, | |
| Url: str, | |
| AnyUrl: str, | |
| } | |
| def generate_encoders_by_class_tuples( | |
| type_encoder_map: Dict[Any, Callable[[Any], Any]] | |
| ) -> Dict[Callable[[Any], Any], Tuple[Any, ...]]: | |
| encoders_by_class_tuples: Dict[Callable[[Any], Any], Tuple[Any, ...]] = defaultdict( | |
| tuple | |
| ) | |
| for type_, encoder in type_encoder_map.items(): | |
| encoders_by_class_tuples[encoder] += (type_,) | |
| return encoders_by_class_tuples | |
| encoders_by_class_tuples = generate_encoders_by_class_tuples(ENCODERS_BY_TYPE) | |
| def jsonable_encoder( | |
| obj: Any, | |
| include: Optional[IncEx] = None, | |
| exclude: Optional[IncEx] = None, | |
| by_alias: bool = True, | |
| exclude_unset: bool = False, | |
| exclude_defaults: bool = False, | |
| exclude_none: bool = False, | |
| custom_encoder: Optional[Dict[Any, Callable[[Any], Any]]] = None, | |
| sqlalchemy_safe: bool = True, | |
| ) -> Any: | |
| custom_encoder = custom_encoder or {} | |
| if custom_encoder: | |
| if type(obj) in custom_encoder: | |
| return custom_encoder[type(obj)](obj) | |
| else: | |
| for encoder_type, encoder_instance in custom_encoder.items(): | |
| if isinstance(obj, encoder_type): | |
| return encoder_instance(obj) | |
| if include is not None and not isinstance(include, (set, dict)): | |
| include = set(include) | |
| if exclude is not None and not isinstance(exclude, (set, dict)): | |
| exclude = set(exclude) | |
| if isinstance(obj, BaseModel): | |
| # TODO: remove when deprecating Pydantic v1 | |
| encoders: Dict[Any, Any] = {} | |
| if not PYDANTIC_V2: | |
| encoders = getattr(obj.__config__, "json_encoders", {}) # type: ignore[attr-defined] | |
| if custom_encoder: | |
| encoders.update(custom_encoder) | |
| obj_dict = _model_dump( | |
| obj, | |
| mode="json", | |
| include=include, | |
| exclude=exclude, | |
| by_alias=by_alias, | |
| exclude_unset=exclude_unset, | |
| exclude_none=exclude_none, | |
| exclude_defaults=exclude_defaults, | |
| ) | |
| if "__root__" in obj_dict: | |
| obj_dict = obj_dict["__root__"] | |
| return jsonable_encoder( | |
| obj_dict, | |
| exclude_none=exclude_none, | |
| exclude_defaults=exclude_defaults, | |
| # TODO: remove when deprecating Pydantic v1 | |
| custom_encoder=encoders, | |
| sqlalchemy_safe=sqlalchemy_safe, | |
| ) | |
| if dataclasses.is_dataclass(obj): | |
| obj_dict = dataclasses.asdict(obj) | |
| return jsonable_encoder( | |
| obj_dict, | |
| include=include, | |
| exclude=exclude, | |
| by_alias=by_alias, | |
| exclude_unset=exclude_unset, | |
| exclude_defaults=exclude_defaults, | |
| exclude_none=exclude_none, | |
| custom_encoder=custom_encoder, | |
| sqlalchemy_safe=sqlalchemy_safe, | |
| ) | |
| if isinstance(obj, Enum): | |
| return obj.value | |
| if isinstance(obj, PurePath): | |
| return str(obj) | |
| if isinstance(obj, (str, int, float, type(None))): | |
| return obj | |
| if isinstance(obj, dict): | |
| encoded_dict = {} | |
| allowed_keys = set(obj.keys()) | |
| if include is not None: | |
| allowed_keys &= set(include) | |
| if exclude is not None: | |
| allowed_keys -= set(exclude) | |
| 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 not exclude_none) | |
| and key in allowed_keys | |
| ): | |
| encoded_key = jsonable_encoder( | |
| key, | |
| by_alias=by_alias, | |
| exclude_unset=exclude_unset, | |
| exclude_none=exclude_none, | |
| custom_encoder=custom_encoder, | |
| sqlalchemy_safe=sqlalchemy_safe, | |
| ) | |
| encoded_value = jsonable_encoder( | |
| value, | |
| by_alias=by_alias, | |
| exclude_unset=exclude_unset, | |
| exclude_none=exclude_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, deque)): | |
| encoded_list = [] | |
| for item in obj: | |
| encoded_list.append( | |
| jsonable_encoder( | |
| item, | |
| include=include, | |
| exclude=exclude, | |
| by_alias=by_alias, | |
| exclude_unset=exclude_unset, | |
| exclude_defaults=exclude_defaults, | |
| exclude_none=exclude_none, | |
| custom_encoder=custom_encoder, | |
| sqlalchemy_safe=sqlalchemy_safe, | |
| ) | |
| ) | |
| return encoded_list | |
| if type(obj) in ENCODERS_BY_TYPE: | |
| return ENCODERS_BY_TYPE[type(obj)](obj) | |
| for encoder, classes_tuple in encoders_by_class_tuples.items(): | |
| if isinstance(obj, classes_tuple): | |
| return encoder(obj) | |
| try: | |
| data = dict(obj) | |
| except Exception as e: | |
| errors: List[Exception] = [] | |
| errors.append(e) | |
| try: | |
| data = vars(obj) | |
| except Exception as e: | |
| errors.append(e) | |
| raise ValueError(errors) from e | |
| return jsonable_encoder( | |
| data, | |
| include=include, | |
| exclude=exclude, | |
| by_alias=by_alias, | |
| exclude_unset=exclude_unset, | |
| exclude_defaults=exclude_defaults, | |
| exclude_none=exclude_none, | |
| custom_encoder=custom_encoder, | |
| sqlalchemy_safe=sqlalchemy_safe, | |
| ) | |