Most efficient way to remove obsolute integration devices from registry

Development
1

Hi there,
I’m developping a cloud based Api integration that acts as a hub for couple of devices that could be configured for a user entry/account.
Since those devices could change, new ones added and old ones removed, I want the integration to refresh the latest hub state as well during the setup routine (which is called when integration is reloaded or has relevant option changes).

What I came up so far in the is following in
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) → bool:
…

    await coordinator.async_config_entry_first_refresh()
    hass.data[DOMAIN][entry.entry_id] = coordinator

    # first check for orphaned devices no longer contained in actual api data upon reloads or config option changes
    # get device registry and device ids that are orphaned
    dev_registry = dr.async_get(coordinator.hass)
    active_serials = coordinator.data.keys()
    dev_list = [dev_entry.id for dev_entry in dev_registry.devices.data.values() if any(identifier for identifier in dev_entry.identifiers if identifier[0] == DOMAIN and identifier[1] not in active_serials)]
    for dev_id in dev_list:
        if device := dev_registry.async_get(dev_id):
            dev_registry.async_remove_device(device.id)
            LOGGER.info(
                "Removed device entry %s from registry for unused configuration entry device %s", device.id, device.serial_number
            )

This way to cycle through the device registry works, but I wonder whether there is a better more elegant way to do so?

For example it would be much easier to get a list of all device.id that belong to a certain config entry, but I failed to find a proper method. Hence the rather complex list comprehension to compile a list with obsolete device entries based on integration domain and device serial number (used for the device identifier)

Also I’m not sure if with the async_remove_device method, the entities bound to that device will also be removed automatically. It looks like this is the case, but I’m not certain.

2

You can use

async_entries_for_config_entry(dev_registry, config_entry.entry_id)

to get the devices for your integration. It however does basically the same thing in the background.

def async_entries_for_config_entry(
    registry: DeviceRegistry, config_entry_id: str
) -> list[DeviceEntry]:
    """Return entries that match a config entry."""
    return [
        device
        for device in registry.devices.values()
        if config_entry_id in device.config_entries
    ]