Skip to content

Improve IPv64 update performance and validation #411

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Ludy87 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Improve IPv64 update performance and validation #411

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion custom_components/ipv64/config_flow.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
GET_ACCOUNT_INFO_URL,
GET_DOMAIN_URL,
TIMEOUT,
is_allowed_domain,
)

_LOGGER = logging.getLogger(__name__)
Expand Down Expand Up @@ -123,7 +124,7 @@ async def check_domain_login(hass: core.HomeAssistant, data: dict[str, Any]) ->

# Validate domain against allowed domains
input_domain = data[CONF_DOMAIN]
if not any(input_domain.endswith(allowed_domain) for allowed_domain in ALLOWED_DOMAINS):
if not is_allowed_domain(input_domain):
_LOGGER.error("Domain %s is not one of the allowed domains: %s", input_domain, ALLOWED_DOMAINS)
raise InvalidDomain(f"Domain {input_domain} is not allowed. Allowed domains: {', '.join(ALLOWED_DOMAINS)}")

Expand Down
9 changes: 9 additions & 0 deletions custom_components/ipv64/const.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,3 +95,12 @@
"root64.de",
"route64.de",
]


def is_allowed_domain(domain: str) -> bool:
"""Return whether a domain belongs to an IPv64 allowed zone."""
normalized_domain = domain.strip().lower().rstrip(".")
return any(
normalized_domain == allowed_domain or normalized_domain.endswith(f".{allowed_domain}")
for allowed_domain in ALLOWED_DOMAINS
)
79 changes: 50 additions & 29 deletions custom_components/ipv64/coordinator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,7 @@
from __future__ import annotations

import asyncio
from datetime import datetime, timedelta
from datetime import UTC, datetime, timedelta
import logging
from typing import Any

Expand Down Expand Up @@ -33,20 +33,21 @@
RETRY_DELAY,
TIMEOUT,
UPDATE_URL,
is_allowed_domain,
)

_LOGGER = logging.getLogger(__name__)


async def get_domain(session: aiohttp.ClientSession, headers: dict[str, str], data: dict[str, Any]) -> None:
async def get_domain(session: aiohttp.ClientSession, headers: dict[str, str], data: dict[str, Any]) -> dict[str, Any]:

Check failure on line 42 in custom_components/ipv64/coordinator.py

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Refactor this function to reduce its Cognitive Complexity from 29 to the 15 allowed. 


See more on https://sonarcloud.io/project/issues?id=Ludy87_ipv64&issues=AZ74qJZynwDRD_WSW7Sl&open=AZ74qJZynwDRD_WSW7Sl&pullRequest=411
"""Fetch domain information from the IPv64.net API."""
config_domain = data.get(CONF_DOMAIN, "")
# Validate domain against allowed domains
if not any(config_domain.endswith(allowed_domain) for allowed_domain in ALLOWED_DOMAINS):
if not is_allowed_domain(config_domain):
_LOGGER.error("Domain %s is not one of the allowed domains: %s", config_domain, ALLOWED_DOMAINS)
data["subdomains"] = []
data["error"] = f"Domain {config_domain} not allowed"
return
return data

for attempt in range(RETRY_ATTEMPTS):
try:
Expand All @@ -58,7 +59,7 @@
_LOGGER.warning("No subdomains found for account")
data["subdomains"] = []
data["error"] = "No subdomains available"
return
return data

sub_domains_list = []
domain_found = False
Expand Down Expand Up @@ -97,9 +98,9 @@
_LOGGER.error("Configured domain %s not found in subdomains", config_domain)
data["subdomains"] = []
data["error"] = f"Domain {config_domain} not found"
return
return data
data["subdomains"] = sub_domains_list
return
return data
except aiohttp.ClientResponseError as error:
if attempt == RETRY_ATTEMPTS - 1:
_LOGGER.error(
Expand All @@ -110,7 +111,7 @@
)
data["subdomains"] = []
data["error"] = "Failed to fetch domains"
return
return data
_LOGGER.warning(
"Failed to fetch domains, retrying (%d/%d): %s",
attempt + 1,
Expand All @@ -123,7 +124,7 @@
_LOGGER.error("Failed to fetch domains after %d attempts: %s", RETRY_ATTEMPTS, err)
data["subdomains"] = []
data["error"] = str(err)
return
return data
_LOGGER.warning(
"Failed to fetch domains, retrying (%d/%d): %s",
attempt + 1,
Expand All @@ -132,10 +133,11 @@
)
await asyncio.sleep(RETRY_DELAY)

return data

async def add_domain(hass: HomeAssistant, coordinator: DataUpdateCoordinator, domain: str, api_key: str) -> None:
"""Add a new domain via the IPv64.net API."""
if not any(domain.endswith(allowed_domain) for allowed_domain in ALLOWED_DOMAINS):
if not is_allowed_domain(domain):
_LOGGER.error("Domain %s is not one of the allowed domains: %s", domain, ALLOWED_DOMAINS)
raise ValueError(f"Domain {domain} not allowed")

Expand Down Expand Up @@ -191,7 +193,7 @@

async def delete_domain(hass: HomeAssistant, coordinator: DataUpdateCoordinator, domain: str, api_key: str) -> None:
"""Delete a domain via the IPv64.net API."""
if not any(domain.endswith(allowed_domain) for allowed_domain in ALLOWED_DOMAINS):
if not is_allowed_domain(domain):
_LOGGER.error("Domain %s is not one of the allowed domains: %s", domain, ALLOWED_DOMAINS)
raise ValueError(f"Domain {domain} not allowed")

Expand Down Expand Up @@ -281,18 +283,24 @@
_LOGGER.debug("self.data was invalid, reinitializing")
self.data = {CONF_DOMAIN: self.config_entry.data.get(CONF_DOMAIN, "")}

await self._async_restore_cache()

session = async_get_clientsession(self.hass)
headers_api = {"Authorization": f"Bearer {self.config_entry.data.get(CONF_API_KEY, '')}"}

try:
for attempt in range(RETRY_ATTEMPTS):
try:
account_info = await get_account_info(session, headers_api, self.config_entry.data)
account_info, domain_info = await asyncio.gather(
get_account_info(session, headers_api, self.config_entry.data),
get_domain(session, headers_api, {CONF_DOMAIN: self.config_entry.data.get(CONF_DOMAIN, "")}),
)
Comment on lines +294 to +297

@chatgpt-codex-connector chatgpt-codex-connector Bot Jun 24, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Avoid issuing domain lookups on failed account retries

When the account-info request fails or is temporarily rate-limited, this gather has already started the get_domains request, and the outer retry loop then starts both calls again. Previously the domain lookup only ran after account info succeeded, so in a 429/5xx/timeout scenario on get_account_info each retry now burns an extra IPv64 API request (and get_domain has its own retries), which can quickly exceed the integration’s 3-requests/10s rate-limit handling and make transient failures worse. Fetch the domain data only after account info succeeds, or explicitly cancel/avoid the second task on retry.

Useful? React with 👍 / 👎.

if not isinstance(account_info, dict):
_LOGGER.error("Received invalid account_info: %s", account_info)
raise UpdateFailed("Received invalid account info")
_LOGGER.debug("Received account info: %s", account_info)
_LOGGER.debug("Received account and domain info for %s", self.config_entry.data.get(CONF_DOMAIN))
self.data.update(account_info)
self.data.update(domain_info)
async_dismiss(
self.hass,
notification_id=f"{DOMAIN}_{self.config_entry.entry_id}_api_error",
Expand Down Expand Up @@ -344,7 +352,8 @@
)
raise UpdateFailed(f"Unexpected error: {err}") from err

await get_domain(session, headers_api, self.data)
if self.data.get("error"):
_LOGGER.warning("Domain data update completed with warning: %s", self.data["error"])

ip_is_changed = False
if self.config_entry.options.get(CONF_API_ECONOMY, True) or is_economy:
Expand All @@ -365,19 +374,19 @@
update_result = await resp.json()
self.data.update({"update_result": update_result.get("status", "unknown")})
_LOGGER.info("IP update successful for %s: %s", self.config_entry.data.get(CONF_DOMAIN), update_result)
async_dismiss(
self.hass,
notification_id=f"{DOMAIN}_{self.config_entry.entry_id}_limit_error",
)
async_dismiss(
self.hass,
notification_id=f"{DOMAIN}_{self.config_entry.entry_id}_auth_error",
)
async_dismiss(
self.hass,
notification_id=f"{DOMAIN}_{self.config_entry.entry_id}_network_update_error",
)
break
async_dismiss(
self.hass,
notification_id=f"{DOMAIN}_{self.config_entry.entry_id}_limit_error",
)
async_dismiss(
self.hass,
notification_id=f"{DOMAIN}_{self.config_entry.entry_id}_auth_error",
)
async_dismiss(
self.hass,
notification_id=f"{DOMAIN}_{self.config_entry.entry_id}_network_update_error",
)
except aiohttp.ClientResponseError as error:
self.data.update({"update_result": "fail"})
if attempt == RETRY_ATTEMPTS - 1:
Expand Down Expand Up @@ -436,8 +445,8 @@

updates_used = self.data.get(CONF_DYNDNS_UPDATES, 0)
updates_limit = self.data.get(CONF_DAILY_UPDATE_LIMIT, 64)
if updates_used > 0 and updates_limit > 0:
remaining_updates = updates_limit - int(updates_used)
if isinstance(updates_used, int | float) and isinstance(updates_limit, int | float) and updates_limit > 0:
remaining_updates = max(0, int(updates_limit) - int(updates_used))
self.data.update({CONF_REMAINING_UPDATES: remaining_updates})
if updates_used >= updates_limit * 0.9:
async_create(
Expand All @@ -452,11 +461,23 @@
notification_id=f"{DOMAIN}_{self.config_entry.entry_id}_update_limit",
)

self.data["cache_time"] = datetime.now().isoformat()
self.data["cache_time"] = datetime.now(UTC).isoformat()
await self._cache.async_save(self.data)

return self.data

async def _async_restore_cache(self) -> None:
"""Restore cached coordinator data for faster startup and offline availability."""
if self.data.get("cache_time"):
return
cached_data = await self._cache.async_load()
if not isinstance(cached_data, dict):
return
if cached_data.get(CONF_DOMAIN) != self.config_entry.data.get(CONF_DOMAIN):
return
self.data.update(cached_data)
_LOGGER.debug("Restored cached IPv64 data for %s", self.config_entry.data.get(CONF_DOMAIN))

async def check_ip_equal(self, session: aiohttp.ClientSession) -> bool:
"""Check if the IP has changed."""
_LOGGER.debug("Checking IP in economy mode for %s", self.config_entry.data.get(CONF_DOMAIN))
Expand Down
Loading