Add JSONType

Author: GrandMoff100

homeassistant_api/models/domains.py

@@ -20,6 +20,7 @@
 from pydantic import Field
 
 from homeassistant_api.errors import RequestError
+from homeassistant_api.utils import JSONType
 
 from .base import BaseModel
 from .states import State
@@ -55,13 +56,13 @@ def __init__(
 
     @classmethod
     def from_json(
-        cls, json: Dict[str, Any], client: Union["Client", "WebsocketClient"]
+        cls, json: Dict[str, JSONType], client: Union["Client", "WebsocketClient"]
     ) -> "Domain":
         """Constructs Domain and Service models from json data."""
         if "domain" not in json or "services" not in json:
             raise ValueError("Missing services or domain attribute in json argument.")
         domain = cls(domain_id=cast(str, json.get("domain")), _client=client)
-        services = json.get("services")
+        services = cast(dict[str, dict[str, JSONType]], json.get("services"))
         assert isinstance(services, dict)
         for service_id, data in services.items():
             domain._add_service(service_id, **data)
@@ -103,8 +104,6 @@ def __getattr__(self, attr: str):
 # https://github.com/home-assistant/frontend/blob/dev/src/data/selector.ts
 # https://github.com/home-assistant/home-assistant-js-websocket/blob/master/lib/types.ts
 
-number = Union[int, float]
-
 
 # Helpers
 class ServiceFieldSelectorEntityFilter(BaseModel):
@@ -124,8 +123,8 @@ class ServiceFieldSelectorDeviceFilter(BaseModel):
 class CropOptions(BaseModel):
     round: bool
     type: Optional[str]  # "image/jpeg" / "image/png"
-    quality: Optional[number] = None
-    aspectRatio: Optional[number] = None
+    quality: Optional[int | float] = None
+    aspectRatio: Optional[int | float] = None
 
 
 class SelectBoxOptionImage(BaseModel):
@@ -226,10 +225,10 @@ class ServiceFieldSelectorColorRGB(BaseModel):
 
 class ServiceFieldSelectorColorTemp(BaseModel):
     unit: Optional[str] = None
-    min: Optional[number] = None
-    max: Optional[number] = None
-    min_mireds: Optional[number] = None
-    max_mireds: Optional[number] = None
+    min: Optional[int | float] = None
+    max: Optional[int | float] = None
+    min_mireds: Optional[int | float] = None
+    max_mireds: Optional[int | float] = None
 
 
 class ServiceFieldSelectorCondition(BaseModel):
@@ -242,7 +241,7 @@ class ServiceFieldSelectorConfigEntry(BaseModel):
 
 class ServiceFieldSelectorConstant(BaseModel):
     label: Optional[str] = None
-    value: Union[str, number, bool]
+    value: Union[str, int, float, bool]
     translation_key: Optional[str] = None
 
 
@@ -349,9 +348,9 @@ class ServiceFieldSelectorNavigation(BaseModel):
 
 
 class ServiceFieldSelectorNumber(BaseModel):
-    min: Optional[number] = None
-    max: Optional[number] = None
-    step: Optional[Union[number, str]] = None
+    min: Optional[int | float] = None
+    max: Optional[int | float] = None
+    step: Optional[Union[int | float, str]] = None
     unit_of_measurement: Optional[str] = None
     mode: Optional[ServiceFieldSelectorNumberMode] = None
     slider_ticks: Optional[bool] = None
@@ -374,7 +373,7 @@ class ServiceFieldSelectorObject(BaseModel):
 
 class ServiceFieldSelectorQRCode(BaseModel):
     data: str
-    scale: Optional[number] = None
+    scale: Optional[int | float] = None
     error_correction_level: Optional[ServiceFieldSelectorQRCodeErrorCorrectionLevel] = (
         None
     )
@@ -555,14 +554,12 @@ class ServiceField(BaseModel):
     """Model for service parameters/fields."""
 
     description: Optional[str] = None
-    example: Optional[Union[str, number, bool, List[str], Dict]] = None
-    default: Optional[Union[str, number, bool, List[str], Dict]] = None
+    example: Optional[JSONType] = None
+    default: Optional[JSONType] = None
     name: Optional[str] = None
     required: Optional[bool] = None
     advanced: Optional[bool] = None
-    selector: Optional[Dict[str, Any]] = (
-        None  # TODO: I believe it would be beneficial to parse it the way I do
-    )
+    selector: Optional[ServiceFieldSelector] = None
     filter: Optional[ServiceFieldFilter] = None
 
 
@@ -588,8 +585,8 @@ class Service(BaseModel):
 
     def trigger(self, entity_id: Optional[str] = None, **service_data) -> Union[
         Tuple[State, ...],
-        Tuple[Tuple[State, ...], Dict[str, Any]],
-        dict[str, Any],
+        Tuple[Tuple[State, ...], dict[str, JSONType]],
+        dict[str, JSONType],
         None,
     ]:
         """Triggers the service associated with this object."""
@@ -612,7 +609,7 @@ def trigger(self, entity_id: Optional[str] = None, **service_data) -> Union[
 
     async def async_trigger(
         self, entity_id: Optional[str] = None, **service_data
-    ) -> Union[Tuple[State, ...], Tuple[Tuple[State, ...], Dict[str, Any]]]:
+    ) -> Union[Tuple[State, ...], Tuple[Tuple[State, ...], dict[str, JSONType]]]:
         """Triggers the service associated with this object."""
         if entity_id is not None:
             service_data["entity_id"] = entity_id
@@ -639,12 +636,14 @@ async def async_trigger(
     def __call__(self, entity_id: Optional[str] = None, **service_data) -> Union[
         Union[
             Tuple[State, ...],
-            Tuple[Tuple[State, ...], Dict[str, Any]],
-            dict[str, Any],
+            Tuple[Tuple[State, ...], dict[str, JSONType]],
+            dict[str, JSONType],
             None,
         ],
         Coroutine[
-            Any, Any, Union[Tuple[State, ...], Tuple[Tuple[State, ...], Dict[str, Any]]]
+            Any,
+            Any,
+            Union[Tuple[State, ...], Tuple[Tuple[State, ...], dict[str, JSONType]]],
         ],
     ]:
         """

homeassistant_api/models/events.py

@@ -1,9 +1,11 @@
 """Event Model File"""
 
-from typing import TYPE_CHECKING, Any, Dict, Optional
+from typing import TYPE_CHECKING, Optional
 
 from pydantic import Field
 
+from homeassistant_api.utils import JSONType
+
 from .base import BaseModel
 
 if TYPE_CHECKING:
@@ -38,6 +40,6 @@ async def async_fire(self, **event_data) -> str:
         return await self._client.async_fire_event(self.event, **event_data)
 
     @classmethod
-    def from_json(cls, json: Dict[str, Any], client: "Client") -> "Event":
+    def from_json(cls, json: dict[str, JSONType], client: "Client") -> "Event":
         """Constructs Event model from json data"""
         return cls(**json, _client=client)

homeassistant_api/models/states.py

@@ -1,11 +1,13 @@
 """Module for the Entity State model."""
 
 from datetime import datetime, timezone
-from typing import Any, Dict, Optional
+from typing import Optional
 
 from pydantic import Field
 
-from .base import BaseModel, DatetimeIsoField
+from homeassistant_api.utils import JSONType
+
+from homeassistant_api.base import BaseModel, DatetimeIsoField
 
 
 class Context(BaseModel):
@@ -25,7 +27,7 @@ class Context(BaseModel):
     )
 
     @classmethod
-    def from_json(cls, json: Dict[str, Any]) -> "Context":
+    def from_json(cls, json: dict[str, JSONType]) -> "Context":
         """Constructs Context model from json data"""
         return cls.model_validate(json)
 
@@ -37,7 +39,7 @@ class State(BaseModel):
     state: str = Field(
         ..., description="The string representation of the state of the entity."
     )
-    attributes: Dict[str, Any] = Field(
+    attributes: dict[str, JSONType] = Field(
         {}, description="A dictionary of extra attributes of the state."
     )
     last_changed: DatetimeIsoField = Field(
@@ -57,6 +59,6 @@ class State(BaseModel):
     )
 
     @classmethod
-    def from_json(cls, json: Dict[str, Any]) -> "State":
+    def from_json(cls, json: dict[str, JSONType]) -> "State":
         """Constructs State model from json data"""
         return cls.model_validate(json)

homeassistant_api/models/websocket.py

@@ -2,6 +2,8 @@
 
 from typing import Any, Literal, Optional, Union
 
+from homeassistant_api.utils import JSONType
+
 from .base import BaseModel
 from .states import Context, DatetimeIsoField
 
@@ -67,7 +69,7 @@ class FiredEvent(BaseModel):
     """A model to parse the `event` key of fired event websocket responses."""
 
     event_type: str
-    data: dict[str, Any]
+    data: dict[str, JSONType]
 
     origin: Literal["LOCAL", "REMOTE"]
     # REMOTE if another API client or webhook fired the event
@@ -79,14 +81,14 @@ class FiredEvent(BaseModel):
 
 class TemplateEvent(BaseModel):
     result: str
-    listeners: dict[str, Any]
+    listeners: dict[str, JSONType]
 
 
 class FiredTrigger(BaseModel):
     """A model to parse the `trigger` key of fired event websocket responses."""
 
     context: Optional[Context]
-    variables: dict[str, Any]
+    variables: dict[str, JSONType]
 
 
 class EventResponse(BaseModel):

homeassistant_api/processing.py

@@ -11,7 +11,9 @@
 from requests import Response
 from requests_cache.models.response import CachedResponse
 
-from .errors import (
+from homeassistant_api.utils import JSONType
+
+from homeassistant_api.errors import (
     EndpointNotFoundError,
     InternalServerError,
     MalformedDataError,
@@ -109,10 +111,10 @@ def process(self) -> Any:
 
 # List of default processors
 @Processing.processor("application/json")  # type: ignore[arg-type]
-def process_json(response: ResponseType) -> dict[str, Any]:
+def process_json(response: ResponseType) -> dict[str, JSONType]:
     """Returns the json dict content of the response."""
     try:
-        return cast(dict[str, Any], response.json())
+        return cast(dict[str, JSONType], response.json())
     except (json.JSONDecodeError, simplejson.JSONDecodeError) as err:
         raise MalformedDataError(
             f"Home Assistant responded with non-json response: {repr(response.text)}"
@@ -127,10 +129,10 @@ def process_text(response: ResponseType) -> str:
 
 
 @Processing.processor("application/json")  # type: ignore[arg-type]
-async def async_process_json(response: AsyncResponseType) -> dict[str, Any]:
+async def async_process_json(response: AsyncResponseType) -> dict[str, JSONType]:
     """Returns the json dict content of the response."""
     try:
-        return cast(dict[str, Any], await response.json())
+        return cast(dict[str, JSONType], await response.json())
     except (json.JSONDecodeError, simplejson.JSONDecodeError) as err:
         raise MalformedDataError(
             f"Home Assistant responded with non-json response: {repr(await response.text())}"

homeassistant_api/rawasyncclient.py

@@ -21,13 +21,13 @@
 )
 
 import aiohttp
-import aiohttp_client_cache
+import aiohttp_client_cache.session
 
 from .errors import BadTemplateError, RequestError, RequestTimeoutError
 from .models import Domain, Entity, Event, Group, History, LogbookEntry, State
 from .processing import AsyncResponseType, Processing
 from .rawbaseclient import RawBaseClient
-from .utils import prepare_entity_id
+from .utils import JSONType, prepare_entity_id
 
 if TYPE_CHECKING:
     from homeassistant_api import Client
@@ -47,14 +47,14 @@ class RawAsyncClient(RawBaseClient):
     """  # pylint: disable=line-too-long
 
     async_cache_session: Union[
-        aiohttp_client_cache.CachedSession, aiohttp.ClientSession
+        aiohttp_client_cache.session.CachedSession, aiohttp.ClientSession
     ]
 
     def __init__(
         self,
         *args,
         async_cache_session: Union[
-            aiohttp_client_cache.CachedSession,
+            aiohttp_client_cache.session.CachedSession,
             Literal[False],
             Literal[None],
         ] = None,  # Explicitly disable cache with async_cache_session=False
@@ -129,12 +129,12 @@ async def async_get_error_log(self) -> str:
         """
         return cast(str, await self.async_request("error_log"))
 
-    async def async_get_config(self) -> Dict[str, Any]:
+    async def async_get_config(self) -> dict[str, JSONType]:
         """
         Returns the yaml configuration of homeassistant.
         :code:`GET /api/config`
         """
-        return cast(Dict[str, Any], await self.async_request("config"))
+        return cast(dict[str, JSONType], await self.async_request("config"))
 
     async def async_get_logbook_entries(
         self,
@@ -272,7 +272,7 @@ async def async_get_domains(self) -> Dict[str, Domain]:
         data = await self.async_request("services")
         domains = map(
             lambda json: Domain.from_json(json, client=cast(Client, self)),
-            cast(Tuple[Dict[str, Any], ...], data),
+            cast(Tuple[dict[str, JSONType], ...], data),
         )
         return {domain.domain_id: domain for domain in domains}
 
@@ -288,7 +288,7 @@ async def async_trigger_service(
         self,
         domain: str,
         service: str,
-        **service_data: Union[Dict[str, Any], List[Any], str],
+        **service_data: Union[dict[str, JSONType], List[Any], str],
     ) -> Tuple[State, ...]:
         """
         Tells Home Assistant to trigger a service, returns all states changed while in the process of being called.
@@ -305,16 +305,16 @@ async def async_trigger_service_with_response(
         self,
         domain: str,
         service: str,
-        **service_data: Union[Dict[str, Any], List[Any], str],
-    ) -> tuple[tuple[State, ...], dict[str, Any]]:
+        **service_data: Union[dict[str, JSONType], List[Any], str],
+    ) -> tuple[tuple[State, ...], dict[str, JSONType]]:
         """
         Tells Home Assistant to trigger a service, returns the response from the service call.
         :code:`POST /api/services/<domain>/<service>`
 
         Returns a list of the states changed and the response from the service call.
         """
         data = cast(
-            dict[str, Any],
+            dict[str, dict[str, JSONType]],
             await self.async_request(
                 join("services", domain, service) + "?return_response",
                 method="POST",
@@ -383,7 +383,7 @@ async def async_get_events(self) -> Tuple[Event, ...]:
         return tuple(
             map(
                 lambda json: Event.from_json(json, client=cast(Client, self)),
-                cast(List[Dict[str, Any]], data),
+                cast(List[dict[str, JSONType]], data),
             )
         )