Alexa typing part 6 (state_report) (#97920)

state_report

homeassistant/components/alexa/state_report.py

@@ -5,7 +5,8 @@ import asyncio
 from http import HTTPStatus
 import json
 import logging
-from typing import TYPE_CHECKING, cast
+from types import MappingProxyType
+from typing import TYPE_CHECKING, Any, cast
 from uuid import uuid4
 
 import aiohttp
@@ -46,17 +47,22 @@ DEFAULT_TIMEOUT = 10
 class AlexaDirective:
     """An incoming Alexa directive."""
 
-    def __init__(self, request):
+    entity: State
+    entity_id: str | None
+    endpoint: AlexaEntity
+    instance: str | None
+
+    def __init__(self, request: dict[str, Any]) -> None:
         """Initialize a directive."""
-        self._directive = request[API_DIRECTIVE]
+        self._directive: dict[str, Any] = request[API_DIRECTIVE]
-        self.namespace = self._directive[API_HEADER]["namespace"]
+        self.namespace: str = self._directive[API_HEADER]["namespace"]
-        self.name = self._directive[API_HEADER]["name"]
+        self.name: str = self._directive[API_HEADER]["name"]
-        self.payload = self._directive[API_PAYLOAD]
+        self.payload: dict[str, Any] = self._directive[API_PAYLOAD]
-        self.has_endpoint = API_ENDPOINT in self._directive
+        self.has_endpoint: bool = API_ENDPOINT in self._directive
+        self.instance = None
+        self.entity_id = None
 
-        self.entity = self.entity_id = self.endpoint = self.instance = None
+    def load_entity(self, hass: HomeAssistant, config: AbstractConfig) -> None:
-
-    def load_entity(self, hass, config):
         """Set attributes related to the entity for this request.
 
         Sets these attributes when self.has_endpoint is True:
@@ -71,18 +77,24 @@ class AlexaDirective:
         Will raise AlexaInvalidEndpointError if the endpoint in the request is
         malformed or nonexistent.
         """
-        _endpoint_id = self._directive[API_ENDPOINT]["endpointId"]
+        _endpoint_id: str = self._directive[API_ENDPOINT]["endpointId"]
         self.entity_id = _endpoint_id.replace("#", ".")
 
-        self.entity = hass.states.get(self.entity_id)
+        entity: State | None = hass.states.get(self.entity_id)
-        if not self.entity or not config.should_expose(self.entity_id):
+        if not entity or not config.should_expose(self.entity_id):
             raise AlexaInvalidEndpointError(_endpoint_id)
+        self.entity = entity
 
         self.endpoint = ENTITY_ADAPTERS[self.entity.domain](hass, config, self.entity)
         if "instance" in self._directive[API_HEADER]:
             self.instance = self._directive[API_HEADER]["instance"]
 
-    def response(self, name="Response", namespace="Alexa", payload=None):
+    def response(
+        self,
+        name: str = "Response",
+        namespace: str = "Alexa",
+        payload: dict[str, Any] | None = None,
+    ) -> AlexaResponse:
         """Create an API formatted response.
 
         Async friendly.
@@ -100,11 +112,11 @@ class AlexaDirective:
 
     def error(
         self,
-        namespace="Alexa",
+        namespace: str = "Alexa",
-        error_type="INTERNAL_ERROR",
+        error_type: str = "INTERNAL_ERROR",
-        error_message="",
+        error_message: str = "",
-        payload=None,
+        payload: dict[str, Any] | None = None,
-    ):
+    ) -> AlexaResponse:
         """Create a API formatted error response.
 
         Async friendly.
@@ -127,10 +139,12 @@ class AlexaDirective:
 class AlexaResponse:
     """Class to hold a response."""
 
-    def __init__(self, name, namespace, payload=None):
+    def __init__(
+        self, name: str, namespace: str, payload: dict[str, Any] | None = None
+    ) -> None:
         """Initialize the response."""
         payload = payload or {}
-        self._response = {
+        self._response: dict[str, Any] = {
             API_EVENT: {
                 API_HEADER: {
                     "namespace": namespace,
@@ -143,16 +157,16 @@ class AlexaResponse:
         }
 
     @property
-    def name(self):
+    def name(self) -> str:
         """Return the name of this response."""
         return self._response[API_EVENT][API_HEADER]["name"]
 
     @property
-    def namespace(self):
+    def namespace(self) -> str:
         """Return the namespace of this response."""
         return self._response[API_EVENT][API_HEADER]["namespace"]
 
-    def set_correlation_token(self, token):
+    def set_correlation_token(self, token: str) -> None:
         """Set the correlationToken.
 
         This should normally mirror the value from a request, and is set by
@@ -160,7 +174,9 @@ class AlexaResponse:
         """
         self._response[API_EVENT][API_HEADER]["correlationToken"] = token
 
-    def set_endpoint_full(self, bearer_token, endpoint_id, cookie=None):
+    def set_endpoint_full(
+        self, bearer_token: str | None, endpoint_id: str | None
+    ) -> None:
         """Set the endpoint dictionary.
 
         This is used to send proactive messages to Alexa.
@@ -172,10 +188,7 @@ class AlexaResponse:
         if endpoint_id is not None:
             self._response[API_EVENT][API_ENDPOINT]["endpointId"] = endpoint_id
 
-        if cookie is not None:
+    def set_endpoint(self, endpoint: dict[str, Any]) -> None:
-            self._response[API_EVENT][API_ENDPOINT]["cookie"] = cookie
-
-    def set_endpoint(self, endpoint):
         """Set the endpoint.
 
         This should normally mirror the value from a request, and is set by
@@ -187,7 +200,7 @@ class AlexaResponse:
         context = self._response.setdefault(API_CONTEXT, {})
         return context.setdefault("properties", [])
 
-    def add_context_property(self, prop):
+    def add_context_property(self, prop: dict[str, Any]) -> None:
         """Add a property to the response context.
 
         The Alexa response includes a list of properties which provides
@@ -204,7 +217,7 @@ class AlexaResponse:
         """
         self._properties().append(prop)
 
-    def merge_context_properties(self, endpoint):
+    def merge_context_properties(self, endpoint: AlexaEntity) -> None:
         """Add all properties from given endpoint if not already set.
 
         Handlers should be using .add_context_property().
@@ -216,12 +229,14 @@ class AlexaResponse:
             if (prop["namespace"], prop["name"]) not in already_set:
                 self.add_context_property(prop)
 
-    def serialize(self):
+    def serialize(self) -> dict[str, Any]:
         """Return response as a JSON-able data structure."""
         return self._response
 
 
-async def async_enable_proactive_mode(hass, smart_home_config):
+async def async_enable_proactive_mode(
+    hass: HomeAssistant, smart_home_config: AbstractConfig
+):
     """Enable the proactive mode.
 
     Proactive mode makes this component report state changes to Alexa.
@@ -233,12 +248,12 @@ async def async_enable_proactive_mode(hass, smart_home_config):
     def extra_significant_check(
         hass: HomeAssistant,
         old_state: str,
-        old_attrs: dict,
+        old_attrs: dict[Any, Any] | MappingProxyType[Any, Any],
-        old_extra_arg: dict,
+        old_extra_arg: Any,
         new_state: str,
-        new_attrs: dict,
+        new_attrs: dict[str, Any] | MappingProxyType[Any, Any],
-        new_extra_arg: dict,
+        new_extra_arg: Any,
-    ):
+    ) -> bool:
         """Check if the serialized data has changed."""
         return old_extra_arg is not None and old_extra_arg != new_extra_arg
 
@@ -248,7 +263,7 @@ async def async_enable_proactive_mode(hass, smart_home_config):
         changed_entity: str,
         old_state: State | None,
         new_state: State | None,
-    ):
+    ) -> None:
         if not hass.is_running:
             return
 
@@ -307,8 +322,13 @@ async def async_enable_proactive_mode(hass, smart_home_config):
 
 
 async def async_send_changereport_message(
-    hass, config, alexa_entity, alexa_properties, *, invalidate_access_token=True
+    hass: HomeAssistant,
-):
+    config: AbstractConfig,
+    alexa_entity: AlexaEntity,
+    alexa_properties: list[dict[str, Any]],
+    *,
+    invalidate_access_token: bool = True,
+) -> None:
     """Send a ChangeReport message for an Alexa entity.
 
     https://developer.amazon.com/docs/smarthome/state-reporting-for-a-smart-home-skill.html#report-state-with-changereport-events
@@ -322,11 +342,11 @@ async def async_send_changereport_message(
         )
         return
 
-    headers = {"Authorization": f"Bearer {token}"}
+    headers: dict[str, Any] = {"Authorization": f"Bearer {token}"}
 
     endpoint = alexa_entity.alexa_id()
 
-    payload = {
+    payload: dict[str, Any] = {
         API_CHANGE: {
             "cause": {"type": Cause.APP_INTERACTION},
             "properties": alexa_properties,
@@ -339,6 +359,7 @@ async def async_send_changereport_message(
     message_serialized = message.serialize()
     session = async_get_clientsession(hass)
 
+    assert config.endpoint is not None
     try:
         async with async_timeout.timeout(DEFAULT_TIMEOUT):
             response = await session.post(
@@ -393,9 +414,9 @@ async def async_send_add_or_update_message(
     """
     token = await config.async_get_access_token()
 
-    headers = {"Authorization": f"Bearer {token}"}
+    headers: dict[str, Any] = {"Authorization": f"Bearer {token}"}
 
-    endpoints = []
+    endpoints: list[dict[str, Any]] = []
 
     for entity_id in entity_ids:
         if (domain := entity_id.split(".", 1)[0]) not in ENTITY_ADAPTERS:
@@ -407,7 +428,10 @@ async def async_send_add_or_update_message(
         alexa_entity = ENTITY_ADAPTERS[domain](hass, config, state)
         endpoints.append(alexa_entity.serialize_discovery())
 
-    payload = {"endpoints": endpoints, "scope": {"type": "BearerToken", "token": token}}
+    payload: dict[str, Any] = {
+        "endpoints": endpoints,
+        "scope": {"type": "BearerToken", "token": token},
+    }
 
     message = AlexaResponse(
         name="AddOrUpdateReport", namespace="Alexa.Discovery", payload=payload
@@ -431,9 +455,9 @@ async def async_send_delete_message(
     """
     token = await config.async_get_access_token()
 
-    headers = {"Authorization": f"Bearer {token}"}
+    headers: dict[str, Any] = {"Authorization": f"Bearer {token}"}
 
-    endpoints = []
+    endpoints: list[dict[str, Any]] = []
 
     for entity_id in entity_ids:
         domain = entity_id.split(".", 1)[0]
@@ -443,7 +467,10 @@ async def async_send_delete_message(
 
         endpoints.append({"endpointId": generate_alexa_id(entity_id)})
 
-    payload = {"endpoints": endpoints, "scope": {"type": "BearerToken", "token": token}}
+    payload: dict[str, Any] = {
+        "endpoints": endpoints,
+        "scope": {"type": "BearerToken", "token": token},
+    }
 
     message = AlexaResponse(
         name="DeleteReport", namespace="Alexa.Discovery", payload=payload
@@ -458,14 +485,16 @@ async def async_send_delete_message(
     )
 
 
-async def async_send_doorbell_event_message(hass, config, alexa_entity):
+async def async_send_doorbell_event_message(
+    hass: HomeAssistant, config: AbstractConfig, alexa_entity: AlexaEntity
+) -> None:
     """Send a DoorbellPress event message for an Alexa entity.
 
     https://developer.amazon.com/en-US/docs/alexa/device-apis/alexa-doorbelleventsource.html
     """
     token = await config.async_get_access_token()
 
-    headers = {"Authorization": f"Bearer {token}"}
+    headers: dict[str, Any] = {"Authorization": f"Bearer {token}"}
 
     endpoint = alexa_entity.alexa_id()
 
@@ -483,6 +512,7 @@ async def async_send_doorbell_event_message(hass, config, alexa_entity):
     message_serialized = message.serialize()
     session = async_get_clientsession(hass)
 
+    assert config.endpoint is not None
     try:
         async with async_timeout.timeout(DEFAULT_TIMEOUT):
             response = await session.post(

homeassistant/components/cloud/alexa_config.py

@@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any
 import aiohttp
 import async_timeout
 from hass_nabucasa import Cloud, cloud_api
+from yarl import URL
 
 from homeassistant.components import persistent_notification
 from homeassistant.components.alexa import (
@@ -149,7 +150,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
         self._token_valid: datetime | None = None
         self._cur_entity_prefs = async_get_assistant_settings(hass, CLOUD_ALEXA)
         self._alexa_sync_unsub: Callable[[], None] | None = None
-        self._endpoint: Any = None
+        self._endpoint: str | URL | None = None
 
     @property
     def enabled(self) -> bool:
@@ -175,7 +176,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
         )
 
     @property
-    def endpoint(self) -> Any | None:
+    def endpoint(self) -> str | URL | None:
         """Endpoint for report state."""
         if self._endpoint is None:
             raise ValueError("No endpoint available. Fetch access token first")
@@ -309,7 +310,7 @@ class CloudAlexaConfig(alexa_config.AbstractConfig):
         """Invalidate access token."""
         self._token_valid = None
 
-    async def async_get_access_token(self) -> Any:
+    async def async_get_access_token(self) -> str | None:
         """Get an access token."""
         if self._token_valid is not None and self._token_valid > utcnow():
             return self._token