From 47beddc6c6d3dc01de9544781c97fa5734bbc28b Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Wed, 21 Aug 2024 22:43:42 +0200 Subject: [PATCH] Improve config flow type hints (part 4) (#124348) --- .../components/plaato/config_flow.py | 21 ++++++++++++++----- homeassistant/components/plex/config_flow.py | 14 ++++++++----- .../components/progettihwsw/config_flow.py | 10 ++++++--- homeassistant/components/ps4/config_flow.py | 11 ++++++---- homeassistant/components/roon/config_flow.py | 19 ++++++++++------- .../components/smarttub/config_flow.py | 8 +++++-- .../components/somfy_mylink/config_flow.py | 13 +++++++----- .../components/soundtouch/config_flow.py | 9 +++++--- .../components/subaru/config_flow.py | 16 ++++++++++---- 9 files changed, 82 insertions(+), 39 deletions(-) diff --git a/homeassistant/components/plaato/config_flow.py b/homeassistant/components/plaato/config_flow.py index 1240abc5e81..3ada4fdc312 100644 --- a/homeassistant/components/plaato/config_flow.py +++ b/homeassistant/components/plaato/config_flow.py @@ -2,11 +2,18 @@ from __future__ import annotations +from typing import Any + from pyplaato.plaato import PlaatoDeviceType import voluptuous as vol from homeassistant.components import cloud, webhook -from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow +from homeassistant.config_entries import ( + ConfigEntry, + ConfigFlow, + ConfigFlowResult, + OptionsFlow, +) from homeassistant.const import CONF_SCAN_INTERVAL, CONF_TOKEN, CONF_WEBHOOK_ID from homeassistant.core import callback import homeassistant.helpers.config_validation as cv @@ -31,11 +38,13 @@ class PlaatoConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 - def __init__(self): + def __init__(self) -> None: """Initialize.""" - self._init_info = {} + self._init_info: dict[str, Any] = {} - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle user step.""" if user_input is not None: @@ -185,7 +194,9 @@ class PlaatoOptionsFlowHandler(OptionsFlow): return await self.async_step_user() - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Manage the options.""" if user_input is not None: return self.async_create_entry(title="", data=user_input) diff --git a/homeassistant/components/plex/config_flow.py b/homeassistant/components/plex/config_flow.py index 374067c94cd..7162e517e23 100644 --- a/homeassistant/components/plex/config_flow.py +++ b/homeassistant/components/plex/config_flow.py @@ -5,7 +5,7 @@ from __future__ import annotations from collections.abc import Mapping import copy import logging -from typing import Any +from typing import TYPE_CHECKING, Any from aiohttp import web_response import plexapi.exceptions @@ -105,15 +105,15 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN): """Get the options flow for this handler.""" return PlexOptionsFlowHandler(config_entry) - def __init__(self): + def __init__(self) -> None: """Initialize the Plex flow.""" - self.current_login = {} + self.current_login: dict[str, Any] = {} self.available_servers = None self.plexauth = None self.token = None self.client_id = None self._manual = False - self._reauth_config = None + self._reauth_config: dict[str, Any] | None = None async def async_step_user(self, user_input=None, errors=None): """Handle a flow initialized by the user.""" @@ -184,7 +184,9 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN): step_id="manual_setup", data_schema=data_schema, errors=errors ) - async def async_step_server_validate(self, server_config): + async def async_step_server_validate( + self, server_config: dict[str, Any] + ) -> ConfigFlowResult: """Validate a provided configuration.""" if self._reauth_config: server_config = {**self._reauth_config, **server_config} @@ -249,6 +251,8 @@ class PlexFlowHandler(ConfigFlow, domain=DOMAIN): entry = await self.async_set_unique_id(server_id) if self.context[CONF_SOURCE] == SOURCE_REAUTH: + if TYPE_CHECKING: + assert entry self.hass.config_entries.async_update_entry(entry, data=data) _LOGGER.debug("Updated config entry for %s", plex_server.friendly_name) await self.hass.config_entries.async_reload(entry.entry_id) diff --git a/homeassistant/components/progettihwsw/config_flow.py b/homeassistant/components/progettihwsw/config_flow.py index dbe12184a10..95596b940a4 100644 --- a/homeassistant/components/progettihwsw/config_flow.py +++ b/homeassistant/components/progettihwsw/config_flow.py @@ -1,9 +1,11 @@ """Config flow for ProgettiHWSW Automation integration.""" +from typing import Any + from ProgettiHWSW.ProgettiHWSWAPI import ProgettiHWSWAPI import voluptuous as vol -from homeassistant.config_entries import ConfigFlow +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError @@ -38,7 +40,7 @@ class ProgettiHWSWConfigFlow(ConfigFlow, domain=DOMAIN): def __init__(self) -> None: """Initialize class variables.""" - self.s1_in = None + self.s1_in: dict[str, Any] | None = None async def async_step_relay_modes(self, user_input=None): """Manage relay modes step.""" @@ -66,7 +68,9 @@ class ProgettiHWSWConfigFlow(ConfigFlow, domain=DOMAIN): errors=errors, ) - async def async_step_user(self, user_input=None): + 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: diff --git a/homeassistant/components/ps4/config_flow.py b/homeassistant/components/ps4/config_flow.py index b842c2f7cfb..cdbf02dcc90 100644 --- a/homeassistant/components/ps4/config_flow.py +++ b/homeassistant/components/ps4/config_flow.py @@ -1,13 +1,14 @@ """Config Flow for PlayStation 4.""" from collections import OrderedDict +from typing import Any from pyps4_2ndscreen.errors import CredentialTimeout from pyps4_2ndscreen.helpers import Helper from pyps4_2ndscreen.media_art import COUNTRIES import voluptuous as vol -from homeassistant.config_entries import ConfigFlow +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult from homeassistant.const import ( CONF_CODE, CONF_HOST, @@ -44,7 +45,7 @@ class PlayStation4FlowHandler(ConfigFlow, domain=DOMAIN): VERSION = CONFIG_ENTRY_VERSION - def __init__(self): + def __init__(self) -> None: """Initialize the config flow.""" self.helper = Helper() self.creds = None @@ -54,9 +55,11 @@ class PlayStation4FlowHandler(ConfigFlow, domain=DOMAIN): self.pin = None self.m_device = None self.location = None - self.device_list = [] + self.device_list: list[str] = [] - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle a user config flow.""" # Check if able to bind to ports: UDP 987, TCP 997. ports = PORT_MSG.keys() diff --git a/homeassistant/components/roon/config_flow.py b/homeassistant/components/roon/config_flow.py index f555cc52dd1..de220454852 100644 --- a/homeassistant/components/roon/config_flow.py +++ b/homeassistant/components/roon/config_flow.py @@ -2,11 +2,12 @@ import asyncio import logging +from typing import Any from roonapi import RoonApi, RoonDiscovery import voluptuous as vol -from homeassistant.config_entries import ConfigFlow +from homeassistant.config_entries import ConfigFlow, ConfigFlowResult from homeassistant.const import CONF_API_KEY, CONF_HOST, CONF_PORT from homeassistant.core import HomeAssistant from homeassistant.exceptions import HomeAssistantError @@ -36,14 +37,14 @@ TIMEOUT = 120 class RoonHub: """Interact with roon during config flow.""" - def __init__(self, hass): + def __init__(self, hass: HomeAssistant) -> None: """Initialise the RoonHub.""" self._hass = hass - async def discover(self): + async def discover(self) -> list[tuple[str, int]]: """Try and discover roon servers.""" - def get_discovered_servers(discovery): + def get_discovered_servers(discovery: RoonDiscovery) -> list[tuple[str, int]]: servers = discovery.all() discovery.stop() return servers @@ -93,7 +94,7 @@ class RoonHub: return (token, core_id, core_name) -async def discover(hass): +async def discover(hass: HomeAssistant) -> list[tuple[str, int]]: """Connect and authenticate home assistant.""" hub = RoonHub(hass) @@ -122,13 +123,15 @@ class RoonConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 - def __init__(self): + def __init__(self) -> None: """Initialize the Roon flow.""" self._host = None self._port = None - self._servers = [] + self._servers: list[tuple[str, int]] = [] - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Get roon core details via discovery.""" self._servers = await discover(self.hass) diff --git a/homeassistant/components/smarttub/config_flow.py b/homeassistant/components/smarttub/config_flow.py index 60f14b03e45..827375c907c 100644 --- a/homeassistant/components/smarttub/config_flow.py +++ b/homeassistant/components/smarttub/config_flow.py @@ -3,7 +3,7 @@ from __future__ import annotations from collections.abc import Mapping -from typing import Any +from typing import TYPE_CHECKING, Any from smarttub import LoginFailed import voluptuous as vol @@ -30,7 +30,9 @@ class SmartTubConfigFlow(ConfigFlow, domain=DOMAIN): self._reauth_input: Mapping[str, Any] | None = None self._reauth_entry: ConfigEntry | None = None - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle a flow initiated by the user.""" errors = {} @@ -53,6 +55,8 @@ class SmartTubConfigFlow(ConfigFlow, domain=DOMAIN): ) # this is a reauth attempt + if TYPE_CHECKING: + assert self._reauth_entry if self._reauth_entry.unique_id != self.unique_id: # there is a config entry matching this account, # but it is not the one we were trying to reauth diff --git a/homeassistant/components/somfy_mylink/config_flow.py b/homeassistant/components/somfy_mylink/config_flow.py index a13f036210d..9a8b5d76d3f 100644 --- a/homeassistant/components/somfy_mylink/config_flow.py +++ b/homeassistant/components/somfy_mylink/config_flow.py @@ -4,6 +4,7 @@ from __future__ import annotations from copy import deepcopy import logging +from typing import Any from somfy_mylink_synergy import SomfyMyLinkSynergy import voluptuous as vol @@ -61,11 +62,11 @@ class SomfyConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 - def __init__(self): + def __init__(self) -> None: """Initialize the somfy_mylink flow.""" - self.host = None - self.mac = None - self.ip_address = None + self.host: str | None = None + self.mac: str | None = None + self.ip_address: str | None = None async def async_step_dhcp( self, discovery_info: dhcp.DhcpServiceInfo @@ -82,7 +83,9 @@ class SomfyConfigFlow(ConfigFlow, domain=DOMAIN): self.context["title_placeholders"] = {"ip": self.ip_address, "mac": self.mac} return await self.async_step_user() - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle the initial step.""" errors = {} diff --git a/homeassistant/components/soundtouch/config_flow.py b/homeassistant/components/soundtouch/config_flow.py index c8e8ce945db..fea63366db9 100644 --- a/homeassistant/components/soundtouch/config_flow.py +++ b/homeassistant/components/soundtouch/config_flow.py @@ -1,6 +1,7 @@ """Config flow for Bose SoundTouch integration.""" import logging +from typing import Any from libsoundtouch import soundtouch_device from requests import RequestException @@ -21,12 +22,14 @@ class SoundtouchConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 - def __init__(self): + def __init__(self) -> None: """Initialize a new SoundTouch config flow.""" - self.host = None + self.host: str | None = None self.name = None - async def async_step_user(self, user_input=None): + async def async_step_user( + self, user_input: dict[str, Any] | None = None + ) -> ConfigFlowResult: """Handle a flow initiated by the user.""" errors = {} diff --git a/homeassistant/components/subaru/config_flow.py b/homeassistant/components/subaru/config_flow.py index 5ecaf9670d7..3d96a89a14f 100644 --- a/homeassistant/components/subaru/config_flow.py +++ b/homeassistant/components/subaru/config_flow.py @@ -4,7 +4,7 @@ from __future__ import annotations from datetime import datetime import logging -from typing import Any +from typing import TYPE_CHECKING, Any from subarulink import ( Controller as SubaruAPI, @@ -44,10 +44,10 @@ class SubaruConfigFlow(ConfigFlow, domain=DOMAIN): VERSION = 1 - def __init__(self): + def __init__(self) -> None: """Initialize config flow.""" - self.config_data = {CONF_PIN: None} - self.controller = None + self.config_data: dict[str, Any] = {CONF_PIN: None} + self.controller: SubaruAPI | None = None async def async_step_user( self, user_input: dict[str, Any] | None = None @@ -66,6 +66,8 @@ class SubaruConfigFlow(ConfigFlow, domain=DOMAIN): _LOGGER.error("Unable to communicate with Subaru API: %s", ex.message) return self.async_abort(reason="cannot_connect") else: + if TYPE_CHECKING: + assert self.controller if not self.controller.device_registered: _LOGGER.debug("2FA validation is required") return await self.async_step_two_factor() @@ -137,6 +139,8 @@ class SubaruConfigFlow(ConfigFlow, domain=DOMAIN): ) -> ConfigFlowResult: """Select contact method and request 2FA code from Subaru.""" error = None + if TYPE_CHECKING: + assert self.controller if user_input: # self.controller.contact_methods is a dict: # {"phone":"555-555-5555", "userName":"my@email.com"} @@ -165,6 +169,8 @@ class SubaruConfigFlow(ConfigFlow, domain=DOMAIN): ) -> ConfigFlowResult: """Validate received 2FA code with Subaru.""" error = None + if TYPE_CHECKING: + assert self.controller if user_input: try: vol.Match(r"^[0-9]{6}$")(user_input[CONF_VALIDATION_CODE]) @@ -190,6 +196,8 @@ class SubaruConfigFlow(ConfigFlow, domain=DOMAIN): ) -> ConfigFlowResult: """Handle second part of config flow, if required.""" error = None + if TYPE_CHECKING: + assert self.controller if user_input and self.controller.update_saved_pin(user_input[CONF_PIN]): try: vol.Match(r"[0-9]{4}")(user_input[CONF_PIN])