Rework devolo Home Control config flow (#147121)

homeassistant/components/devolo_home_control/config_flow.py

@@ -7,45 +7,39 @@ from typing import Any
 
 import voluptuous as vol
 
-from homeassistant.config_entries import (
-    SOURCE_REAUTH,
-    ConfigEntry,
-    ConfigFlow,
-    ConfigFlowResult,
-)
+from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlow, ConfigFlowResult
 from homeassistant.const import CONF_PASSWORD, CONF_USERNAME
-from homeassistant.core import callback
 from homeassistant.helpers.service_info.zeroconf import ZeroconfServiceInfo
 
 from . import configure_mydevolo
 from .const import DOMAIN, SUPPORTED_MODEL_TYPES
 from .exceptions import CredentialsInvalid, UuidChanged
 
+DATA_SCHEMA = vol.Schema(
+    {vol.Required(CONF_USERNAME): str, vol.Required(CONF_PASSWORD): str}
+)
+
 
 class DevoloHomeControlFlowHandler(ConfigFlow, domain=DOMAIN):
     """Handle a devolo HomeControl config flow."""
 
     VERSION = 1
 
-    _reauth_entry: ConfigEntry
-
-    def __init__(self) -> None:
-        """Initialize devolo Home Control flow."""
-        self.data_schema = {
-            vol.Required(CONF_USERNAME): str,
-            vol.Required(CONF_PASSWORD): str,
-        }
-
     async def async_step_user(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
         """Handle a flow initiated by the user."""
-        if user_input is None:
-            return self._show_form(step_id="user")
-        try:
-            return await self._connect_mydevolo(user_input)
-        except CredentialsInvalid:
-            return self._show_form(step_id="user", errors={"base": "invalid_auth"})
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            try:
+                return await self._connect_mydevolo(user_input)
+            except CredentialsInvalid:
+                errors["base"] = "invalid_auth"
+
+        return self.async_show_form(
+            step_id="user", data_schema=DATA_SCHEMA, errors=errors
+        )
 
     async def async_step_zeroconf(
         self, discovery_info: ZeroconfServiceInfo
@@ -61,42 +55,47 @@ class DevoloHomeControlFlowHandler(ConfigFlow, domain=DOMAIN):
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
         """Handle a flow initiated by zeroconf."""
-        if user_input is None:
-            return self._show_form(step_id="zeroconf_confirm")
-        try:
-            return await self._connect_mydevolo(user_input)
-        except CredentialsInvalid:
-            return self._show_form(
-                step_id="zeroconf_confirm", errors={"base": "invalid_auth"}
-            )
+        errors: dict[str, str] = {}
+
+        if user_input is not None:
+            try:
+                return await self._connect_mydevolo(user_input)
+            except CredentialsInvalid:
+                errors["base"] = "invalid_auth"
+
+        return self.async_show_form(
+            step_id="zeroconf_confirm", data_schema=DATA_SCHEMA, errors=errors
+        )
 
     async def async_step_reauth(
         self, entry_data: Mapping[str, Any]
     ) -> ConfigFlowResult:
         """Handle reauthentication."""
-        self._reauth_entry = self._get_reauth_entry()
-        self.data_schema = {
-            vol.Required(CONF_USERNAME, default=entry_data[CONF_USERNAME]): str,
-            vol.Required(CONF_PASSWORD): str,
-        }
         return await self.async_step_reauth_confirm()
 
     async def async_step_reauth_confirm(
         self, user_input: dict[str, Any] | None = None
     ) -> ConfigFlowResult:
         """Handle a flow initiated by reauthentication."""
-        if user_input is None:
-            return self._show_form(step_id="reauth_confirm")
-        try:
-            return await self._connect_mydevolo(user_input)
-        except CredentialsInvalid:
-            return self._show_form(
-                step_id="reauth_confirm", errors={"base": "invalid_auth"}
-            )
-        except UuidChanged:
-            return self._show_form(
-                step_id="reauth_confirm", errors={"base": "reauth_failed"}
-            )
+        errors: dict[str, str] = {}
+        data_schema = vol.Schema(
+            {
+                vol.Required(CONF_USERNAME, default=self.init_data[CONF_USERNAME]): str,
+                vol.Required(CONF_PASSWORD): str,
+            }
+        )
+
+        if user_input is not None:
+            try:
+                return await self._connect_mydevolo(user_input)
+            except CredentialsInvalid:
+                errors["base"] = "invalid_auth"
+            except UuidChanged:
+                errors["base"] = "reauth_failed"
+
+        return self.async_show_form(
+            step_id="reauth_confirm", data_schema=data_schema, errors=errors
+        )
 
     async def _connect_mydevolo(self, user_input: dict[str, Any]) -> ConfigFlowResult:
         """Connect to mydevolo."""
@@ -119,21 +118,11 @@ class DevoloHomeControlFlowHandler(ConfigFlow, domain=DOMAIN):
                 },
             )
 
-        if self._reauth_entry.unique_id != uuid:
+        if self.unique_id != uuid:
             # The old user and the new user are not the same. This could mess-up everything as all unique IDs might change.
             raise UuidChanged
 
+        reauth_entry = self._get_reauth_entry()
         return self.async_update_reload_and_abort(
-            self._reauth_entry, data=user_input, unique_id=uuid
-        )
-
-    @callback
-    def _show_form(
-        self, step_id: str, errors: dict[str, str] | None = None
-    ) -> ConfigFlowResult:
-        """Show the form to the user."""
-        return self.async_show_form(
-            step_id=step_id,
-            data_schema=vol.Schema(self.data_schema),
-            errors=errors if errors else {},
+            reauth_entry, data=user_input, unique_id=uuid
         )