diff --git a/homeassistant/components/blink/config_flow.py b/homeassistant/components/blink/config_flow.py index 3fd1b7d91e5..445a84f838c 100644 --- a/homeassistant/components/blink/config_flow.py +++ b/homeassistant/components/blink/config_flow.py @@ -9,7 +9,7 @@ from blinkpy.auth import Auth, LoginError, TokenRefreshFailed from blinkpy.blinkpy import Blink, BlinkSetupError import voluptuous as vol -from homeassistant import config_entries, core, exceptions +from homeassistant.config_entries import ConfigEntry, ConfigFlow from homeassistant.const import ( CONF_PASSWORD, CONF_PIN, @@ -18,6 +18,7 @@ from homeassistant.const import ( ) from homeassistant.core import callback from homeassistant.data_entry_flow import FlowResult +from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers import selector from homeassistant.helpers.schema_config_entry_flow import ( SchemaFlowFormStep, @@ -48,7 +49,7 @@ OPTIONS_FLOW = { } -def validate_input(hass: core.HomeAssistant, auth): +def validate_input(auth: Auth) -> None: """Validate the user input allows us to connect.""" try: auth.startup() @@ -58,7 +59,7 @@ def validate_input(hass: core.HomeAssistant, auth): raise Require2FA -def _send_blink_2fa_pin(auth, pin): +def _send_blink_2fa_pin(auth: Auth, pin: str) -> bool: """Send 2FA pin to blink servers.""" blink = Blink() blink.auth = auth @@ -67,38 +68,34 @@ def _send_blink_2fa_pin(auth, pin): return auth.send_auth_key(blink, pin) -class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): +class BlinkConfigFlow(ConfigFlow, domain=DOMAIN): """Handle a Blink config flow.""" VERSION = 3 def __init__(self) -> None: """Initialize the blink flow.""" - self.auth = None + self.auth: Auth | None = None @staticmethod @callback def async_get_options_flow( - config_entry: config_entries.ConfigEntry, + config_entry: ConfigEntry, ) -> SchemaOptionsFlowHandler: """Get options flow for this handler.""" return SchemaOptionsFlowHandler(config_entry, OPTIONS_FLOW) - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Handle a flow initiated by the user.""" errors = {} - data = {CONF_USERNAME: "", CONF_PASSWORD: "", "device_id": DEVICE_ID} if user_input is not None: - data[CONF_USERNAME] = user_input["username"] - data[CONF_PASSWORD] = user_input["password"] - - self.auth = Auth(data, no_prompt=True) - await self.async_set_unique_id(data[CONF_USERNAME]) + self.auth = Auth({**user_input, "device_id": DEVICE_ID}, no_prompt=True) + await self.async_set_unique_id(user_input[CONF_USERNAME]) try: - await self.hass.async_add_executor_job( - validate_input, self.hass, self.auth - ) + await self.hass.async_add_executor_job(validate_input, self.auth) return self._async_finish_flow() except Require2FA: return await self.async_step_2fa() @@ -108,18 +105,20 @@ class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): _LOGGER.exception("Unexpected exception") errors["base"] = "unknown" - data_schema = { - vol.Required("username"): str, - vol.Required("password"): str, - } - return self.async_show_form( step_id="user", - data_schema=vol.Schema(data_schema), + data_schema=vol.Schema( + { + vol.Required(CONF_USERNAME): str, + vol.Required(CONF_PASSWORD): str, + } + ), errors=errors, ) - async def async_step_2fa(self, user_input=None): + async def async_step_2fa( + self, user_input: dict[str, Any] | None = None + ) -> FlowResult: """Handle 2FA step.""" errors = {} if user_input is not None: @@ -142,7 +141,7 @@ class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return self.async_show_form( step_id="2fa", data_schema=vol.Schema( - {vol.Optional("pin"): vol.All(str, vol.Length(min=1))} + {vol.Optional(CONF_PIN): vol.All(str, vol.Length(min=1))} ), errors=errors, ) @@ -152,14 +151,15 @@ class BlinkConfigFlow(config_entries.ConfigFlow, domain=DOMAIN): return await self.async_step_user(dict(entry_data)) @callback - def _async_finish_flow(self): + def _async_finish_flow(self) -> FlowResult: """Finish with setup.""" + assert self.auth return self.async_create_entry(title=DOMAIN, data=self.auth.login_attributes) -class Require2FA(exceptions.HomeAssistantError): +class Require2FA(HomeAssistantError): """Error to indicate we require 2FA.""" -class InvalidAuth(exceptions.HomeAssistantError): +class InvalidAuth(HomeAssistantError): """Error to indicate there is invalid auth."""