Use typed ConfigEntry in ring coordinator (#136457)

* Use typed ConfigEntry in ring coordinator

* Make config_entry a positional argument for coordinator
This commit is contained in:
Steven B. 2025-01-27 14:56:04 +00:00 committed by GitHub
parent 037a0f25a4
commit 84561b7446
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 36 additions and 28 deletions

View File

@ -2,39 +2,29 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass
import logging import logging
from typing import Any, cast from typing import Any, cast
import uuid import uuid
from ring_doorbell import Auth, Ring, RingDevices from ring_doorbell import Auth, Ring
from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN from homeassistant.components.camera import DOMAIN as CAMERA_DOMAIN
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import APPLICATION_NAME, CONF_DEVICE_ID, CONF_TOKEN from homeassistant.const import APPLICATION_NAME, CONF_DEVICE_ID, CONF_TOKEN
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import device_registry as dr, entity_registry as er from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import CONF_LISTEN_CREDENTIALS, DOMAIN, PLATFORMS from .const import CONF_LISTEN_CREDENTIALS, DOMAIN, PLATFORMS
from .coordinator import RingDataCoordinator, RingListenCoordinator from .coordinator import (
RingConfigEntry,
RingData,
RingDataCoordinator,
RingListenCoordinator,
)
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@dataclass
class RingData:
"""Class to support type hinting of ring data collection."""
api: Ring
devices: RingDevices
devices_coordinator: RingDataCoordinator
listen_coordinator: RingListenCoordinator
type RingConfigEntry = ConfigEntry[RingData]
def get_auth_user_agent() -> str: def get_auth_user_agent() -> str:
"""Return user-agent for Auth instantiation. """Return user-agent for Auth instantiation.
@ -71,10 +61,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: RingConfigEntry) -> bool
) )
ring = Ring(auth) ring = Ring(auth)
devices_coordinator = RingDataCoordinator(hass, ring) devices_coordinator = RingDataCoordinator(hass, entry, ring)
listen_credentials = entry.data.get(CONF_LISTEN_CREDENTIALS) listen_credentials = entry.data.get(CONF_LISTEN_CREDENTIALS)
listen_coordinator = RingListenCoordinator( listen_coordinator = RingListenCoordinator(
hass, ring, listen_credentials, listen_credentials_updater hass, entry, ring, listen_credentials, listen_credentials_updater
) )
await devices_coordinator.async_config_entry_first_refresh() await devices_coordinator.async_config_entry_first_refresh()
@ -91,19 +81,19 @@ async def async_setup_entry(hass: HomeAssistant, entry: RingConfigEntry) -> bool
return True return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_unload_entry(hass: HomeAssistant, entry: RingConfigEntry) -> bool:
"""Unload Ring entry.""" """Unload Ring entry."""
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS) return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def async_remove_config_entry_device( async def async_remove_config_entry_device(
hass: HomeAssistant, config_entry: ConfigEntry, device_entry: dr.DeviceEntry hass: HomeAssistant, entry: RingConfigEntry, device_entry: dr.DeviceEntry
) -> bool: ) -> bool:
"""Remove a config entry from a device.""" """Remove a config entry from a device."""
return True return True
async def async_migrate_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_migrate_entry(hass: HomeAssistant, entry: RingConfigEntry) -> bool:
"""Migrate old config entry.""" """Migrate old config entry."""
entry_version = entry.version entry_version = entry.version
entry_minor_version = entry.minor_version entry_minor_version = entry.minor_version

View File

@ -1,9 +1,12 @@
"""Data coordinators for the ring integration.""" """Data coordinators for the ring integration."""
from __future__ import annotations
from asyncio import TaskGroup from asyncio import TaskGroup
from collections.abc import Callable, Coroutine from collections.abc import Callable, Coroutine
from dataclasses import dataclass
import logging import logging
from typing import TYPE_CHECKING, Any from typing import Any
from ring_doorbell import ( from ring_doorbell import (
AuthenticationError, AuthenticationError,
@ -15,7 +18,7 @@ from ring_doorbell import (
) )
from ring_doorbell.listen import RingEventListener from ring_doorbell.listen import RingEventListener
from homeassistant import config_entries from homeassistant.config_entries import ConfigEntry
from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryAuthFailed from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import ( from homeassistant.helpers.update_coordinator import (
@ -29,6 +32,19 @@ from .const import SCAN_INTERVAL
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@dataclass
class RingData:
"""Class to support type hinting of ring data collection."""
api: Ring
devices: RingDevices
devices_coordinator: RingDataCoordinator
listen_coordinator: RingListenCoordinator
type RingConfigEntry = ConfigEntry[RingData]
async def _call_api[*_Ts, _R]( async def _call_api[*_Ts, _R](
hass: HomeAssistant, hass: HomeAssistant,
target: Callable[[*_Ts], Coroutine[Any, Any, _R]], target: Callable[[*_Ts], Coroutine[Any, Any, _R]],
@ -52,9 +68,12 @@ async def _call_api[*_Ts, _R](
class RingDataCoordinator(DataUpdateCoordinator[RingDevices]): class RingDataCoordinator(DataUpdateCoordinator[RingDevices]):
"""Base class for device coordinators.""" """Base class for device coordinators."""
config_entry: RingConfigEntry
def __init__( def __init__(
self, self,
hass: HomeAssistant, hass: HomeAssistant,
config_entry: RingConfigEntry,
ring_api: Ring, ring_api: Ring,
) -> None: ) -> None:
"""Initialize my coordinator.""" """Initialize my coordinator."""
@ -63,6 +82,7 @@ class RingDataCoordinator(DataUpdateCoordinator[RingDevices]):
name="devices", name="devices",
logger=_LOGGER, logger=_LOGGER,
update_interval=SCAN_INTERVAL, update_interval=SCAN_INTERVAL,
config_entry=config_entry,
) )
self.ring_api: Ring = ring_api self.ring_api: Ring = ring_api
self.first_call: bool = True self.first_call: bool = True
@ -107,11 +127,12 @@ class RingDataCoordinator(DataUpdateCoordinator[RingDevices]):
class RingListenCoordinator(BaseDataUpdateCoordinatorProtocol): class RingListenCoordinator(BaseDataUpdateCoordinatorProtocol):
"""Global notifications coordinator.""" """Global notifications coordinator."""
config_entry: config_entries.ConfigEntry config_entry: RingConfigEntry
def __init__( def __init__(
self, self,
hass: HomeAssistant, hass: HomeAssistant,
config_entry: RingConfigEntry,
ring_api: Ring, ring_api: Ring,
listen_credentials: dict[str, Any] | None, listen_credentials: dict[str, Any] | None,
listen_credentials_updater: Callable[[dict[str, Any]], None], listen_credentials_updater: Callable[[dict[str, Any]], None],
@ -126,9 +147,6 @@ class RingListenCoordinator(BaseDataUpdateCoordinatorProtocol):
self._listeners: dict[CALLBACK_TYPE, tuple[CALLBACK_TYPE, object | None]] = {} self._listeners: dict[CALLBACK_TYPE, tuple[CALLBACK_TYPE, object | None]] = {}
self._listen_callback_id: int | None = None self._listen_callback_id: int | None = None
config_entry = config_entries.current_entry.get()
if TYPE_CHECKING:
assert config_entry
self.config_entry = config_entry self.config_entry = config_entry
self.start_timeout = 10 self.start_timeout = 10
self.config_entry.async_on_unload(self.async_shutdown) self.config_entry.async_on_unload(self.async_shutdown)