(
obj: Any, custom_encoder: Optional[Dict[Any, Callable[[Any], Any]]] = None
)
| 30 | |
| 31 | |
| 32 | def jsonable_encoder( |
| 33 | obj: Any, custom_encoder: Optional[Dict[Any, Callable[[Any], Any]]] = None |
| 34 | ) -> Any: |
| 35 | custom_encoder = custom_encoder or {} |
| 36 | if custom_encoder: |
| 37 | if type(obj) in custom_encoder: |
| 38 | return custom_encoder[type(obj)](obj) |
| 39 | else: |
| 40 | for encoder_type, encoder_instance in custom_encoder.items(): |
| 41 | if isinstance(obj, encoder_type): |
| 42 | return encoder_instance(obj) |
| 43 | if isinstance(obj, pydantic.BaseModel): |
| 44 | if IS_PYDANTIC_V2: |
| 45 | encoder = getattr(obj.model_config, "json_encoders", {}) # type: ignore # Pydantic v2 |
| 46 | else: |
| 47 | encoder = getattr(obj.__config__, "json_encoders", {}) # type: ignore # Pydantic v1 |
| 48 | if custom_encoder: |
| 49 | encoder.update(custom_encoder) |
| 50 | obj_dict = obj.dict(by_alias=True) |
| 51 | if "__root__" in obj_dict: |
| 52 | obj_dict = obj_dict["__root__"] |
| 53 | if "root" in obj_dict: |
| 54 | obj_dict = obj_dict["root"] |
| 55 | return jsonable_encoder(obj_dict, custom_encoder=encoder) |
| 56 | if dataclasses.is_dataclass(obj): |
| 57 | obj_dict = dataclasses.asdict(obj) # type: ignore |
| 58 | return jsonable_encoder(obj_dict, custom_encoder=custom_encoder) |
| 59 | if isinstance(obj, bytes): |
| 60 | return base64.b64encode(obj).decode("utf-8") |
| 61 | if isinstance(obj, Enum): |
| 62 | return obj.value |
| 63 | if isinstance(obj, PurePath): |
| 64 | return str(obj) |
| 65 | if isinstance(obj, (str, int, float, type(None))): |
| 66 | return obj |
| 67 | if isinstance(obj, dt.datetime): |
| 68 | return serialize_datetime(obj) |
| 69 | if isinstance(obj, dt.date): |
| 70 | return str(obj) |
| 71 | if isinstance(obj, dict): |
| 72 | encoded_dict = {} |
| 73 | allowed_keys = set(obj.keys()) |
| 74 | for key, value in obj.items(): |
| 75 | if key in allowed_keys: |
| 76 | encoded_key = jsonable_encoder(key, custom_encoder=custom_encoder) |
| 77 | encoded_value = jsonable_encoder(value, custom_encoder=custom_encoder) |
| 78 | encoded_dict[encoded_key] = encoded_value |
| 79 | return encoded_dict |
| 80 | if isinstance(obj, (list, set, frozenset, GeneratorType, tuple)): |
| 81 | encoded_list = [] |
| 82 | for item in obj: |
| 83 | encoded_list.append(jsonable_encoder(item, custom_encoder=custom_encoder)) |
| 84 | return encoded_list |
| 85 | |
| 86 | def fallback_serializer(o: Any) -> Any: |
| 87 | attempt_encode = encode_by_type(o) |
| 88 | if attempt_encode is not None: |
| 89 | return attempt_encode |
no test coverage detected