Add typing to Blink config flow (#98873)

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"]
+            self.auth = Auth({**user_input, "device_id": DEVICE_ID}, no_prompt=True)
-            data[CONF_PASSWORD] = user_input["password"]
+            await self.async_set_unique_id(user_input[CONF_USERNAME])
-
-            self.auth = Auth(data, no_prompt=True)
-            await self.async_set_unique_id(data[CONF_USERNAME])
 
             try:
-                await self.hass.async_add_executor_job(
+                await self.hass.async_add_executor_job(validate_input, self.auth)
-                    validate_input, self.hass, 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."""