config_flow.py

Go to the documentation of this file.

"""Config flow for JuiceNet integration."""
 
import logging
from typing import Any
 
import aiohttp
from pyjuicenet import Api, TokenError
import voluptuous as vol
 
from homeassistant import core, exceptions
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_ACCESS_TOKEN
from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
from .const import DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
DATA_SCHEMA = vol.Schema({vol.Required(CONF_ACCESS_TOKEN): str})
 
 
async def validate_input(hass: core.HomeAssistant, data):
    """Validate the user input allows us to connect.
 
    Data has the keys from DATA_SCHEMA with values provided by the user.
    """
    session = async_get_clientsession(hass)
    juicenet = Api(data[CONF_ACCESS_TOKEN], session)
 
    try:
        await juicenet.get_devices()
    except TokenError as error:
        _LOGGER.error("Token Error %s", error)
        raise InvalidAuth from error
    except aiohttp.ClientError as error:
        _LOGGER.error("Error connecting %s", error)
        raise CannotConnect from error
 
    # Return info that you want to store in the config entry.
    return {"title": "JuiceNet"}
 
 
class JuiceNetConfigFlow(ConfigFlow, domain=DOMAIN):
    """Handle a config flow for JuiceNet."""
 
    VERSION = 1
 
    async def async_step_user(
        self, user_input: dict[str, Any] | None = None
    ) -> ConfigFlowResult:
        """Handle the initial step."""
        errors = {}
        if user_input is not None:
            await self.async_set_unique_idasync_set_unique_id(user_input[CONF_ACCESS_TOKEN])
            self._abort_if_unique_id_configured_abort_if_unique_id_configured()
 
            try:
                info = await validate_input(self.hass, user_input)
                return self.async_create_entryasync_create_entryasync_create_entry(title=info["title"], data=user_input)
            except CannotConnect:
                errors["base"] = "cannot_connect"
            except InvalidAuth:
                errors["base"] = "invalid_auth"
            except Exception:
                _LOGGER.exception("Unexpected exception")
                errors["base"] = "unknown"
 
        return self.async_show_formasync_show_formasync_show_form(
            step_id="user", data_schema=DATA_SCHEMA, errors=errors
        )
 
    async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
        """Handle import."""
        return await self.async_step_userasync_step_userasync_step_user(import_data)
 
 
class CannotConnect(exceptions.HomeAssistantError):
    """Error to indicate we cannot connect."""
 
 
class InvalidAuth(exceptions.HomeAssistantError):
    """Error to indicate there is invalid auth."""