From 0fb1fbf0d14e82e371ed9f14784549c325033d53 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Wed, 4 Sep 2024 18:38:34 +0200 Subject: [PATCH] Improve config flow type hints (q-s) (#125198) * Improve config flow type hints (q-s) * Revert screenlogic * Revert starline --- .../components/rachio/config_flow.py | 4 ++- .../components/radiotherm/config_flow.py | 4 ++- .../components/reolink/config_flow.py | 2 +- homeassistant/components/roon/config_flow.py | 10 ++++--- homeassistant/components/sense/config_flow.py | 13 ++++----- .../components/smappee/config_flow.py | 14 +++++++--- .../components/smartthings/config_flow.py | 27 ++++++++++++------- .../components/smarttub/config_flow.py | 6 ++++- homeassistant/components/soma/config_flow.py | 2 +- .../components/somfy_mylink/config_flow.py | 14 +++++++--- .../components/songpal/config_flow.py | 20 +++++++------- .../components/soundtouch/config_flow.py | 4 ++- .../components/syncthru/config_flow.py | 4 ++- 13 files changed, 82 insertions(+), 42 deletions(-) diff --git a/homeassistant/components/rachio/config_flow.py b/homeassistant/components/rachio/config_flow.py index bdd2f81536d..66811091820 100644 --- a/homeassistant/components/rachio/config_flow.py +++ b/homeassistant/components/rachio/config_flow.py @@ -118,7 +118,9 @@ class OptionsFlowHandler(OptionsFlow): """Initialize options flow.""" self.config_entry = config_entry - async def async_step_init(self, user_input=None): + async def async_step_init( + self, user_input: dict[str, int] | None = None + ) -> ConfigFlowResult: """Handle options flow.""" if user_input is not None: return self.async_create_entry(title="", data=user_input) diff --git a/homeassistant/components/radiotherm/config_flow.py b/homeassistant/components/radiotherm/config_flow.py index e9904318ae9..6bcbe11872d 100644 --- a/homeassistant/components/radiotherm/config_flow.py +++ b/homeassistant/components/radiotherm/config_flow.py @@ -60,7 +60,9 @@ class RadioThermConfigFlow(ConfigFlow, domain=DOMAIN): self.discovered_ip = discovery_info.ip return await self.async_step_confirm() - async def async_step_confirm(self, user_input=None): + async def async_step_confirm( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Attempt to confirm.""" ip_address = self.discovered_ip init_data = self.discovered_init_data diff --git a/homeassistant/components/reolink/config_flow.py b/homeassistant/components/reolink/config_flow.py index 6d0381b025f..067a7e24b8e 100644 --- a/homeassistant/components/reolink/config_flow.py +++ b/homeassistant/components/reolink/config_flow.py @@ -48,7 +48,7 @@ DEFAULT_OPTIONS = {CONF_PROTOCOL: DEFAULT_PROTOCOL} class ReolinkOptionsFlowHandler(OptionsFlow): """Handle Reolink options.""" - def __init__(self, config_entry): + def __init__(self, config_entry: ConfigEntry) -> None: """Initialize ReolinkOptionsFlowHandler.""" self.config_entry = config_entry diff --git a/homeassistant/components/roon/config_flow.py b/homeassistant/components/roon/config_flow.py index de220454852..b896f6775ae 100644 --- a/homeassistant/components/roon/config_flow.py +++ b/homeassistant/components/roon/config_flow.py @@ -142,9 +142,11 @@ class RoonConfigFlow(ConfigFlow, domain=DOMAIN): return await self.async_step_fallback() - async def async_step_fallback(self, user_input=None): + async def async_step_fallback( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Get host and port details from the user.""" - errors = {} + errors: dict[str, str] = {} if user_input is not None: self._host = user_input["host"] @@ -155,7 +157,9 @@ class RoonConfigFlow(ConfigFlow, domain=DOMAIN): step_id="fallback", data_schema=DATA_SCHEMA, errors=errors ) - async def async_step_link(self, user_input=None): + async def async_step_link( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle linking and authenticating with the roon server.""" errors = {} if user_input is not None: diff --git a/homeassistant/components/sense/config_flow.py b/homeassistant/components/sense/config_flow.py index 222c6b30f79..c0df40aec9d 100644 --- a/homeassistant/components/sense/config_flow.py +++ b/homeassistant/components/sense/config_flow.py @@ -3,7 +3,7 @@ from collections.abc import Mapping from functools import partial import logging -from typing import TYPE_CHECKING, Any +from typing import Any from sense_energy import ( ASyncSenseable, @@ -34,9 +34,10 @@ class SenseConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 + _gateway: ASyncSenseable + def __init__(self) -> None: """Init Config .""" - self._gateway: ASyncSenseable | None = None self._auth_data: dict[str, Any] = {} async def validate_input(self, data: Mapping[str, Any]) -> None: @@ -58,14 +59,12 @@ class SenseConfigFlow(ConfigFlow, domain=DOMAIN): client_session=client_session, ) ) - if TYPE_CHECKING: - assert self._gateway self._gateway.rate_limit = ACTIVE_UPDATE_RATE await self._gateway.authenticate( self._auth_data[CONF_EMAIL], self._auth_data[CONF_PASSWORD] ) - async def create_entry_from_data(self): + async def create_entry_from_data(self) -> ConfigFlowResult: """Create the entry from the config data.""" self._auth_data["access_token"] = self._gateway.sense_access_token self._auth_data["user_id"] = self._gateway.sense_user_id @@ -99,7 +98,9 @@ class SenseConfigFlow(ConfigFlow, domain=DOMAIN): return await self.create_entry_from_data() return None - async def async_step_validation(self, user_input=None): + async def async_step_validation( + self, user_input: dict[str, str] | None = None + ) -> ConfigFlowResult: """Handle validation (2fa) step.""" errors = {} if user_input: diff --git a/homeassistant/components/smappee/config_flow.py b/homeassistant/components/smappee/config_flow.py index d5073bd9c34..f92f8b17662 100644 --- a/homeassistant/components/smappee/config_flow.py +++ b/homeassistant/components/smappee/config_flow.py @@ -69,9 +69,11 @@ class SmappeeFlowHandler( return await self.async_step_zeroconf_confirm() - async def async_step_zeroconf_confirm(self, user_input=None): + async def async_step_zeroconf_confirm( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Confirm zeroconf flow.""" - errors = {} + errors: dict[str, str] = {} # Check if already configured (cloud) if self.is_cloud_device_already_added(): @@ -118,7 +120,9 @@ class SmappeeFlowHandler( return await self.async_step_environment() - async def async_step_environment(self, user_input=None): + async def async_step_environment( + self, user_input: dict[str, str] | None = None + ) -> ConfigFlowResult: """Decide environment, cloud or local.""" if user_input is None: return self.async_show_form( @@ -144,7 +148,9 @@ class SmappeeFlowHandler( return await self.async_step_pick_implementation() - async def async_step_local(self, user_input=None): + async def async_step_local( + self, user_input: dict[str, str] | None = None + ) -> ConfigFlowResult: """Handle local flow.""" if user_input is None: return self.async_show_form( diff --git a/homeassistant/components/smartthings/config_flow.py b/homeassistant/components/smartthings/config_flow.py index df5b7a8acfa..081f833787e 100644 --- a/homeassistant/components/smartthings/config_flow.py +++ b/homeassistant/components/smartthings/config_flow.py @@ -42,16 +42,17 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN): VERSION = 2 + api: SmartThings + app_id: str + location_id: str + def __init__(self) -> None: """Create a new instance of the flow handler.""" - self.access_token = None - self.app_id = None - self.api = None + self.access_token: str | None = None self.oauth_client_secret = None self.oauth_client_id = None self.installed_app_id = None self.refresh_token = None - self.location_id = None self.endpoints_initialized = False async def async_step_import(self, import_data: None) -> ConfigFlowResult: @@ -91,9 +92,11 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN): # Show the next screen return await self.async_step_pat() - async def async_step_pat(self, user_input=None): + async def async_step_pat( + self, user_input: dict[str, str] | None = None + ) -> ConfigFlowResult: """Get the Personal Access Token and validate it.""" - errors = {} + errors: dict[str, str] = {} if user_input is None or CONF_ACCESS_TOKEN not in user_input: return self._show_step_pat(errors) @@ -169,7 +172,9 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN): return await self.async_step_select_location() - async def async_step_select_location(self, user_input=None): + async def async_step_select_location( + self, user_input: dict[str, str] | None = None + ) -> ConfigFlowResult: """Ask user to select the location to setup.""" if user_input is None or CONF_LOCATION_ID not in user_input: # Get available locations @@ -196,7 +201,9 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN): await self.async_set_unique_id(format_unique_id(self.app_id, self.location_id)) return await self.async_step_authorize() - async def async_step_authorize(self, user_input=None): + async def async_step_authorize( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Wait for the user to authorize the app installation.""" user_input = {} if user_input is None else user_input self.installed_app_id = user_input.get(CONF_INSTALLED_APP_ID) @@ -233,7 +240,9 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN): }, ) - async def async_step_install(self, data=None): + async def async_step_install( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Create a config entry at completion of a flow and authorization of the app.""" data = { CONF_ACCESS_TOKEN: self.access_token, diff --git a/homeassistant/components/smarttub/config_flow.py b/homeassistant/components/smarttub/config_flow.py index 827375c907c..5caff953d6d 100644 --- a/homeassistant/components/smarttub/config_flow.py +++ b/homeassistant/components/smarttub/config_flow.py @@ -81,9 +81,13 @@ class SmartTubConfigFlow(ConfigFlow, domain=DOMAIN): ) return await self.async_step_reauth_confirm() - async def async_step_reauth_confirm(self, user_input=None): + async def async_step_reauth_confirm( + self, user_input: dict[str, str] | None = None + ) -> ConfigFlowResult: """Dialog that informs the user that reauth is required.""" if user_input is None: + if TYPE_CHECKING: + assert self._reauth_input is not None # same as DATA_SCHEMA but with default email data_schema = vol.Schema( { diff --git a/homeassistant/components/soma/config_flow.py b/homeassistant/components/soma/config_flow.py index 586567611f7..caf361d5c3c 100644 --- a/homeassistant/components/soma/config_flow.py +++ b/homeassistant/components/soma/config_flow.py @@ -39,7 +39,7 @@ class SomaFlowHandler(ConfigFlow, domain=DOMAIN): return await self.async_step_creation(user_input) - async def async_step_creation(self, user_input=None): + async def async_step_creation(self, user_input: dict[str, Any]) -> ConfigFlowResult: """Finish config flow.""" try: api = await self.hass.async_add_executor_job( diff --git a/homeassistant/components/somfy_mylink/config_flow.py b/homeassistant/components/somfy_mylink/config_flow.py index 231f93b0cb7..705db43362e 100644 --- a/homeassistant/components/somfy_mylink/config_flow.py +++ b/homeassistant/components/somfy_mylink/config_flow.py @@ -132,7 +132,7 @@ class OptionsFlowHandler(OptionsFlow): """Initialize options flow.""" self.config_entry = config_entry self.options = deepcopy(dict(config_entry.options)) - self._target_id = None + self._target_id: str | None = None @callback def _async_callback_targets(self): @@ -150,7 +150,9 @@ class OptionsFlowHandler(OptionsFlow): return cover["name"] raise KeyError - async def async_step_init(self, user_input=None): + async def async_step_init( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle options flow.""" if self.config_entry.state is not ConfigEntryState.LOADED: @@ -173,9 +175,13 @@ class OptionsFlowHandler(OptionsFlow): return self.async_show_form(step_id="init", data_schema=data_schema, errors={}) - async def async_step_target_config(self, user_input=None, target_id=None): + async def async_step_target_config( + self, user_input: dict[str, bool] | None = None, target_id: str | None = None + ) -> ConfigFlowResult: """Handle options flow for target.""" - reversed_target_ids = self.options.setdefault(CONF_REVERSED_TARGET_IDS, {}) + reversed_target_ids: dict[str | None, bool] = self.options.setdefault( + CONF_REVERSED_TARGET_IDS, {} + ) if user_input is not None: if user_input[CONF_REVERSE] != reversed_target_ids.get(self._target_id): diff --git a/homeassistant/components/songpal/config_flow.py b/homeassistant/components/songpal/config_flow.py index 9ccf7a8f19c..7f10d22b8c6 100644 --- a/homeassistant/components/songpal/config_flow.py +++ b/homeassistant/components/songpal/config_flow.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging -from typing import Any +from typing import TYPE_CHECKING, Any from urllib.parse import urlparse from songpal import Device, SongpalException @@ -21,7 +21,7 @@ _LOGGER = logging.getLogger(__name__) class SongpalConfig: """Device Configuration.""" - def __init__(self, name, host, endpoint): + def __init__(self, name: str, host: str | None, endpoint: str) -> None: """Initialize Configuration.""" self.name = name self.host = host @@ -33,12 +33,10 @@ class SongpalConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 - def __init__(self) -> None: - """Initialize the flow.""" - self.conf: SongpalConfig | None = None + conf: SongpalConfig async def async_step_user( - self, user_input: dict[str, Any] | None = None + self, user_input: dict[str, str] | None = None ) -> ConfigFlowResult: """Handle a flow initiated by the user.""" if user_input is None: @@ -75,7 +73,9 @@ class SongpalConfigFlow(ConfigFlow, domain=DOMAIN): return await self.async_step_init(user_input) - async def async_step_init(self, user_input=None): + async def async_step_init( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle a flow start.""" # Check if already configured self._async_abort_entries_match({CONF_ENDPOINT: self.conf.endpoint}) @@ -122,14 +122,16 @@ class SongpalConfigFlow(ConfigFlow, domain=DOMAIN): CONF_HOST: parsed_url.hostname, } + if TYPE_CHECKING: + assert isinstance(parsed_url.hostname, str) self.conf = SongpalConfig(friendly_name, parsed_url.hostname, endpoint) return await self.async_step_init() - async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult: + async def async_step_import(self, import_data: dict[str, str]) -> ConfigFlowResult: """Import a config entry.""" name = import_data.get(CONF_NAME) - endpoint = import_data.get(CONF_ENDPOINT) + endpoint = import_data[CONF_ENDPOINT] parsed_url = urlparse(endpoint) # Try to connect to test the endpoint diff --git a/homeassistant/components/soundtouch/config_flow.py b/homeassistant/components/soundtouch/config_flow.py index fea63366db9..7c637d71111 100644 --- a/homeassistant/components/soundtouch/config_flow.py +++ b/homeassistant/components/soundtouch/config_flow.py @@ -68,7 +68,9 @@ class SoundtouchConfigFlow(ConfigFlow, domain=DOMAIN): self.context["title_placeholders"] = {"name": self.name} return await self.async_step_zeroconf_confirm() - async def async_step_zeroconf_confirm(self, user_input=None): + async def async_step_zeroconf_confirm( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle user-confirmation of discovered node.""" if user_input is not None: return await self._async_create_soundtouch_entry() diff --git a/homeassistant/components/syncthru/config_flow.py b/homeassistant/components/syncthru/config_flow.py index 180ba0d9e34..1fb155a5648 100644 --- a/homeassistant/components/syncthru/config_flow.py +++ b/homeassistant/components/syncthru/config_flow.py @@ -64,7 +64,9 @@ class SyncThruConfigFlow(ConfigFlow, domain=DOMAIN): self.context["title_placeholders"] = {CONF_NAME: self.name} return await self.async_step_confirm() - async def async_step_confirm(self, user_input=None): + async def async_step_confirm( + self, user_input: dict[str, str] | None = None + ) -> ConfigFlowResult: """Handle discovery confirmation by user.""" if user_input is not None: return await self._async_check_and_create("confirm", user_input)