Config flows

A config flow lets users set up an integration from the UI. The result of a successful flow is a config entry — stored configuration that the core passes to your integration's async_setup_entry. Config flows live in apexos/config_entries.py; your integration implements one in config_flow.py and sets "config_flow": true in its manifest.

The minimal flow

The sun integration ships the smallest possible flow (apexos/components/sun/config_flow.py):

from apexos.config_entries import ConfigFlow, ConfigFlowResult

from .const import DEFAULT_NAME, DOMAIN


class SunConfigFlow(ConfigFlow, domain=DOMAIN):
    """Config flow for Sun."""

    VERSION = 1

    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initialized by the user."""
        if user_input is not None:
            return self.async_create_entry(title=DEFAULT_NAME, data={})

        return self.async_show_form(step_id="user")

Key mechanics:

• Subclassing ConfigFlow with the domain=DOMAIN class keyword registers the handler for your domain (__init_subclass__ does the registration).
• VERSION is stored on created entries and drives config-entry migration later.
• Inside a flow, the core object is available as self.apex.

Steps and results

Each step is a method named async_step_<step_id> returning a ConfigFlowResult. The first user-initiated step is always async_step_user. A step either shows a form, creates an entry, or aborts:

• self.async_show_form(step_id=..., data_schema=..., errors=...) — render a form. data_schema is a voluptuous schema; errors maps a field name (or "base") to an error key defined in strings.json.
• self.async_create_entry(title=..., data=...) — finish the flow and store the entry.
• self.async_abort(reason=...) — abort; reason must have a matching entry under config.abort in strings.json.

The scaffolded template (python3 -m script.scaffold config_flow) shows the standard validate-then-create pattern:

async def async_step_user(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Handle the initial step."""
    errors: dict[str, str] = {}
    if user_input is not None:
        try:
            info = await validate_input(self.apex, user_input)
        except CannotConnect:
            errors["base"] = "cannot_connect"
        except InvalidAuth:
            errors["base"] = "invalid_auth"
        except Exception:
            _LOGGER.exception("Unexpected exception")
            errors["base"] = "unknown"
        else:
            return self.async_create_entry(title=info["title"], data=user_input)

    return self.async_show_form(
        step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
    )

validate_input(apex, data) should actually talk to the device/service before the entry is created. If your client library is synchronous, wrap it:

await apex.async_add_executor_job(your_validate_func, data[CONF_USERNAME], data[CONF_PASSWORD])

Unique IDs

Assign each configured device/service a unique ID so it can only be set up once and so discovery can update existing entries:

await self.async_set_unique_id(serial)
self._abort_if_unique_id_configured(updates={CONF_HOST: discovery_info.host})

• async_set_unique_id(unique_id, raise_on_progress=True) records the ID on the flow context (readable via the unique_id property).
• _abort_if_unique_id_configured(updates=..., reload_on_update=True, error="already_configured") aborts if an entry with that unique ID exists — optionally updating (and reloading) the existing entry with fresh data such as a changed host. The abort reason requires an already_configured entry in strings.json.
• _abort_if_unique_id_mismatch(reason="unique_id_mismatch") is used in reauth and reconfigure flows to guarantee the user is still talking to the same device.
• _async_abort_entries_match({...}) aborts when an existing entry matches all the given data — useful when no natural unique ID exists.

Discovery steps

If your manifest declares discovery (see the manifest reference), implement the matching step; the flow's source will be one of the SOURCE_* constants from apexos.config_entries:

async_step_zeroconf, async_step_ssdp, async_step_dhcp, async_step_usb, async_step_bluetooth, async_step_mqtt, async_step_homekit, async_step_discovery, async_step_hardware, async_step_integration_discovery, plus async_step_import (migrating YAML config, SOURCE_IMPORT). Each receives a typed service-info object (e.g. ZeroconfServiceInfo from apexos.helpers.service_info.zeroconf).

Options flows

To let users tweak settings after setup, expose an options flow from your config flow class:

@staticmethod
@callback
def async_get_options_flow(config_entry: ConfigEntry) -> OptionsFlow:
    """Get the options flow for this handler."""
    return MyOptionsFlow()

• OptionsFlow gives you self.config_entry (not available inside __init__ — only after the flow is initialized) and the same step/form/result API; the first step is async_step_init.
• OptionsFlowWithReload automatically reloads the config entry when the flow finishes with changed options (automatic_reload = True). Don't combine it with config-entry update listeners.
• OptionsFlowWithConfigEntry is deprecated — kept only for backward compatibility with custom integrations; do not use it in new code.

Reauthentication

When credentials stop working, start a reauth flow; the core calls async_step_reauth with the entry's current data. The established pattern (from apexos/components/nrgkick/config_flow.py):

async def async_step_reauth(
    self, entry_data: Mapping[str, Any]
) -> ConfigFlowResult:
    """Handle initiation of reauthentication."""
    return await self.async_step_reauth_confirm()

async def async_step_reauth_confirm(
    self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
    """Handle reauthentication."""
    errors: dict[str, str] = {}
    if user_input is not None:
        reauth_entry = self._get_reauth_entry()
        if info := await self._async_validate_credentials(...):
            await self.async_set_unique_id(info["serial"], raise_on_progress=False)
            self._abort_if_unique_id_mismatch()
            return self.async_update_reload_and_abort(
                reauth_entry,
                data_updates=user_input,
            )
    return self.async_show_form(step_id="reauth_confirm", ...)

async_update_reload_and_abort(entry, *, unique_id, title, data, data_updates, options, reason, reload_even_if_entry_is_unchanged=True) updates the entry, schedules a reload, and aborts the flow. data_updates merges into existing data (mutually exclusive with data, which replaces it). The abort reason defaults to reauth_successful — or reconfigure_successful when the source is a reconfigure flow. async_update_and_abort is the same without the reload.

Reconfiguration

async_step_reconfigure lets the user change settings such as the host. Use self._get_reconfigure_entry() to fetch the entry, re-validate, verify identity with _abort_if_unique_id_mismatch(), then finish with async_update_reload_and_abort(reconfigure_entry, data_updates={...}).

Setting up from the entry

Once a flow creates an entry, the core calls into your __init__.py:

type MyConfigEntry = ConfigEntry[MyApi]

async def async_setup_entry(apex: ApexOS, entry: MyConfigEntry) -> bool:
    """Set up from a config entry."""
    entry.runtime_data = MyApi(...)          # typed per-entry runtime object
    await apex.config_entries.async_forward_entry_setups(entry, _PLATFORMS)
    return True

async def async_unload_entry(apex: ApexOS, entry: MyConfigEntry) -> bool:
    """Unload a config entry."""
    return await apex.config_entries.async_unload_platforms(entry, _PLATFORMS)

Texts for the flow

All user-visible flow texts live in strings.json under the config key: config.step.<step_id> (title, description, data field labels), config.error.<error_key>, and config.abort.<reason>. Shared texts can be referenced with the [%key:...%] syntax, e.g. "[%key:common::config_flow::description::confirm_setup%]" as used by sun.