Add config_entries subscribe method

Author: mj23000

homeassistant_api/models/__init__.py

@@ -3,7 +3,9 @@
 from .base import BaseModel
 from .config_entries import (
     ConfigEntry,
+    ConfigEntryChange,
     ConfigEntryDisabler,
+    ConfigEntryEvent,
     ConfigEntryState,
     ConfigFlowContext,
     ConfigSubEntry,
@@ -45,4 +47,6 @@
     "ConfigEntryState",
     "ConfigEntry",
     "ConfigSubEntry",
+    "ConfigEntryChange",
+    "ConfigEntryEvent",
 )

homeassistant_api/models/config_entries.py

@@ -143,3 +143,16 @@ class ConfigSubEntry(BaseModel):
     subentry_type: str
     title: str
     unique_id: Optional[str]
+
+
+class ConfigEntryChange(str, Enum):
+    """What was changed in a config entry."""
+
+    ADDED = "added"
+    REMOVED = "removed"
+    UPDATED = "updated"
+
+
+class ConfigEntryEvent(BaseModel):
+    type: Optional[ConfigEntryChange]
+    entry: ConfigEntry

homeassistant_api/models/websocket.py

@@ -1,10 +1,11 @@
 """A module defining the responses we expect from the websocket API."""
 
-from typing import Any, Literal, Optional, Union
+from typing import Any, List, Literal, Optional, Union
 
 from homeassistant_api.utils import JSONType
 
 from .base import BaseModel, DatetimeIsoField
+from .config_entries import ConfigEntryEvent
 from .states import Context
 
 __all__ = (
@@ -99,4 +100,4 @@ class EventResponse(BaseModel):
 
     id: int
     type: Literal["event"]
-    event: Union[FiredEvent, FiredTrigger, TemplateEvent]
+    event: Union[FiredEvent, FiredTrigger, TemplateEvent, List[ConfigEntryEvent]]

homeassistant_api/websocket.py

@@ -5,6 +5,7 @@
 
 from homeassistant_api.models import (
     ConfigEntry,
+    ConfigEntryEvent,
     ConfigSubEntry,
     DisableEnableResult,
     Domain,
@@ -287,7 +288,38 @@ def get_config_entries(
             )
         )
 
-    # TODO: config_entries/subscribe
+    def _subscribe_config_entries(self) -> int:
+        """
+        Subscribe to config entry flows.
+
+        Sends command :code:`{"type": "config_entries/subscribe"}`.
+        """
+
+        return self.recv(self.send("config_entries/subscribe")).id
+
+    @contextlib.contextmanager
+    def listen_config_entries(
+        self, disconnect_client: bool = True
+    ) -> Generator[Generator[ConfigEntryEvent, None, None], None, None]:
+        """
+        Listen to all config entry flow events.
+
+        For example:
+
+        .. code-block:: python
+
+            with ws_client.listen_config_entries() as flows:
+                for i, flow in zip(range(2), flows):  # to only wait for two flows to be received
+                    print(flow)
+        """
+        subscription = self._subscribe_config_entries()
+        cast(Generator[ConfigEntryEvent, None, None], self._wait_for(subscription))
+        # There is no "unsubscribe" method available for these events.
+        # Provide the ability to "unsubscribe" by disconnecting and reconnecting the Websocket client.
+        if disconnect_client:
+            logger.info("Reloading Websocket Client. Undefined behavior may occur.")
+            self.__exit__(None, None, None)
+            self.__enter__()
 
     # UNTESTED
     def get_entry_subentries(self, entry_id: str) -> Tuple[ConfigSubEntry, ...]:
@@ -468,14 +500,14 @@ def _subscribe_trigger(self, trigger: str, **trigger_fields) -> int:
 
     def _wait_for(
         self, subscription_id: int
-    ) -> Generator[Union[FiredEvent, FiredTrigger], None, None]:
+    ) -> Generator[Union[FiredEvent, FiredTrigger, ConfigEntryEvent], None, None]:
         """
         An iterator that waits for events of a certain type.
         """
         while True:
             yield cast(
                 Union[
-                    FiredEvent, FiredTrigger
+                    FiredEvent, FiredTrigger, ConfigEntryEvent
                 ],  # we can cast this because TemplateEvent is only used for rendering templates
                 cast(EventResponse, self.recv(subscription_id)).event,
             )