ha-coordinator

Home Assistant DataUpdateCoordinator Pattern

The DataUpdateCoordinator is the most important architectural pattern for polling-based integrations. It centralizes data fetching, distributes updates to entities, and handles error/retry logic automatically.

Complete Coordinator Template (2025)

"""DataUpdateCoordinator for {Name}."""
from __future__ import annotations

import logging
from datetime import timedelta
from typing import Any

from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed

from .const import DEFAULT_SCAN_INTERVAL, DOMAIN

_LOGGER = logging.getLogger(__name__)


class {Name}Coordinator(DataUpdateCoordinator[dict[str, Any]]):
    """Coordinator to manage data fetching."""

    config_entry: ConfigEntry

    def __init__(self, hass: HomeAssistant, entry: ConfigEntry) -> None:
        """Initialize coordinator."""
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            config_entry=entry,
            update_interval=timedelta(
                seconds=entry.options.get("scan_interval", DEFAULT_SCAN_INTERVAL)
            ),
            always_update=False,  # Safe only if returned data has value-based __eq__ (plain dict/dataclass); use True otherwise
        )
        # MyClient, AuthenticationError, RateLimitError are placeholders for your integration's API library:
        # from my_api import MyClient, AuthenticationError, RateLimitError
        self.client = MyClient(
            host=entry.data["host"],
            username=entry.data["username"],
            password=entry.data["password"],
        )

    async def _async_setup(self) -> None:
        """One-time initialization (Home Assistant 2024.8+).

        Called once during async_config_entry_first_refresh() before _async_update_data.
        Use for data that doesn't change: device info, serial numbers, capabilities.
        Same error handling as _async_update_data (ConfigEntryAuthFailed, UpdateFailed).
        """
        self.device_info = await self.client.async_get_device_info()

    async def _async_update_data(self) -> dict[str, Any]:
        """Fetch data from the device/API.

        Called on every polling cycle.

        Raises:
            ConfigEntryAuthFailed: Triggers reauth flow (expired/invalid credentials)
            UpdateFailed: Marks entities unavailable, logs once, retries on schedule
        """
        try:
            return await self.client.async_get_data()
        except AuthenticationError as err:
            raise ConfigEntryAuthFailed("Authentication failed") from err
        except ConnectionError as err:
            raise UpdateFailed(f"Connection error: {err}") from err
        except RateLimitError as err:
            # HA 2025.11+: retry_after defers the next scheduled refresh
            raise UpdateFailed(
                f"Rate limited: {err}",
                retry_after=err.retry_after,
            ) from err
        except TimeoutError as err:
            raise UpdateFailed(f"Timeout: {err}") from err