From 093eae15216dde1e2f6eae4a9128cb559dc64287 Mon Sep 17 00:00:00 2001 From: ludeeus Date: Tue, 23 Jul 2024 07:14:40 +0000 Subject: [PATCH] Add and use coordinator --- custom_components/healthchecksio/__init__.py | 94 ++++---------- .../healthchecksio/binary_sensor.py | 116 +++++++++--------- .../healthchecksio/config_flow.py | 7 +- custom_components/healthchecksio/const.py | 12 +- .../healthchecksio/coordinator.py | 78 ++++++++++++ 5 files changed, 175 insertions(+), 132 deletions(-) create mode 100644 custom_components/healthchecksio/coordinator.py diff --git a/custom_components/healthchecksio/__init__.py b/custom_components/healthchecksio/__init__.py index 0dfb5c9..eda4187 100644 --- a/custom_components/healthchecksio/__init__.py +++ b/custom_components/healthchecksio/__init__.py @@ -5,25 +5,21 @@ https://github.com/custom-components/healthchecksio """ -import asyncio -from datetime import timedelta +from __future__ import annotations + from logging import getLogger +from typing import TYPE_CHECKING -import async_timeout -from homeassistant import config_entries, core -from homeassistant.const import Platform from homeassistant.helpers.aiohttp_client import async_get_clientsession -from homeassistant.util import Throttle from .const import ( - DOMAIN_DATA, OFFICIAL_SITE_ROOT, + PLATFORMS, ) +from .coordinator import HealthchecksioDataUpdateCoordinator -MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=300) -PLATFORMS = [ - Platform.BINARY_SENSOR, -] +if TYPE_CHECKING: + from homeassistant import config_entries, core LOGGER = getLogger(__name__) @@ -33,72 +29,32 @@ async def async_setup_entry( config_entry: config_entries.ConfigEntry, ) -> bool: """Set up this integration using UI.""" - # Create DATA dict - if DOMAIN_DATA not in hass.data: - hass.data[DOMAIN_DATA] = {} - if "data" not in hass.data[DOMAIN_DATA]: - hass.data[DOMAIN_DATA] = {} - - # Get "global" configuration. - api_key = config_entry.data.get("api_key") - check = config_entry.data.get("check") - self_hosted = config_entry.data.get("self_hosted") - site_root = config_entry.data.get("site_root") - ping_endpoint = config_entry.data.get("ping_endpoint") + self_hosted = config_entry.data.get("self_hosted", False) + site_root = config_entry.data.get("site_root", OFFICIAL_SITE_ROOT) # Configure the client. - hass.data[DOMAIN_DATA]["client"] = HealthchecksioData( - hass, api_key, check, self_hosted, site_root, ping_endpoint + config_entry.runtime_data = coordinator = HealthchecksioDataUpdateCoordinator( + hass=hass, + api_key=config_entry.data["api_key"], + session=async_get_clientsession( + hass=hass, + verify_ssl=not self_hosted or site_root.startswith("https"), + ), + self_hosted=self_hosted, + check_id=config_entry.data["check"], + site_root=site_root, + ping_endpoint=config_entry.data.get("ping_endpoint"), ) + await coordinator.async_config_entry_first_refresh() + await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS) return True async def async_unload_entry( - hass: core.HomeAssistant, config_entry: config_entries.ConfigEntry + hass: core.HomeAssistant, + config_entry: config_entries.ConfigEntry, ) -> bool: """Unload a config entry.""" - if unload_ok := await hass.config_entries.async_unload_platforms( - config_entry, PLATFORMS - ): - hass.data.pop(DOMAIN_DATA, None) - LOGGER.info("Successfully removed the healthchecksio integration") - return unload_ok - - -class HealthchecksioData: - """This class handle communication and stores the data.""" - - def __init__(self, hass, api_key, check, self_hosted, site_root, ping_endpoint): - """Initialize the class.""" - self.hass = hass - self.api_key = api_key - self.check = check - self.self_hosted = self_hosted - self.site_root = site_root if self_hosted else OFFICIAL_SITE_ROOT - self.ping_endpoint = ping_endpoint - - @Throttle(MIN_TIME_BETWEEN_UPDATES) - async def update_data(self): - """Update data.""" - LOGGER.debug("Running update") - # This is where the main logic to update platform data goes. - try: - verify_ssl = not self.self_hosted or self.site_root.startswith("https") - session = async_get_clientsession(self.hass, verify_ssl) - headers = {"X-Api-Key": self.api_key} - async with async_timeout.timeout(10): - data = await session.get( - f"{self.site_root}/api/v1/checks/", headers=headers - ) - self.hass.data[DOMAIN_DATA]["data"] = await data.json() - - if self.self_hosted: - check_url = f"{self.site_root}/{self.ping_endpoint}/{self.check}" - else: - check_url = f"https://hc-ping.com/{self.check}" - await asyncio.sleep(1) # needed for self-hosted instances - await session.get(check_url) - except Exception: # pylint: disable=broad-except - LOGGER.exception("Could not update data") + return await hass.config_entries.async_unload_platforms(config_entry, PLATFORMS) diff --git a/custom_components/healthchecksio/binary_sensor.py b/custom_components/healthchecksio/binary_sensor.py index 12647df..957a92e 100644 --- a/custom_components/healthchecksio/binary_sensor.py +++ b/custom_components/healthchecksio/binary_sensor.py @@ -1,87 +1,87 @@ """Binary sensor platform for Healthchecksio.""" -try: - from homeassistant.components.binary_sensor import BinarySensorEntity -except ImportError: - from homeassistant.components.binary_sensor import ( - BinarySensorDevice as BinarySensorEntity, - ) +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from homeassistant.components.binary_sensor import ( + BinarySensorDeviceClass, + BinarySensorEntity, +) +from homeassistant.helpers.update_coordinator import CoordinatorEntity + +from custom_components.healthchecksio import LOGGER -from .const import ATTRIBUTION, BINARY_SENSOR_DEVICE_CLASS, DOMAIN, DOMAIN_DATA +from .const import ATTRIBUTION, DOMAIN + +if TYPE_CHECKING: + from homeassistant.core import HomeAssistant + + from .coordinator import HealthchecksioDataUpdateCoordinator async def async_setup_entry(hass, config_entry, async_add_devices): """Setup sensor platform.""" # Send update "signal" to the component - await hass.data[DOMAIN_DATA]["client"].update_data() - checks = [] - for check in hass.data[DOMAIN_DATA].get("data", {}).get("checks", []): - check_data = { - "name": check.get("name"), - "last_ping": check.get("last_ping"), - "status": check.get("status"), - "ping_url": check.get("ping_url"), - } - checks.append(HealthchecksioBinarySensor(hass, check_data, config_entry)) - async_add_devices(checks, True) + coordinator: HealthchecksioDataUpdateCoordinator = config_entry.runtime_data + async_add_devices( + HealthchecksioBinarySensor( + hass=hass, + ping_url=check.get("ping_url"), + coordinator=coordinator, + ) + for check in coordinator.data.get("checks", []) + ) -class HealthchecksioBinarySensor(BinarySensorEntity): +class HealthchecksioBinarySensor(BinarySensorEntity, CoordinatorEntity): """Healthchecksio binary_sensor class.""" - def __init__(self, hass, check_data, config_entry): - self.hass = hass - self.attr = {} - self.config_entry = config_entry - self._status = None - self.check_data = check_data - self.check = {} - - async def async_update(self): - """Update the binary_sensor.""" - # Send update "signal" to the component - await self.hass.data[DOMAIN_DATA]["client"].update_data() - - # Check the data and update the value. - for check in self.hass.data[DOMAIN_DATA].get("data", {}).get("checks", []): - if self.unique_id == check.get("ping_url").split("/")[-1]: - self.check = check - break - self._status = self.check.get("status") != "down" - - # Set/update attributes - self.attr["attribution"] = ATTRIBUTION - self.attr["last_ping"] = self.check.get("last_ping") + coordinator: HealthchecksioDataUpdateCoordinator - @property - def unique_id(self): - """Return a unique ID to use for this binary_sensor.""" - return self.check_data.get("ping_url", "").split("/")[-1] + _attr_attribution = ATTRIBUTION + _attr_device_class = BinarySensorDeviceClass.CONNECTIVITY - @property - def device_info(self): - return { - "identifiers": {(DOMAIN, self.config_entry.entry_id)}, + def __init__( + self, + *, + hass: HomeAssistant, + coordinator: HealthchecksioDataUpdateCoordinator, + ping_url: str, + ): + super().__init__(coordinator) + self.hass = hass + self._ping_url = ping_url + + self._attr_unique_id = ping_url.split("/")[-1] + self._attr_device_info = { + "identifiers": {(DOMAIN, self.coordinator.config_entry.entry_id)}, "name": "Healthchecks.io", "manufacturer": "SIA Monkey See Monkey Do", } + def get_check(self) -> dict[str, Any]: + """Get check data.""" + for check in self.coordinator.data.get("checks", []): + if self._ping_url == check.get("ping_url"): + return check + return {} + @property def name(self): """Return the name of the binary_sensor.""" - return self.check.get("name") - - @property - def device_class(self): - """Return the class of this binary_sensor.""" - return BINARY_SENSOR_DEVICE_CLASS + check = self.get_check() + return check.get("name") @property def is_on(self): """Return true if the binary_sensor is on.""" - return self._status + check = self.get_check() + LOGGER.debug("Check: %s", check) + return check.get("status") != "down" @property def extra_state_attributes(self): """Return the state attributes.""" - return self.attr + check = self.get_check() + return {"last_ping": check.get("last_ping")} diff --git a/custom_components/healthchecksio/config_flow.py b/custom_components/healthchecksio/config_flow.py index 381b22b..9901d96 100644 --- a/custom_components/healthchecksio/config_flow.py +++ b/custom_components/healthchecksio/config_flow.py @@ -1,5 +1,7 @@ """Adds config flow for Blueprint.""" +from __future__ import annotations + import asyncio from collections import OrderedDict from logging import getLogger @@ -9,7 +11,7 @@ from homeassistant import config_entries from homeassistant.helpers.aiohttp_client import async_get_clientsession -from .const import DOMAIN, DOMAIN_DATA, OFFICIAL_SITE_ROOT +from .const import DOMAIN, OFFICIAL_SITE_ROOT LOGGER = getLogger(__name__) @@ -129,8 +131,7 @@ async def _test_credentials( headers = {"X-Api-Key": api_key} async with async_timeout.timeout(10): LOGGER.info("Checking API Key") - data = await session.get(f"{site_root}/api/v1/checks/", headers=headers) - self.hass.data[DOMAIN_DATA] = {"data": await data.json()} + await session.get(f"{site_root}/api/v1/checks/", headers=headers) LOGGER.info("Checking Check ID") if self_hosted: diff --git a/custom_components/healthchecksio/const.py b/custom_components/healthchecksio/const.py index 815fbb3..7f8e2e8 100644 --- a/custom_components/healthchecksio/const.py +++ b/custom_components/healthchecksio/const.py @@ -1,9 +1,17 @@ """Constants for blueprint.""" -# Base component constants +from __future__ import annotations + +from datetime import timedelta + +from homeassistant.const import Platform + DOMAIN = "healthchecksio" -DOMAIN_DATA = f"{DOMAIN}_data" INTEGRATION_VERSION = "main" +MIN_TIME_BETWEEN_UPDATES = timedelta(seconds=300) +PLATFORMS = [ + Platform.BINARY_SENSOR, +] ISSUE_URL = "https://github.com/custom-components/healthchecksio/issues" ATTRIBUTION = "Data from this is provided by healthchecks.io." diff --git a/custom_components/healthchecksio/coordinator.py b/custom_components/healthchecksio/coordinator.py new file mode 100644 index 0000000..247f755 --- /dev/null +++ b/custom_components/healthchecksio/coordinator.py @@ -0,0 +1,78 @@ +"""Add coordinator for Healthchecks.io integration.""" + +from __future__ import annotations + +import asyncio +from logging import getLogger +from typing import TYPE_CHECKING, Any + +from aiohttp import ClientSession, ClientTimeout +from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed + +from .const import DOMAIN, MIN_TIME_BETWEEN_UPDATES + +if TYPE_CHECKING: + from homeassistant.config_entries import ConfigEntry + from homeassistant.core import HomeAssistant + +LOGGER = getLogger(__name__) + + +class HealthchecksioDataUpdateCoordinator(DataUpdateCoordinator): + """Class to manage fetching Healthchecks.io data.""" + + config_entry: ConfigEntry + + def __init__( + self, + hass: HomeAssistant, + *, + api_key: str, + session: ClientSession, + self_hosted: bool = False, + check_id: str | None = None, + site_root: str | None = None, + ping_endpoint: str | None = None, + ): + """Initialize.""" + super().__init__( + hass=hass, + logger=LOGGER, + name=DOMAIN, + update_interval=MIN_TIME_BETWEEN_UPDATES, + ) + self._api_key = api_key + self._site_root = site_root + self._session = session + self._self_hosted = self_hosted + self._check_id = check_id + self._ping_endpoint = ping_endpoint + + async def _async_update_data(self) -> dict[str, Any]: + """Update data.""" + try: + data = await self._session.get( + f"{self._site_root}/api/v1/checks/", + headers={"X-Api-Key": self._api_key}, + timeout=ClientTimeout(total=10), + ) + except Exception as error: + raise UpdateFailed(error) from error + + check_url = ( + f"https://hc-ping.com/{self._check_id}" + if not self._self_hosted + else f"{self._site_root}/{self._ping_endpoint}/{self._check_id}" + ) + if self._self_hosted: + await asyncio.sleep(1) # needed for self-hosted instances + + try: + await self._session.get( + check_url, + timeout=ClientTimeout(total=10), + ) + except Exception: + LOGGER.exception("Could not ping") + + return await data.json()