Skip to content

fix: throttle concurrent API calls and 429 backoff #477

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
dennismdejong wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix: throttle concurrent API calls and 429 backoff #477

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
8 changes: 6 additions & 2 deletions custom_components/panasonic_cc/panasonic/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
)
from .const import DATA_COORDINATORS, ENERGY_COORDINATORS
from .coordinator import (
MAX_CONCURRENT_API_CALLS,
PanasonicDeviceCoordinator,
PanasonicDeviceEnergyCoordinator,
)
Expand Down Expand Up @@ -63,18 +64,21 @@ async def async_setup_panasonic(
devices = api.get_devices()
_LOGGER.info("Got %s Panasonic devices", len(devices))

api_semaphore = asyncio.Semaphore(MAX_CONCURRENT_API_CALLS)
hass.data[DOMAIN]["_api_semaphore"] = api_semaphore

data_coordinators: list[PanasonicDeviceCoordinator] = []
energy_coordinators: list[PanasonicDeviceEnergyCoordinator] = []

# Create all device coordinators first
device_coordinators_uninitialized: list[tuple[PanasonicDeviceCoordinator, PanasonicDeviceInfo]] = []
for device in devices:
try:
device_coordinator = PanasonicDeviceCoordinator(hass, config, api, device)
device_coordinator = PanasonicDeviceCoordinator(hass, config, api, device, api_semaphore)
device_coordinators_uninitialized.append((device_coordinator, device))
if enable_daily_energy_sensor:
energy_coordinators.append(
PanasonicDeviceEnergyCoordinator(hass, config, api, device)
PanasonicDeviceEnergyCoordinator(hass, config, api, device, api_semaphore)
)
except Exception as exc:
_LOGGER.warning("Failed to create coordinator for device %s: %s", device.name, exc, exc_info=True)
Expand Down
43 changes: 36 additions & 7 deletions custom_components/panasonic_cc/panasonic/coordinator.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
"""Coordinators for Panasonic Comfort Cloud devices."""
import asyncio
import logging
import random
from datetime import timedelta

from aiohttp import ClientResponseError
Expand Down Expand Up @@ -32,6 +33,9 @@
MAX_CONSECUTIVE_FAILURES = 5
BACKOFF_MULTIPLIER = 2
MAX_UPDATE_INTERVAL = 600 # seconds
MAX_CONCURRENT_API_CALLS = 3
MIN_RATE_LIMIT_INTERVAL = 30
BACKOFF_MULTIPLIER_RATE_LIMIT = 4

_LOGGER = logging.getLogger(__name__)

Expand Down Expand Up @@ -63,6 +67,7 @@ def __init__(
config: dict,
api_client: ApiClient,
device_info: PanasonicDeviceInfo,
api_semaphore: asyncio.Semaphore,
) -> None:
"""Initialize the coordinator."""
self._base_interval = config.get(
Expand All @@ -84,6 +89,7 @@ def __init__(
self._auth_failed = False
self._last_error: FriendlyError | None = None
self._last_command_error: FriendlyError | None = None
self._api_semaphore = api_semaphore

@property
def last_error(self) -> FriendlyError | None:
Expand Down Expand Up @@ -141,7 +147,8 @@ def get_change_request_builder(self) -> ChangeRequestBuilder:
async def async_apply_changes(self, request_builder: ChangeRequestBuilder) -> None:
"""Apply changes to the device."""
try:
await self._api_client.set_device_raw(self.device, request_builder.build())
async with self._api_semaphore:
await self._api_client.set_device_raw(self.device, request_builder.build())
# Clear command error on success
self._last_command_error = None
except Exception as err:
Expand Down Expand Up @@ -193,7 +200,9 @@ async def _async_update_data(self) -> int:

try:
if self._device is None:
self._device = await self._api_client.get_device(self._device_info)
async with self._api_semaphore:
await asyncio.sleep(random.uniform(0.0, 0.5))
self._device = await self._api_client.get_device(self._device_info)
_LOGGER.debug(
"%s Device features - Nanoe: %s, Eco Navi: %s, AI Eco: %s",
self._device_info.name,
Expand All @@ -204,7 +213,10 @@ async def _async_update_data(self) -> int:
self._update_id = 1
self._reset_backoff()
return self._update_id
if await self._api_client.try_update_device(self._device):
async with self._api_semaphore:
await asyncio.sleep(random.uniform(0.0, 0.5))
updated = await self._api_client.try_update_device(self._device)
if updated:
self._update_id += 1
self._reset_backoff()
return self._update_id
Expand Down Expand Up @@ -238,10 +250,15 @@ def _reset_backoff(self) -> None:
def _handle_failure(self, err: Exception | None = None) -> None:
"""Handle API failure with exponential backoff."""
self._consecutive_failures += 1
multiplier = BACKOFF_MULTIPLIER
if err is not None and classify_error(err).category == ErrorCategory.RATE_LIMIT:
multiplier = BACKOFF_MULTIPLIER_RATE_LIMIT
new_interval = min(
self._base_interval * (BACKOFF_MULTIPLIER ** self._consecutive_failures),
self._base_interval * (multiplier ** self._consecutive_failures),
MAX_UPDATE_INTERVAL,
)
if err is not None and classify_error(err).category == ErrorCategory.RATE_LIMIT:
new_interval = max(new_interval, MIN_RATE_LIMIT_INTERVAL)
self.update_interval = timedelta(seconds=new_interval)
if err is not None:
self._last_error = classify_error(err)
Expand Down Expand Up @@ -273,6 +290,7 @@ def __init__(
config: dict,
api_client: ApiClient,
device_info: PanasonicDeviceInfo,
api_semaphore: asyncio.Semaphore,
) -> None:
"""Initialize the coordinator."""
self._base_interval = config.get(
Expand All @@ -291,6 +309,7 @@ def __init__(
self._consecutive_failures = 0
self._auth_failed = False
self._last_error: FriendlyError | None = None
self._api_semaphore = api_semaphore

@property
def last_error(self) -> FriendlyError | None:
Expand Down Expand Up @@ -341,11 +360,16 @@ async def _async_update_data(self) -> int:

try:
if self._energy is None:
self._energy = await self._api_client.async_get_energy(self._device_info)
async with self._api_semaphore:
await asyncio.sleep(random.uniform(0.0, 0.5))
self._energy = await self._api_client.async_get_energy(self._device_info)
self._update_id = 1
self._reset_backoff()
return self._update_id
if await self._api_client.async_try_update_energy(self._energy):
async with self._api_semaphore:
await asyncio.sleep(random.uniform(0.0, 0.5))
updated = await self._api_client.async_try_update_energy(self._energy)
if updated:
self._update_id += 1
self._reset_backoff()
return self._update_id
Expand Down Expand Up @@ -379,10 +403,15 @@ def _reset_backoff(self) -> None:
def _handle_failure(self, err: Exception | None = None) -> None:
"""Handle API failure with exponential backoff."""
self._consecutive_failures += 1
multiplier = BACKOFF_MULTIPLIER
if err is not None and classify_error(err).category == ErrorCategory.RATE_LIMIT:
multiplier = BACKOFF_MULTIPLIER_RATE_LIMIT
new_interval = min(
self._base_interval * (BACKOFF_MULTIPLIER ** self._consecutive_failures),
self._base_interval * (multiplier ** self._consecutive_failures),
MAX_UPDATE_INTERVAL,
)
if err is not None and classify_error(err).category == ErrorCategory.RATE_LIMIT:
new_interval = max(new_interval, MIN_RATE_LIMIT_INTERVAL)
self.update_interval = timedelta(seconds=new_interval)
if err is not None:
self._last_error = classify_error(err)
Expand Down