Add typed helpers and improve type hints in util/json

homeassistant/components/ios/__init__.py

@@ -1,7 +1,6 @@
 """Native Home Assistant iOS app component."""
 import datetime
 from http import HTTPStatus
-from typing import TYPE_CHECKING
 
 import voluptuous as vol
 
@@ -14,7 +13,7 @@
 from homeassistant.helpers.dispatcher import async_dispatcher_send
 from homeassistant.helpers.json import save_json
 from homeassistant.helpers.typing import ConfigType
-from homeassistant.util.json import load_json
+from homeassistant.util.json import load_json_object
 
 from .const import (
     CONF_ACTION_BACKGROUND_COLOR,
@@ -252,22 +251,19 @@ def device_name_for_push_id(hass, push_id):
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Set up the iOS component."""
-    conf = config.get(DOMAIN)
+    conf: ConfigType | None = config.get(DOMAIN)
 
     ios_config = await hass.async_add_executor_job(
-        load_json, hass.config.path(CONFIGURATION_FILE)
+        load_json_object, hass.config.path(CONFIGURATION_FILE)
     )
 
-    if TYPE_CHECKING:
-        assert isinstance(ios_config, dict)
-
     if ios_config == {}:
         ios_config[ATTR_DEVICES] = {}
 
-    ios_config[CONF_USER] = conf or {}
+    if CONF_PUSH not in (conf_user := conf or {}):
+        conf_user[CONF_PUSH] = {}
 
-    if CONF_PUSH not in ios_config[CONF_USER]:
-        ios_config[CONF_USER][CONF_PUSH] = {}
+    ios_config[CONF_USER] = conf_user
 
     hass.data[DOMAIN] = ios_config
 

homeassistant/components/nest/config_flow.py

@@ -33,7 +33,7 @@
 from homeassistant.exceptions import HomeAssistantError
 from homeassistant.helpers import config_entry_oauth2_flow
 from homeassistant.util import get_random_string
-from homeassistant.util.json import load_json
+from homeassistant.util.json import JsonObjectType, load_json_object
 
 from . import api
 from .const import (
@@ -570,15 +570,15 @@ async def async_step_import(self, info: dict[str, Any]) -> FlowResult:
             return await self.async_step_link()
 
         flow = self.hass.data[DATA_FLOW_IMPL][DOMAIN]
-        tokens = await self.hass.async_add_executor_job(load_json, config_path)
+        tokens = await self.hass.async_add_executor_job(load_json_object, config_path)
 
         return self._entry_from_tokens(
             "Nest (import from configuration.yaml)", flow, tokens
         )
 
     @callback
     def _entry_from_tokens(
-        self, title: str, flow: dict[str, Any], tokens: list[Any] | dict[Any, Any]
+        self, title: str, flow: dict[str, Any], tokens: JsonObjectType
     ) -> FlowResult:
         """Create an entry from tokens."""
         return self.async_create_entry(

homeassistant/components/shopping_list/__init__.py

@@ -15,7 +15,7 @@
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.json import save_json
 from homeassistant.helpers.typing import ConfigType
-from homeassistant.util.json import load_json
+from homeassistant.util.json import JsonArrayType, load_json_array
 
 from .const import (
     DOMAIN,
@@ -174,10 +174,10 @@ class NoMatchingShoppingListItem(Exception):
 class ShoppingData:
     """Class to hold shopping list data."""
 
-    def __init__(self, hass):
+    def __init__(self, hass: HomeAssistant) -> None:
         """Initialize the shopping list."""
         self.hass = hass
-        self.items = []
+        self.items: JsonArrayType = []
 
     async def async_add(self, name, context=None):
         """Add a shopping list item."""
@@ -277,16 +277,16 @@ def async_reorder(self, item_ids, context=None):
             context=context,
         )
 
-    async def async_load(self):
+    async def async_load(self) -> None:
         """Load items."""
 
-        def load():
+        def load() -> JsonArrayType:
             """Load the items synchronously."""
-            return load_json(self.hass.config.path(PERSISTENCE), default=[])
+            return load_json_array(self.hass.config.path(PERSISTENCE))
 
         self.items = await self.hass.async_add_executor_job(load)
 
-    def save(self):
+    def save(self) -> None:
         """Save the items."""
         save_json(self.hass.config.path(PERSISTENCE), self.items)
 

homeassistant/util/json.py

@@ -4,6 +4,7 @@
 from collections.abc import Callable
 import json
 import logging
+from os import PathLike
 from typing import Any
 
 import orjson
@@ -12,6 +13,7 @@
 
 from .file import WriteError  # pylint: disable=unused-import # noqa: F401
 
+_SENTINEL = object()
 _LOGGER = logging.getLogger(__name__)
 
 JsonValueType = (
@@ -54,8 +56,10 @@ def json_loads_object(__obj: bytes | bytearray | memoryview | str) -> JsonObject
     raise ValueError(f"Expected JSON to be parsed as a dict got {type(value)}")
 
 
-def load_json(filename: str, default: list | dict | None = None) -> list | dict:
-    """Load JSON data from a file and return as dict or list.
+def load_json(
+    filename: str | PathLike, default: JsonValueType = _SENTINEL  # type: ignore[assignment]
+) -> JsonValueType:
+    """Load JSON data from a file.
 
     Defaults to returning empty dict if file is not found.
     """
@@ -71,7 +75,45 @@ def load_json(filename: str, default: list | dict | None = None) -> list | dict:
     except OSError as error:
         _LOGGER.exception("JSON file reading failed: %s", filename)
         raise HomeAssistantError(error) from error
-    return {} if default is None else default
+    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)}")
 
 
 def save_json(

tests/components/ios/test_init.py

@@ -13,7 +13,7 @@
 @pytest.fixture(autouse=True)
 def mock_load_json():
     """Mock load_json."""
-    with patch("homeassistant.components.ios.load_json", return_value={}):
+    with patch("homeassistant.components.ios.load_json_object", return_value={}):
         yield
 
 

tests/components/nest/test_config_flow_legacy.py

@@ -228,7 +228,7 @@ async def test_step_import(hass: HomeAssistant) -> None:
 async def test_step_import_with_token_cache(hass: HomeAssistant) -> None:
     """Test that we import existing token cache."""
     with patch("os.path.isfile", return_value=True), patch(
-        "homeassistant.components.nest.config_flow.load_json",
+        "homeassistant.components.nest.config_flow.load_json_object",
         return_value={"access_token": "yo"},
     ), patch(
         "homeassistant.components.nest.async_setup_legacy_entry", return_value=True

tests/util/test_json.py

@@ -4,7 +4,13 @@
 import pytest
 
 from homeassistant.exceptions import HomeAssistantError
-from homeassistant.util.json import json_loads_array, json_loads_object, load_json
+from homeassistant.util.json import (
+    json_loads_array,
+    json_loads_object,
+    load_json,
+    load_json_array,
+    load_json_object,
+)
 
 # Test data that can be saved as JSON
 TEST_JSON_A = {"a": 1, "B": "two"}
@@ -17,8 +23,74 @@ def test_load_bad_data(tmp_path: Path) -> None:
     fname = tmp_path / "test5.json"
     with open(fname, "w") as fh:
         fh.write(TEST_BAD_SERIALIED)
-    with pytest.raises(HomeAssistantError):
+    with pytest.raises(HomeAssistantError) as err:
         load_json(fname)
+    assert isinstance(err.value.__cause__, ValueError)
+
+
+def test_load_json_os_error() -> None:
+    """Test trying to load JSON data from a directory."""
+    fname = "/"
+    with pytest.raises(HomeAssistantError) as err:
+        load_json(fname)
+    assert isinstance(err.value.__cause__, OSError)
+
+
+def test_load_json_file_not_found_error() -> None:
+    """Test trying to load object data from inexistent JSON file."""
+    fname = "invalid_file.json"
+
+    assert load_json(fname) == {}
+    assert load_json(fname, default="") == ""
+    assert load_json_object(fname) == {}
+    assert load_json_object(fname, default={"Hi": "Peter"}) == {"Hi": "Peter"}
+    assert load_json_array(fname) == []
+    assert load_json_array(fname, default=["Hi"]) == ["Hi"]
+
+
+def test_load_json_value_data(tmp_path: Path) -> None:
+    """Test trying to load object data from JSON file."""
+    fname = tmp_path / "test5.json"
+    with open(fname, "w", encoding="utf8") as handle:
+        handle.write('"two"')
+
+    assert load_json(fname) == "two"
+    with pytest.raises(
+        HomeAssistantError, match="Expected JSON to be parsed as a dict"
+    ):
+        load_json_object(fname)
+    with pytest.raises(
+        HomeAssistantError, match="Expected JSON to be parsed as a list"
+    ):
+        load_json_array(fname)
+
+
+def test_load_json_object_data(tmp_path: Path) -> None:
+    """Test trying to load object data from JSON file."""
+    fname = tmp_path / "test5.json"
+    with open(fname, "w", encoding="utf8") as handle:
+        handle.write('{"a": 1, "B": "two"}')
+
+    assert load_json(fname) == {"a": 1, "B": "two"}
+    assert load_json_object(fname) == {"a": 1, "B": "two"}
+    with pytest.raises(
+        HomeAssistantError, match="Expected JSON to be parsed as a list"
+    ):
+        load_json_array(fname)
+
+
+def test_load_json_array_data(tmp_path: Path) -> None:
+    """Test trying to load array data from JSON file."""
+    fname = tmp_path / "test5.json"
+    with open(fname, "w", encoding="utf8") as handle:
+        handle.write('[{"a": 1, "B": "two"}]')
+
+    assert load_json(fname) == [{"a": 1, "B": "two"}]
+    assert load_json_array(fname) == [{"a": 1, "B": "two"}]
+    with pytest.raises(
+        HomeAssistantError, match="Expected JSON to be parsed as a dict"
+    ):
+        load_json_object(fname)
 
 
 def test_json_loads_array() -> None: