evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Move NextDNS data update coordinators to the coordinator module (#115919)

Browse source 
Co-authored-by: Maciej Bieniek <478555+bieniu@users.noreply.github.com>
This commit is contained in:
Maciej Bieniek 2024-04-22 11:19:35 +02:00 • committed by GitHub
parent 09ae8b9f52
commit 354e8e92f3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 139 additions and 119 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

126
homeassistant/components/nextdns/__init__.py
View file

@ -4,31 +4,15 @@ from __future__ import annotations
import asyncio
from datetime import timedelta
import logging
from typing import TypeVar
from aiohttp.client_exceptions import ClientConnectorError
from nextdns import (
AnalyticsDnssec,
AnalyticsEncryption,
AnalyticsIpVersions,
AnalyticsProtocols,
AnalyticsStatus,
ApiError,
ConnectionStatus,
InvalidApiKeyError,
NextDns,
Settings,
)
from nextdns.model import NextDnsData
from nextdns import ApiError, NextDns
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_API_KEY, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import (
ATTR_CONNECTION,
@ -44,104 +28,16 @@ from .const import (
UPDATE_INTERVAL_CONNECTION,
UPDATE_INTERVAL_SETTINGS,
)
CoordinatorDataT = TypeVar("CoordinatorDataT", bound=NextDnsData)
class NextDnsUpdateCoordinator(DataUpdateCoordinator[CoordinatorDataT]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS data API."""
def __init__(
self,
hass: HomeAssistant,
nextdns: NextDns,
profile_id: str,
update_interval: timedelta,
) -> None:
"""Initialize."""
self.nextdns = nextdns
self.profile_id = profile_id
self.profile_name = nextdns.get_profile_name(profile_id)
self.device_info = DeviceInfo(
configuration_url=f"https://my.nextdns.io/{profile_id}/setup",
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, str(profile_id))},
manufacturer="NextDNS Inc.",
name=self.profile_name,
)
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
async def _async_update_data(self) -> CoordinatorDataT:
"""Update data via internal method."""
try:
async with asyncio.timeout(10):
return await self._async_update_data_internal()
except (ApiError, ClientConnectorError, InvalidApiKeyError) as err:
raise UpdateFailed(err) from err
async def _async_update_data_internal(self) -> CoordinatorDataT:
"""Update data via library."""
raise NotImplementedError("Update method not implemented")
class NextDnsStatusUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics status data from API."""
async def _async_update_data_internal(self) -> AnalyticsStatus:
"""Update data via library."""
return await self.nextdns.get_analytics_status(self.profile_id)
class NextDnsDnssecUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsDnssec]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics Dnssec data from API."""
async def _async_update_data_internal(self) -> AnalyticsDnssec:
"""Update data via library."""
return await self.nextdns.get_analytics_dnssec(self.profile_id)
class NextDnsEncryptionUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsEncryption]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics encryption data from API."""
async def _async_update_data_internal(self) -> AnalyticsEncryption:
"""Update data via library."""
return await self.nextdns.get_analytics_encryption(self.profile_id)
class NextDnsIpVersionsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsIpVersions]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics IP versions data from API."""
async def _async_update_data_internal(self) -> AnalyticsIpVersions:
"""Update data via library."""
return await self.nextdns.get_analytics_ip_versions(self.profile_id)
class NextDnsProtocolsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsProtocols]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS analytics protocols data from API."""
async def _async_update_data_internal(self) -> AnalyticsProtocols:
"""Update data via library."""
return await self.nextdns.get_analytics_protocols(self.profile_id)
class NextDnsSettingsUpdateCoordinator(NextDnsUpdateCoordinator[Settings]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS connection data from API."""
async def _async_update_data_internal(self) -> Settings:
"""Update data via library."""
return await self.nextdns.get_settings(self.profile_id)
class NextDnsConnectionUpdateCoordinator(NextDnsUpdateCoordinator[ConnectionStatus]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching NextDNS connection data from API."""
async def _async_update_data_internal(self) -> ConnectionStatus:
"""Update data via library."""
return await self.nextdns.connection_status(self.profile_id)
_LOGGER = logging.getLogger(__name__)
from .coordinator import (
NextDnsConnectionUpdateCoordinator,
NextDnsDnssecUpdateCoordinator,
NextDnsEncryptionUpdateCoordinator,
NextDnsIpVersionsUpdateCoordinator,
NextDnsProtocolsUpdateCoordinator,
NextDnsSettingsUpdateCoordinator,
NextDnsStatusUpdateCoordinator,
NextDnsUpdateCoordinator,
)
PLATFORMS = [Platform.BINARY_SENSOR, Platform.BUTTON, Platform.SENSOR, Platform.SWITCH]
COORDINATORS: list[tuple[str, type[NextDnsUpdateCoordinator], timedelta]] = [

2
homeassistant/components/nextdns/binary_sensor.py
View file

@ -19,8 +19,8 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import CoordinatorDataT, NextDnsConnectionUpdateCoordinator
from .const import ATTR_CONNECTION, DOMAIN
from .coordinator import CoordinatorDataT, NextDnsConnectionUpdateCoordinator
PARALLEL_UPDATES = 1

2
homeassistant/components/nextdns/button.py
View file

@ -9,8 +9,8 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import NextDnsStatusUpdateCoordinator
from .const import ATTR_STATUS, DOMAIN
from .coordinator import NextDnsStatusUpdateCoordinator
PARALLEL_UPDATES = 1

124
homeassistant/components/nextdns/coordinator.py Normal file
View file

@ -0,0 +1,124 @@
"""NextDns coordinator."""
import asyncio
from datetime import timedelta
import logging
from typing import TypeVar
from aiohttp.client_exceptions import ClientConnectorError
from nextdns import (
AnalyticsDnssec,
AnalyticsEncryption,
AnalyticsIpVersions,
AnalyticsProtocols,
AnalyticsStatus,
ApiError,
ConnectionStatus,
InvalidApiKeyError,
NextDns,
Settings,
)
from nextdns.model import NextDnsData
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN
_LOGGER = logging.getLogger(__name__)
CoordinatorDataT = TypeVar("CoordinatorDataT", bound=NextDnsData)
class NextDnsUpdateCoordinator(DataUpdateCoordinator[CoordinatorDataT]):
"""Class to manage fetching NextDNS data API."""
def __init__(
self,
hass: HomeAssistant,
nextdns: NextDns,
profile_id: str,
update_interval: timedelta,
) -> None:
"""Initialize."""
self.nextdns = nextdns
self.profile_id = profile_id
self.profile_name = nextdns.get_profile_name(profile_id)
self.device_info = DeviceInfo(
configuration_url=f"https://my.nextdns.io/{profile_id}/setup",
entry_type=DeviceEntryType.SERVICE,
identifiers={(DOMAIN, str(profile_id))},
manufacturer="NextDNS Inc.",
name=self.profile_name,
)
super().__init__(hass, _LOGGER, name=DOMAIN, update_interval=update_interval)
async def _async_update_data(self) -> CoordinatorDataT:
"""Update data via internal method."""
try:
async with asyncio.timeout(10):
return await self._async_update_data_internal()
except (ApiError, ClientConnectorError, InvalidApiKeyError) as err:
raise UpdateFailed(err) from err
async def _async_update_data_internal(self) -> CoordinatorDataT:
"""Update data via library."""
raise NotImplementedError("Update method not implemented")
class NextDnsStatusUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsStatus]):
"""Class to manage fetching NextDNS analytics status data from API."""
async def _async_update_data_internal(self) -> AnalyticsStatus:
"""Update data via library."""
return await self.nextdns.get_analytics_status(self.profile_id)
class NextDnsDnssecUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsDnssec]):
"""Class to manage fetching NextDNS analytics Dnssec data from API."""
async def _async_update_data_internal(self) -> AnalyticsDnssec:
"""Update data via library."""
return await self.nextdns.get_analytics_dnssec(self.profile_id)
class NextDnsEncryptionUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsEncryption]):
"""Class to manage fetching NextDNS analytics encryption data from API."""
async def _async_update_data_internal(self) -> AnalyticsEncryption:
"""Update data via library."""
return await self.nextdns.get_analytics_encryption(self.profile_id)
class NextDnsIpVersionsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsIpVersions]):
"""Class to manage fetching NextDNS analytics IP versions data from API."""
async def _async_update_data_internal(self) -> AnalyticsIpVersions:
"""Update data via library."""
return await self.nextdns.get_analytics_ip_versions(self.profile_id)
class NextDnsProtocolsUpdateCoordinator(NextDnsUpdateCoordinator[AnalyticsProtocols]):
"""Class to manage fetching NextDNS analytics protocols data from API."""
async def _async_update_data_internal(self) -> AnalyticsProtocols:
"""Update data via library."""
return await self.nextdns.get_analytics_protocols(self.profile_id)
class NextDnsSettingsUpdateCoordinator(NextDnsUpdateCoordinator[Settings]):
"""Class to manage fetching NextDNS connection data from API."""
async def _async_update_data_internal(self) -> Settings:
"""Update data via library."""
return await self.nextdns.get_settings(self.profile_id)
class NextDnsConnectionUpdateCoordinator(NextDnsUpdateCoordinator[ConnectionStatus]):
"""Class to manage fetching NextDNS connection data from API."""
async def _async_update_data_internal(self) -> ConnectionStatus:
"""Update data via library."""
return await self.nextdns.connection_status(self.profile_id)

2
homeassistant/components/nextdns/sensor.py
View file

@ -26,7 +26,6 @@ from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import CoordinatorDataT, NextDnsUpdateCoordinator
from .const import (
ATTR_DNSSEC,
ATTR_ENCRYPTION,
@ -35,6 +34,7 @@ from .const import (
ATTR_STATUS,
DOMAIN,
)
from .coordinator import CoordinatorDataT, NextDnsUpdateCoordinator
PARALLEL_UPDATES = 1

2
homeassistant/components/nextdns/switch.py
View file

@ -18,8 +18,8 @@ from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import CoordinatorDataT, NextDnsSettingsUpdateCoordinator
from .const import ATTR_SETTINGS, DOMAIN
from .coordinator import CoordinatorDataT, NextDnsSettingsUpdateCoordinator
PARALLEL_UPDATES = 1