config_flow.py

Go to the documentation of this file.

"""Config flow for FireServiceRota."""
 
from __future__ import annotations
 
from collections.abc import Mapping
from typing import Any
 
from pyfireservicerota import FireServiceRota, InvalidAuthError
import voluptuous as vol
 
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_PASSWORD, CONF_TOKEN, CONF_URL, CONF_USERNAME
 
from .const import DOMAIN, URL_LIST
 
DATA_SCHEMA = vol.Schema(
    {
        vol.Required(CONF_URL, default="www.brandweerrooster.nl"): vol.In(URL_LIST),
        vol.Required(CONF_USERNAME): str,
        vol.Required(CONF_PASSWORD): str,
    }
)
 
 
class FireServiceRotaFlowHandler(ConfigFlow, domain=DOMAIN):
    """Handle a FireServiceRota config flow."""
 
    VERSION = 1
 
    def __init__(self) -> None:
        """Initialize config flow."""
        self.apiapi = None
        self._base_url_base_url = None
        self._username_username = None
        self._password_password = None
        self._existing_entry_existing_entry: dict[str, Any] | None = None
        self._description_placeholders_description_placeholders: dict[str, str] | None = None
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle a flow initiated by the user."""
        errors: dict[str, str] = {}
 
        if user_input is None:
            return self._show_setup_form_show_setup_form(user_input, errors)
 
        return await self._validate_and_create_entry_validate_and_create_entry(user_input, "user")
 
    async def _validate_and_create_entry(self, user_input, step_id):
        """Check if config is valid and create entry if so."""
        self._password_password = user_input[CONF_PASSWORD]
 
        extra_inputs = user_input
 
        if self._existing_entry_existing_entry:
            extra_inputs = self._existing_entry_existing_entry
 
        self._username_username = extra_inputs[CONF_USERNAME]
        self._base_url_base_url = extra_inputs[CONF_URL]
 
        if self.unique_idunique_id is None:
            await self.async_set_unique_idasync_set_unique_id(self._username_username)
            self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
        self.apiapi = FireServiceRota(
            base_url=self._base_url_base_url,
            username=self._username_username,
            password=self._password_password,
        )
 
        try:
            token_info = await self.hass.async_add_executor_job(self.apiapi.request_tokens)
        except InvalidAuthError:
            self.apiapi = None
            return self.async_show_formasync_show_formasync_show_form(
                step_id=step_id,
                data_schema=DATA_SCHEMA,
                errors={"base": "invalid_auth"},
            )
 
        data = {
            "auth_implementation": DOMAIN,
            CONF_URL: self._base_url_base_url,
            CONF_USERNAME: self._username_username,
            CONF_TOKEN: token_info,
        }
 
        if step_id == "user":
            return self.async_create_entryasync_create_entryasync_create_entry(title=self._username_username, data=data)
 
        entry = await self.async_set_unique_idasync_set_unique_id(self.unique_idunique_id)
        self.hass.config_entries.async_update_entry(entry, data=data)
        await self.hass.config_entries.async_reload(entry.entry_id)
        return self.async_abortasync_abortasync_abort(reason="reauth_successful")
 
    def _show_setup_form(self, user_input=None, errors=None, step_id="user"):
        """Show the setup form to the user."""
 
        if user_input is None:
            user_input = {}
 
        if step_id == "user":
            schema = {
                vol.Required(CONF_URL, default="www.brandweerrooster.nl"): vol.In(
                    URL_LIST
                ),
                vol.Required(CONF_USERNAME): str,
                vol.Required(CONF_PASSWORD): str,
            }
        else:
            schema = {vol.Required(CONF_PASSWORD): str}
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id=step_id,
            data_schema=vol.Schema(schema),
            errors=errors or {},
            description_placeholders=self._description_placeholders_description_placeholders,
        )
 
    async def async_step_reauth(
        self, entry_data: Mapping[str, Any]
    ) -> ConfigFlowResult:
        """Initialise re-authentication."""
        await self.async_set_unique_idasync_set_unique_id(entry_data[CONF_USERNAME])
        self._existing_entry_existing_entry = {**entry_data}
        self._description_placeholders_description_placeholders = {CONF_USERNAME: entry_data[CONF_USERNAME]}
        return await self.async_step_reauth_confirmasync_step_reauth_confirm()
 
    async def async_step_reauth_confirm(
        self, user_input: dict[str, str] | None = None
    ) -> ConfigFlowResult:
        """Get new tokens for a config entry that can't authenticate."""
        if user_input is None:
            return self._show_setup_form_show_setup_form(step_id="reauth_confirm")
 
        return await self._validate_and_create_entry_validate_and_create_entry(user_input, "reauth_confirm")