Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/deepgram/core/client_wrapper.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ def get_headers(self) -> typing.Dict[str, str]:
"X-Fern-Language": "Python",
"X-Fern-SDK-Name": "deepgram",
# x-release-please-start-version
"X-Fern-SDK-Version": "5.2.0",
"X-Fern-SDK-Version": "5.2.0",
# x-release-please-end
**(self.get_custom_headers() or {}),
}
Expand Down
75 changes: 42 additions & 33 deletions src/deepgram/core/jsonable_encoder.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,13 +30,26 @@

def jsonable_encoder(obj: Any, custom_encoder: Optional[Dict[Any, Callable[[Any], Any]]] = None) -> Any:
custom_encoder = custom_encoder or {}
# Short-circuit most common types (early returns improve performance)
if isinstance(obj, (str, int, float, type(None))):
return obj
if isinstance(obj, bytes):
return base64.b64encode(obj).decode("utf-8")
if isinstance(obj, Enum):
return obj.value
if isinstance(obj, PurePath):
return str(obj)
if isinstance(obj, dt.datetime):
return serialize_datetime(obj)
if isinstance(obj, dt.date):
return str(obj)
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)
obj_type = type(obj)
if obj_type in custom_encoder:
return custom_encoder[obj_type](obj)
for encoder_type, encoder_instance in custom_encoder.items():
if isinstance(obj, encoder_type):
return encoder_instance(obj)
if isinstance(obj, pydantic.BaseModel):
if IS_PYDANTIC_V2:
encoder = getattr(obj.model_config, "json_encoders", {}) # type: ignore # Pydantic v2
Expand All @@ -45,6 +58,7 @@ def jsonable_encoder(obj: Any, custom_encoder: Optional[Dict[Any, Callable[[Any]
if custom_encoder:
encoder.update(custom_encoder)
obj_dict = obj.dict(by_alias=True)
# Collapse single-root models quickly
if "__root__" in obj_dict:
obj_dict = obj_dict["__root__"]
if "root" in obj_dict:
Expand All @@ -53,32 +67,27 @@ def jsonable_encoder(obj: Any, custom_encoder: Optional[Dict[Any, Callable[[Any]
if dataclasses.is_dataclass(obj):
obj_dict = dataclasses.asdict(obj) # type: ignore
return jsonable_encoder(obj_dict, custom_encoder=custom_encoder)
if isinstance(obj, bytes):
return base64.b64encode(obj).decode("utf-8")
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, dt.datetime):
return serialize_datetime(obj)
if isinstance(obj, dt.date):
return str(obj)
if isinstance(obj, dict):
encoded_dict = {}
allowed_keys = set(obj.keys())
for key, value in obj.items():
if key in allowed_keys:
encoded_key = jsonable_encoder(key, custom_encoder=custom_encoder)
encoded_value = jsonable_encoder(value, custom_encoder=custom_encoder)
encoded_dict[encoded_key] = encoded_value
return encoded_dict
# More efficient: use comprehension to avoid explicit allowed_keys/set logic
return {
jsonable_encoder(k, custom_encoder=custom_encoder): jsonable_encoder(v, custom_encoder=custom_encoder)
for k, v in obj.items()
}
if isinstance(obj, (list, set, frozenset, GeneratorType, tuple)):
encoded_list = []
for item in obj:
encoded_list.append(jsonable_encoder(item, custom_encoder=custom_encoder))
return encoded_list
# Use generator expression with appropriate constructor, faster than append-hit loop
typ = type(obj)
# Maintain original type for tuple/list; sets frozenset supported
if typ is tuple:
return tuple(jsonable_encoder(item, custom_encoder=custom_encoder) for item in obj)
elif typ is list:
return [jsonable_encoder(item, custom_encoder=custom_encoder) for item in obj]
elif typ is set:
return {jsonable_encoder(item, custom_encoder=custom_encoder) for item in obj}
elif typ is frozenset:
return frozenset(jsonable_encoder(item, custom_encoder=custom_encoder) for item in obj)
else:
# for generators
return [jsonable_encoder(item, custom_encoder=custom_encoder) for item in obj]

def fallback_serializer(o: Any) -> Any:
attempt_encode = encode_by_type(o)
Expand All @@ -92,9 +101,9 @@ def fallback_serializer(o: Any) -> Any:
errors.append(e)
try:
data = vars(o)
except Exception as e:
errors.append(e)
raise ValueError(errors) from e
except Exception as e2:
errors.append(e2)
raise ValueError(errors) from e2
return jsonable_encoder(data, custom_encoder=custom_encoder)

return to_jsonable_with_fallback(obj, fallback_serializer)