Common Patterns

Common Patterns

Code patterns and examples for common Home Assistant integration scenarios.

Table of Contents

1. Authentication
2. Data Polling
3. State Restoration
4. Error Recovery
5. Device Discovery
6. Entity Organization

---

Authentication

API Key Authentication

"""Config flow with API key."""
import voluptuous as vol
from homeassistant.const import CONF_API_KEY

async def async_step_user(self, user_input=None):
    """Handle user step with API key."""
    if user_input is not None:
        api_key = user_input[CONF_API_KEY]
        
        try:
            # Validate API key
            client = MyApiClient(api_key)
            await client.async_validate()
        except InvalidAuth:
            return self.async_show_form(
                step_id="user",
                data_schema=vol.Schema({vol.Required(CONF_API_KEY): str}),
                errors={"base": "invalid_auth"},
            )
        
        return self.async_create_entry(
            title="My Integration",
            data={CONF_API_KEY: api_key},
        )
    
    return self.async_show_form(
        step_id="user",
        data_schema=vol.Schema({vol.Required(CONF_API_KEY): str}),
    )

OAuth2 Authentication

"""OAuth2 config flow."""
from homeassistant.helpers import config_entry_oauth2_flow

class MyOAuth2FlowHandler(
    config_entry_oauth2_flow.AbstractOAuth2FlowHandler,
    domain=DOMAIN
):
    """Handle OAuth2 flow."""
    
    DOMAIN = DOMAIN
    
    @property
    def logger(self) -> logging.Logger:
        """Return logger."""
        return _LOGGER
    
    @property
    def extra_authorize_data(self) -> dict:
        """Extra data for authorization."""
        return {"scope": "read write"}

Username/Password Authentication

"""Username/password config flow."""
from homeassistant.const import CONF_USERNAME, CONF_PASSWORD

async def async_step_user(self, user_input=None):
    """Handle user step."""
    errors = {}
    
    if user_input is not None:
        try:
            client = MyApiClient(
                user_input[CONF_USERNAME],
                user_input[CONF_PASSWORD],
            )
            await client.async_login()
            
            return self.async_create_entry(
                title=user_input[CONF_USERNAME],
                data=user_input,
            )
        except InvalidAuth:
            errors["base"] = "invalid_auth"
    
    return self.async_show_form(
        step_id="user",
        data_schema=vol.Schema({
            vol.Required(CONF_USERNAME): str,
            vol.Required(CONF_PASSWORD): str,
        }),
        errors=errors,
    )

---

Data Polling

Fixed Interval Polling

"""Coordinator with fixed 30-second interval."""
from datetime import timedelta

class MyCoordinator(DataUpdateCoordinator):
    """Coordinator with fixed interval."""
    
    def __init__(self, hass, client):
        """Initialize."""
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=30),
        )
        self.client = client
    
    async def _async_update_data(self):
        """Fetch data."""
        return await self.client.async_get_data()

Exponential Backoff on Errors

"""Coordinator with exponential backoff."""
from datetime import timedelta

class MyCoordinator(DataUpdateCoordinator):
    """Coordinator with exponential backoff."""
    
    def __init__(self, hass, client):
        """Initialize."""
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=30),
        )
        self.client = client
        self._error_count = 0
    
    async def _async_update_data(self):
        """Fetch data with backoff."""
        try:
            data = await self.client.async_get_data()
            self._error_count = 0  # Reset on success
            self.update_interval = timedelta(seconds=30)
            return data
        except UpdateFailed as err:
            self._error_count += 1
            # Exponential backoff: 30s, 60s, 120s, max 300s
            backoff = min(30 * (2 ** self._error_count), 300)
            self.update_interval = timedelta(seconds=backoff)
            raise

Event-Driven Updates

"""Coordinator with webhook updates."""
from homeassistant.helpers import webhook

class MyCoordinator(DataUpdateCoordinator):
    """Event-driven coordinator."""
    
    async def async_setup_webhook(self):
        """Set up webhook for push updates."""
        webhook_id = webhook.async_generate_id()
        webhook.async_register(
            self.hass,
            DOMAIN,
            "My Integration",
            webhook_id,
            self._handle_webhook,
        )
        
        # Register webhook with device
        await self.client.async_register_webhook(
            webhook.async_generate_url(self.hass, webhook_id)
        )
    
    async def _handle_webhook(self, hass, webhook_id, request):
        """Handle webhook callback."""
        data = await request.json()
        self.async_set_updated_data(data)
        return web.Response(status=200)

---

State Restoration

RestoreEntity Pattern

"""Entity that restores state after restart."""
from homeassistant.helpers.restore_state import RestoreEntity

class MySwitch(RestoreEntity, SwitchEntity):
    """Switch with state restoration."""
    
    async def async_added_to_hass(self):
        """Restore state when added to HA."""
        await super().async_added_to_hass()
        
        # Restore previous state
        if (last_state := await self.async_get_last_state()) is not None:
            self._attr_is_on = last_state.state == STATE_ON

Storage Helper

"""Using storage helper for persistent data."""
from homeassistant.helpers.storage import Store

STORAGE_VERSION = 1
STORAGE_KEY = f"{DOMAIN}_data"

class MyIntegration:
    """Integration with persistent storage."""
    
    def __init__(self, hass):
        """Initialize."""
        self.store = Store(hass, STORAGE_VERSION, STORAGE_KEY)
        self.data = {}
    
    async def async_load(self):
        """Load data from storage."""
        if (data := await self.store.async_load()) is not None:
            self.data = data
    
    async def async_save(self):
        """Save data to storage."""
        await self.store.async_save(self.data)

---

Error Recovery

Connection Loss Handling

"""Handle connection loss gracefully."""
class MyCoordinator(DataUpdateCoordinator):
    """Coordinator with connection recovery."""
    
    def __init__(self, hass, client):
        """Initialize."""
        super().__init__(
            hass,
            _LOGGER,
            name=DOMAIN,
            update_interval=timedelta(seconds=30),
        )
        self.client = client
        self._connection_lost = False
    
    async def _async_update_data(self):
        """Fetch data with connection recovery."""
        try:
            data = await self.client.async_get_data()
            
            # Log reconnection
            if self._connection_lost:
                _LOGGER.info("Connection restored")
                self._connection_lost = False
            
            return data
            
        except ConnectionError as err:
            # Log disconnection once
            if not self._connection_lost:
                _LOGGER.warning("Connection lost: %s", err)
                self._connection_lost = True
            raise UpdateFailed(f"Connection error: {err}") from err

Authentication Token Refresh

"""Refresh auth token when expired."""
class MyCoordinator(DataUpdateCoordinator):
    """Coordinator with token refresh."""
    
    async def _async_update_data(self):
        """Fetch data with token refresh."""
        try:
            return await self.client.async_get_data()
            
        except TokenExpired:
            # Try to refresh token
            try:
                await self.client.async_refresh_token()
                return await self.client.async_get_data()
            except RefreshFailed as err:
                # Can't refresh, need re-authentication
                raise ConfigEntryAuthFailed from err

Rate Limiting

"""Handle rate limiting."""
import asyncio

class MyApiClient:
    """API client with rate limiting."""
    
    def __init__(self):
        """Initialize."""
        self._last_request_time = 0
        self._min_interval = 1.0  # Minimum 1 second between requests
    
    async def async_make_request(self, endpoint):
        """Make rate-limited request."""
        # Wait if needed
        now = time.time()
        time_since_last = now - self._last_request_time
        if time_since_last < self._min_interval:
            await asyncio.sleep(self._min_interval - time_since_last)
        
        # Make request
        self._last_request_time = time.time()
        return await self._async_request(endpoint)

---

Device Discovery

mDNS/Zeroconf Discovery

"""Discover devices via mDNS."""
from homeassistant.components import zeroconf

async def async_step_zeroconf(
    self, discovery_info: zeroconf.ZeroconfServiceInfo
):
    """Handle zeroconf discovery."""
    # Extract info from discovery
    host = discovery_info.host
    port = discovery_info.port
    name = discovery_info.name
    
    # Set unique ID from discovery
    await self.async_set_unique_id(discovery_info.properties["id"])
    self._abort_if_unique_id_configured()
    
    # Store discovered info
    self.context["title_placeholders"] = {"name": name}
    
    # Validate device is reachable
    try:
        client = MyApiClient(host, port)
        await client.async_validate()
    except CannotConnect:
        return self.async_abort(reason="cannot_connect")
    
    # Show confirmation form
    return await self.async_step_confirm()

SSDP Discovery

"""Discover devices via SSDP."""
from homeassistant.components import ssdp

async def async_step_ssdp(self, discovery_info: ssdp.SsdpServiceInfo):
    """Handle SSDP discovery."""
    # Extract device ID from UDN
    unique_id = discovery_info.upnp[ssdp.ATTR_UPNP_UDN]
    
    await self.async_set_unique_id(unique_id)
    self._abort_if_unique_id_configured()
    
    # Store discovered URL
    self._discovered_url = discovery_info.ssdp_location
    
    return await self.async_step_confirm()

---

Entity Organization

Device Grouping

"""Group multiple entities under one device."""
from homeassistant.helpers.device_registry import DeviceInfo

class MyEntity(CoordinatorEntity):
    """Base entity with device info."""
    
    _attr_has_entity_name = True
    
    def __init__(self, coordinator, device_id):
        """Initialize."""
        super().__init__(coordinator)
        self._device_id = device_id
    
    @property
    def device_info(self) -> DeviceInfo:
        """Return device info."""
        device = self.coordinator.data[self._device_id]
        return DeviceInfo(
            identifiers={(DOMAIN, self._device_id)},
            name=device["name"],
            manufacturer=device.get("manufacturer"),
            model=device.get("model"),
            sw_version=device.get("sw_version"),
            hw_version=device.get("hw_version"),
            configuration_url=device.get("url"),
        )

Multiple Platforms from One Coordinator

"""Setup multiple platforms from coordinator."""
# __init__.py
PLATFORMS = [Platform.SENSOR, Platform.BINARY_SENSOR, Platform.SWITCH]

async def async_setup_entry(hass, entry):
    """Set up from config entry."""
    coordinator = MyCoordinator(hass, entry.data)
    await coordinator.async_config_entry_first_refresh()
    
    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = coordinator
    
    await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
    return True

# sensor.py
async def async_setup_entry(hass, entry, async_add_entities):
    """Set up sensors."""
    coordinator = hass.data[DOMAIN][entry.entry_id]
    
    entities = []
    for device_id in coordinator.data:
        entities.extend([
            TemperatureSensor(coordinator, device_id),
            HumiditySensor(coordinator, device_id),
        ])
    
    async_add_entities(entities)

---

More Patterns

See also:

• Development Guide - Core implementation patterns
• Testing Guide - Testing patterns
• Quality Scale Guide - Best practices

Resources

• HA Developer Docs
• Entity Documentation
• DataUpdateCoordinator