2017-11-01 08:08:28 +00:00
|
|
|
"""JSON utility functions."""
|
2021-03-17 20:46:07 +00:00
|
|
|
from __future__ import annotations
|
|
|
|
|
2021-09-29 14:32:11 +00:00
|
|
|
from collections.abc import Callable
|
2017-11-01 08:08:28 +00:00
|
|
|
import json
|
2019-12-09 15:42:10 +00:00
|
|
|
import logging
|
2023-02-22 09:09:28 +00:00
|
|
|
from os import PathLike
|
2021-09-29 14:32:11 +00:00
|
|
|
from typing import Any
|
2017-11-01 08:08:28 +00:00
|
|
|
|
2022-06-22 19:59:51 +00:00
|
|
|
import orjson
|
|
|
|
|
2017-11-01 08:08:28 +00:00
|
|
|
from homeassistant.exceptions import HomeAssistantError
|
|
|
|
|
2023-02-16 10:37:57 +00:00
|
|
|
from .file import WriteError # pylint: disable=unused-import # noqa: F401
|
2021-11-11 06:19:56 +00:00
|
|
|
|
2023-02-22 09:09:28 +00:00
|
|
|
_SENTINEL = object()
|
2017-11-01 08:08:28 +00:00
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
|
2023-02-16 10:37:57 +00:00
|
|
|
JsonValueType = (
|
|
|
|
dict[str, "JsonValueType"] | list["JsonValueType"] | str | int | float | bool | None
|
|
|
|
)
|
|
|
|
"""Any data that can be returned by the standard JSON deserializing process."""
|
2023-02-16 18:40:47 +00:00
|
|
|
JsonArrayType = list[JsonValueType]
|
|
|
|
"""List that can be returned by the standard JSON deserializing process."""
|
2023-02-16 10:37:57 +00:00
|
|
|
JsonObjectType = dict[str, JsonValueType]
|
|
|
|
"""Dictionary that can be returned by the standard JSON deserializing process."""
|
|
|
|
|
|
|
|
JSON_ENCODE_EXCEPTIONS = (TypeError, ValueError)
|
|
|
|
JSON_DECODE_EXCEPTIONS = (orjson.JSONDecodeError,)
|
|
|
|
|
2018-02-18 21:11:24 +00:00
|
|
|
|
2018-06-25 16:53:49 +00:00
|
|
|
class SerializationError(HomeAssistantError):
|
|
|
|
"""Error serializing the data to JSON."""
|
|
|
|
|
|
|
|
|
2023-02-16 10:37:57 +00:00
|
|
|
json_loads: Callable[[bytes | bytearray | memoryview | str], JsonValueType]
|
|
|
|
json_loads = orjson.loads
|
|
|
|
"""Parse JSON data."""
|
|
|
|
|
|
|
|
|
2023-02-16 18:40:47 +00:00
|
|
|
def json_loads_array(__obj: bytes | bytearray | memoryview | str) -> JsonArrayType:
|
|
|
|
"""Parse JSON data and ensure result is a list."""
|
|
|
|
value: JsonValueType = json_loads(__obj)
|
|
|
|
# Avoid isinstance overhead as we are not interested in list subclasses
|
|
|
|
if type(value) is list: # pylint: disable=unidiomatic-typecheck
|
|
|
|
return value
|
|
|
|
raise ValueError(f"Expected JSON to be parsed as a list got {type(value)}")
|
|
|
|
|
|
|
|
|
2023-02-16 10:37:57 +00:00
|
|
|
def json_loads_object(__obj: bytes | bytearray | memoryview | str) -> JsonObjectType:
|
|
|
|
"""Parse JSON data and ensure result is a dictionary."""
|
|
|
|
value: JsonValueType = json_loads(__obj)
|
|
|
|
# Avoid isinstance overhead as we are not interested in dict subclasses
|
|
|
|
if type(value) is dict: # pylint: disable=unidiomatic-typecheck
|
|
|
|
return value
|
|
|
|
raise ValueError(f"Expected JSON to be parsed as a dict got {type(value)}")
|
|
|
|
|
|
|
|
|
2023-02-22 09:09:28 +00:00
|
|
|
def load_json(
|
|
|
|
filename: str | PathLike, default: JsonValueType = _SENTINEL # type: ignore[assignment]
|
|
|
|
) -> JsonValueType:
|
|
|
|
"""Load JSON data from a file.
|
2017-11-01 08:08:28 +00:00
|
|
|
|
|
|
|
Defaults to returning empty dict if file is not found.
|
|
|
|
"""
|
|
|
|
try:
|
2019-07-31 19:25:30 +00:00
|
|
|
with open(filename, encoding="utf-8") as fdesc:
|
2022-06-22 19:59:51 +00:00
|
|
|
return orjson.loads(fdesc.read()) # type: ignore[no-any-return]
|
2017-11-01 08:08:28 +00:00
|
|
|
except FileNotFoundError:
|
|
|
|
# This is not a fatal error
|
2019-07-31 19:25:30 +00:00
|
|
|
_LOGGER.debug("JSON file not found: %s", filename)
|
2017-11-01 08:08:28 +00:00
|
|
|
except ValueError as error:
|
2019-07-31 19:25:30 +00:00
|
|
|
_LOGGER.exception("Could not parse JSON content: %s", filename)
|
2020-08-28 11:50:32 +00:00
|
|
|
raise HomeAssistantError(error) from error
|
2017-11-01 08:08:28 +00:00
|
|
|
except OSError as error:
|
2019-07-31 19:25:30 +00:00
|
|
|
_LOGGER.exception("JSON file reading failed: %s", filename)
|
2020-08-28 11:50:32 +00:00
|
|
|
raise HomeAssistantError(error) from error
|
2023-02-22 09:09:28 +00:00
|
|
|
return {} if default is _SENTINEL else default
|
|
|
|
|
|
|
|
|
|
|
|
def load_json_array(
|
|
|
|
filename: str | PathLike, default: JsonArrayType = _SENTINEL # type: ignore[assignment]
|
|
|
|
) -> JsonArrayType:
|
|
|
|
"""Load JSON data from a file and return as list.
|
|
|
|
|
|
|
|
Defaults to returning empty list if file is not found.
|
|
|
|
"""
|
|
|
|
if default is _SENTINEL:
|
|
|
|
default = []
|
|
|
|
value: JsonValueType = load_json(filename, default=default)
|
|
|
|
# Avoid isinstance overhead as we are not interested in list subclasses
|
|
|
|
if type(value) is list: # pylint: disable=unidiomatic-typecheck
|
|
|
|
return value
|
|
|
|
_LOGGER.exception(
|
|
|
|
"Expected JSON to be parsed as a list got %s in: %s", {type(value)}, filename
|
|
|
|
)
|
|
|
|
raise HomeAssistantError(f"Expected JSON to be parsed as a list got {type(value)}")
|
|
|
|
|
|
|
|
|
|
|
|
def load_json_object(
|
|
|
|
filename: str | PathLike, default: JsonObjectType = _SENTINEL # type: ignore[assignment]
|
|
|
|
) -> JsonObjectType:
|
|
|
|
"""Load JSON data from a file and return as dict.
|
|
|
|
|
|
|
|
Defaults to returning empty dict if file is not found.
|
|
|
|
"""
|
|
|
|
if default is _SENTINEL:
|
|
|
|
default = {}
|
|
|
|
value: JsonValueType = load_json(filename, default=default)
|
|
|
|
# Avoid isinstance overhead as we are not interested in dict subclasses
|
|
|
|
if type(value) is dict: # pylint: disable=unidiomatic-typecheck
|
|
|
|
return value
|
|
|
|
_LOGGER.exception(
|
|
|
|
"Expected JSON to be parsed as a dict got %s in: %s", {type(value)}, filename
|
|
|
|
)
|
|
|
|
raise HomeAssistantError(f"Expected JSON to be parsed as a dict got {type(value)}")
|
2017-11-01 08:08:28 +00:00
|
|
|
|
|
|
|
|
2019-07-31 19:25:30 +00:00
|
|
|
def save_json(
|
|
|
|
filename: str,
|
2021-03-17 20:46:07 +00:00
|
|
|
data: list | dict,
|
2019-07-31 19:25:30 +00:00
|
|
|
private: bool = False,
|
|
|
|
*,
|
2021-03-17 20:46:07 +00:00
|
|
|
encoder: type[json.JSONEncoder] | None = None,
|
2021-11-15 10:19:31 +00:00
|
|
|
atomic_writes: bool = False,
|
2019-07-31 19:25:30 +00:00
|
|
|
) -> None:
|
2023-02-16 10:37:57 +00:00
|
|
|
"""Save JSON data to a file."""
|
|
|
|
# pylint: disable-next=import-outside-toplevel
|
|
|
|
from homeassistant.helpers.frame import report
|
2017-11-01 08:08:28 +00:00
|
|
|
|
2023-02-16 10:37:57 +00:00
|
|
|
report(
|
|
|
|
(
|
|
|
|
"uses save_json from homeassistant.util.json module."
|
|
|
|
" This is deprecated and will stop working in Home Assistant 2022.4, it"
|
|
|
|
" should be updated to use homeassistant.helpers.json module instead"
|
|
|
|
),
|
|
|
|
error_if_core=False,
|
|
|
|
)
|
|
|
|
|
|
|
|
# pylint: disable-next=import-outside-toplevel
|
|
|
|
import homeassistant.helpers.json as json_helper
|
|
|
|
|
|
|
|
json_helper.save_json(
|
|
|
|
filename, data, private, encoder=encoder, atomic_writes=atomic_writes
|
|
|
|
)
|
2020-02-17 18:49:42 +00:00
|
|
|
|
|
|
|
|
2021-03-17 20:46:07 +00:00
|
|
|
def format_unserializable_data(data: dict[str, Any]) -> str:
|
2020-04-14 06:46:41 +00:00
|
|
|
"""Format output of find_paths in a friendly way.
|
|
|
|
|
|
|
|
Format is comma separated: <path>=<value>(<type>)
|
|
|
|
"""
|
|
|
|
return ", ".join(f"{path}={value}({type(value)}" for path, value in data.items())
|
|
|
|
|
|
|
|
|
|
|
|
def find_paths_unserializable_data(
|
|
|
|
bad_data: Any, *, dump: Callable[[Any], str] = json.dumps
|
2021-03-17 20:46:07 +00:00
|
|
|
) -> dict[str, Any]:
|
2020-02-17 18:49:42 +00:00
|
|
|
"""Find the paths to unserializable data.
|
|
|
|
|
|
|
|
This method is slow! Only use for error handling.
|
|
|
|
"""
|
2023-02-16 10:37:57 +00:00
|
|
|
# pylint: disable-next=import-outside-toplevel
|
|
|
|
from homeassistant.helpers.frame import report
|
|
|
|
|
|
|
|
report(
|
|
|
|
(
|
|
|
|
"uses find_paths_unserializable_data from homeassistant.util.json module."
|
|
|
|
" This is deprecated and will stop working in Home Assistant 2022.4, it"
|
|
|
|
" should be updated to use homeassistant.helpers.json module instead"
|
|
|
|
),
|
|
|
|
error_if_core=False,
|
|
|
|
)
|
|
|
|
|
|
|
|
# pylint: disable-next=import-outside-toplevel
|
|
|
|
import homeassistant.helpers.json as json_helper
|
|
|
|
|
|
|
|
return json_helper.find_paths_unserializable_data(bad_data, dump=dump)
|