coordinator.py

Go to the documentation of this file.

"""DataUpdateCoordinator for the Bring! integration."""
 
from __future__ import annotations
 
from datetime import timedelta
import logging
 
from bring_api import (
    Bring,
    BringAuthException,
    BringParseException,
    BringRequestException,
)
from bring_api.types import BringItemsResponse, BringList, BringUserSettingsResponse
 
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_EMAIL
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
from .const import DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
 
class BringData(BringList, BringItemsResponse):
    """Coordinator data class."""
 
 
class BringDataUpdateCoordinator(DataUpdateCoordinator[dict[str, BringData]]):
    """A Bring Data Update Coordinator."""
 
    config_entry: ConfigEntry
    user_settings: BringUserSettingsResponse
 
    def __init__(self, hass: HomeAssistant, bring: Bring) -> None:
        """Initialize the Bring data coordinator."""
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=90),
        )
        self.bringbring = bring
 
    async def _async_update_data(self) -> dict[str, BringData]:
        try:
            lists_response = await self.bringbring.load_lists()
        except BringRequestException as e:
            raise UpdateFailed("Unable to connect and retrieve data from bring") from e
        except BringParseException as e:
            raise UpdateFailed("Unable to parse response from bring") from e
        except BringAuthException as e:
            # try to recover by refreshing access token, otherwise
            # initiate reauth flow
            try:
                await self.bringbring.retrieve_new_access_token()
            except (BringRequestException, BringParseException) as exc:
                raise UpdateFailed("Refreshing authentication token failed") from exc
            except BringAuthException as exc:
                raise ConfigEntryAuthFailed(
                    translation_domain=DOMAIN,
                    translation_key="setup_authentication_exception",
                    translation_placeholders={CONF_EMAIL: self.bringbring.mail},
                ) from exc
            raise UpdateFailed(
                "Authentication failed but re-authentication was successful, trying again later"
            ) from e
 
        list_dict: dict[str, BringData] = {}
        for lst in lists_response["lists"]:
            try:
                items = await self.bringbring.get_list(lst["listUuid"])
            except BringRequestException as e:
                raise UpdateFailed(
                    "Unable to connect and retrieve data from bring"
                ) from e
            except BringParseException as e:
                raise UpdateFailed("Unable to parse response from bring") from e
            else:
                list_dict[lst["listUuid"]] = BringData(**lst, **items)
 
        return list_dict
 
    async def _async_setup(self) -> None:
        """Set up coordinator."""
 
        await self.async_refresh_user_settingsasync_refresh_user_settings()
 
    async def async_refresh_user_settings(self) -> None:
        """Refresh user settings."""
        try:
            self.user_settingsuser_settings = await self.bringbring.get_all_user_settings()
        except (BringAuthException, BringRequestException, BringParseException) as e:
            raise UpdateFailed(
                "Unable to connect and retrieve user settings from bring"
            ) from e