Skip to content
Permalink
 
 
Cannot retrieve contributors at this time
"""Helpers to help with encoding Home Assistant objects in JSON."""
import datetime
import json
from pathlib import Path
from typing import Any, Final
import orjson
JSON_ENCODE_EXCEPTIONS = (TypeError, ValueError)
JSON_DECODE_EXCEPTIONS = (orjson.JSONDecodeError,)
class JSONEncoder(json.JSONEncoder):
"""JSONEncoder that supports Home Assistant objects."""
def default(self, o: Any) -> Any:
"""Convert Home Assistant objects.
Hand other objects to the original method.
"""
if isinstance(o, datetime.datetime):
return o.isoformat()
if isinstance(o, set):
return list(o)
if hasattr(o, "as_dict"):
return o.as_dict()
return json.JSONEncoder.default(self, o)
def json_encoder_default(obj: Any) -> Any:
"""Convert Home Assistant objects.
Hand other objects to the original method.
"""
if isinstance(obj, (set, tuple)):
return list(obj)
if isinstance(obj, float):
return float(obj)
if hasattr(obj, "as_dict"):
return obj.as_dict()
if isinstance(obj, Path):
return obj.as_posix()
raise TypeError
class ExtendedJSONEncoder(JSONEncoder):
"""JSONEncoder that supports Home Assistant objects and falls back to repr(o)."""
def default(self, o: Any) -> Any:
"""Convert certain objects.
Fall back to repr(o).
"""
if isinstance(o, datetime.timedelta):
return {"__type": str(type(o)), "total_seconds": o.total_seconds()}
if isinstance(o, datetime.datetime):
return o.isoformat()
if isinstance(o, (datetime.date, datetime.time)):
return {"__type": str(type(o)), "isoformat": o.isoformat()}
try:
return super().default(o)
except TypeError:
return {"__type": str(type(o)), "repr": repr(o)}
def json_bytes(data: Any) -> bytes:
"""Dump json bytes."""
return orjson.dumps(
data, option=orjson.OPT_NON_STR_KEYS, default=json_encoder_default
)
def json_dumps(data: Any) -> str:
"""Dump json string.
orjson supports serializing dataclasses natively which
eliminates the need to implement as_dict in many places
when the data is already in a dataclass. This works
well as long as all the data in the dataclass can also
be serialized.
If it turns out to be a problem we can disable this
with option |= orjson.OPT_PASSTHROUGH_DATACLASS and it
will fallback to as_dict
"""
return orjson.dumps(
data, option=orjson.OPT_NON_STR_KEYS, default=json_encoder_default
).decode("utf-8")
def json_dumps_sorted(data: Any) -> str:
"""Dump json string with keys sorted."""
return orjson.dumps(
data,
option=orjson.OPT_NON_STR_KEYS | orjson.OPT_SORT_KEYS,
default=json_encoder_default,
).decode("utf-8")
json_loads = orjson.loads
JSON_DUMP: Final = json_dumps