evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Remove incorrect use of ConfigType in config flows (#53544)

Browse source
This commit is contained in:
Milan Meulemans 2021-07-27 12:33:17 +02:00 • committed by GitHub
parent 0471b27179
commit 348d7a5622
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 83 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
homeassistant/components/bsblan/config_flow.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from bsblan import BSBLan, BSBLanError, Info
import voluptuous as vol
@ -10,7 +11,6 @@ from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_PORT, CONF_USERNAME
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import CONF_DEVICE_IDENT, CONF_PASSKEY, DOMAIN
@ -22,7 +22,9 @@ class BSBLanFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
if user_input is None:
return self._show_setup_form()

12
homeassistant/components/canary/config_flow.py
View file

@ -2,7 +2,7 @@
from __future__ import annotations
import logging
from typing import Final
from typing import Any, Final
from canary.api import Api
from requests.exceptions import ConnectTimeout, HTTPError
@ -24,7 +24,7 @@ from .const import (
_LOGGER: Final = logging.getLogger(__name__)
def validate_input(hass: HomeAssistant, data: ConfigType) -> bool:
def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> bool:
"""Validate the user input allows us to connect.
Data has the keys from DATA_SCHEMA with values provided by the user.
@ -56,7 +56,9 @@ class CanaryConfigFlow(ConfigFlow, domain=DOMAIN):
"""Handle a flow initiated by configuration file."""
return await self.async_step_user(user_input)
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
@ -104,7 +106,9 @@ class CanaryOptionsFlowHandler(OptionsFlow):
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage Canary options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

8
homeassistant/components/directv/config_flow.py
View file

@ -14,7 +14,7 @@ from homeassistant.const import CONF_HOST, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import CONF_RECEIVER_ID, DOMAIN
@ -45,7 +45,9 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
"""Set up the instance."""
self.discovery_info = {}
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
if user_input is None:
return self._show_setup_form()
@ -97,7 +99,7 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_ssdp_confirm()
async def async_step_ssdp_confirm(
self, user_input: ConfigType = None
self, user_input: dict[str, Any] = None
) -> FlowResult:
"""Handle a confirmation flow initiated by SSDP."""
if user_input is None:

16
homeassistant/components/esphome/config_flow.py
View file

@ -12,7 +12,7 @@ from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PASSWORD, CONF_PORT
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import DiscoveryInfoType
from . import DOMAIN, DomainData
@ -29,7 +29,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
self._password: str | None = None
async def _async_step_user_base(
self, user_input: ConfigType | None = None, error: str | None = None
self, user_input: dict[str, Any] | None = None, error: str | None = None
) -> FlowResult:
if user_input is not None:
return await self._async_authenticate_or_add(user_input)
@ -46,7 +46,9 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=vol.Schema(fields), errors=errors
)
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
return await self._async_step_user_base(user_input=user_input)
@ -59,14 +61,14 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
self.context[CONF_NAME] = value
self.context["title_placeholders"] = {"name": self._name}
def _set_user_input(self, user_input: ConfigType | None) -> None:
def _set_user_input(self, user_input: dict[str, Any] | None) -> None:
if user_input is None:
return
self._host = user_input[CONF_HOST]
self._port = user_input[CONF_PORT]
async def _async_authenticate_or_add(
self, user_input: ConfigType | None
self, user_input: dict[str, Any] | None
) -> FlowResult:
self._set_user_input(user_input)
error, device_info = await self.fetch_device_info()
@ -82,7 +84,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
return self._async_get_entry()
async def async_step_discovery_confirm(
self, user_input: ConfigType | None = None
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle user-confirmation of discovered node."""
if user_input is not None:
@ -154,7 +156,7 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
)
async def async_step_authenticate(
self, user_input: ConfigType | None = None, error: str | None = None
self, user_input: dict[str, Any] | None = None, error: str | None = None
) -> FlowResult:
"""Handle getting password for authentication."""
if user_input is not None:

17
homeassistant/components/hyperion/config_flow.py
View file

@ -28,7 +28,6 @@ from homeassistant.const import (
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType
from . import create_hyperion_client
from .const import (
@ -143,7 +142,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_reauth(
self,
config_data: ConfigType,
config_data: dict[str, Any],
) -> FlowResult:
"""Handle a reauthentication flow."""
self._data = dict(config_data)
@ -222,7 +221,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_user(
self,
user_input: ConfigType | None = None,
user_input: dict[str, Any] | None = None,
) -> FlowResult:
"""Handle a flow initiated by the user."""
errors = {}
@ -293,7 +292,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
async def async_step_auth(
self,
user_input: ConfigType | None = None,
user_input: dict[str, Any] | None = None,
) -> FlowResult:
"""Handle the auth step of a flow."""
errors = {}
@ -322,7 +321,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
)
async def async_step_create_token(
self, user_input: ConfigType | None = None
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Send a request for a new token."""
if user_input is None:
@ -348,7 +347,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
)
async def async_step_create_token_external(
self, auth_resp: ConfigType | None = None
self, auth_resp: dict[str, Any] | None = None
) -> FlowResult:
"""Handle completion of the request for a new token."""
if auth_resp is not None and client.ResponseOK(auth_resp):
@ -361,7 +360,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_external_step_done(next_step_id="create_token_fail")
async def async_step_create_token_success(
self, _: ConfigType | None = None
self, _: dict[str, Any] | None = None
) -> FlowResult:
"""Create an entry after successful token creation."""
# Clean-up the request task.
@ -377,7 +376,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_confirm()
async def async_step_create_token_fail(
self, _: ConfigType | None = None
self, _: dict[str, Any] | None = None
) -> FlowResult:
"""Show an error on the auth form."""
# Clean-up the request task.
@ -385,7 +384,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="auth_new_token_not_granted_error")
async def async_step_confirm(
self, user_input: ConfigType | None = None
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Get final confirmation before entry creation."""
if user_input is None and self._require_confirm:

4
homeassistant/components/iaqualink/config_flow.py
View file

@ -1,6 +1,8 @@
"""Config flow to configure zone component."""
from __future__ import annotations
from typing import Any
from iaqualink import AqualinkClient, AqualinkLoginException
import voluptuous as vol
@ -16,7 +18,7 @@ class AqualinkFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
async def async_step_user(self, user_input: ConfigType | None = None):
async def async_step_user(self, user_input: dict[str, Any] | None = None):
"""Handle a flow start."""
# Supporting a single account.
entries = self._async_current_entries()

9
homeassistant/components/ipp/config_flow.py
View file

@ -26,7 +26,6 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import CONF_BASE_PATH, CONF_SERIAL, CONF_UUID, DOMAIN
@ -62,7 +61,9 @@ class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
"""Set up the instance."""
self.discovery_info = {}
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
if user_input is None:
return self._show_setup_form()
@ -98,7 +99,7 @@ class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=user_input[CONF_HOST], data=user_input)
async def async_step_zeroconf(self, discovery_info: ConfigType) -> FlowResult:
async def async_step_zeroconf(self, discovery_info: dict[str, Any]) -> FlowResult:
"""Handle zeroconf discovery."""
port = discovery_info[CONF_PORT]
zctype = discovery_info["type"]
@ -165,7 +166,7 @@ class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
return await self.async_step_zeroconf_confirm()
async def async_step_zeroconf_confirm(
self, user_input: ConfigType = None
self, user_input: dict[str, Any] = None
) -> FlowResult:
"""Handle a confirmation flow initiated by zeroconf."""
if user_input is None:

13
homeassistant/components/keenetic_ndms2/config_flow.py
View file

@ -1,6 +1,7 @@
"""Config flow for Keenetic NDMS2."""
from __future__ import annotations
from typing import Any
from urllib.parse import urlparse
from ndms2_client import Client, ConnectionException, InterfaceInfo, TelnetConnection
@ -50,7 +51,9 @@ class KeeneticFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler."""
return KeeneticOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user."""
errors = {}
if user_input is not None:
@ -135,7 +138,9 @@ class KeeneticOptionsFlowHandler(config_entries.OptionsFlow):
self.config_entry = config_entry
self._interface_options = {}
async def async_step_init(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the options."""
router: KeeneticRouter = self.hass.data[DOMAIN][self.config_entry.entry_id][
ROUTER
@ -152,7 +157,9 @@ class KeeneticOptionsFlowHandler(config_entries.OptionsFlow):
}
return await self.async_step_user()
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Manage the device tracker options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

6
homeassistant/components/nzbget/config_flow.py
View file

@ -75,7 +75,9 @@ class NZBGetConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_user(user_input)
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
@ -129,7 +131,7 @@ class NZBGetOptionsFlowHandler(OptionsFlow):
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input: ConfigType | None = None):
async def async_step_init(self, user_input: dict[str, Any] | None = None):
"""Manage NZBGet options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

5
homeassistant/components/plum_lightpad/config_flow.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from aiohttp import ContentTypeError
from requests.exceptions import ConnectTimeout, HTTPError
@ -35,7 +36,9 @@ class PlumLightpadConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors=errors or {},
)
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user or redirected to by import."""
if not user_input:
return self._show_form()

23
homeassistant/components/sia/config_flow.py
View file

@ -19,7 +19,6 @@ from homeassistant import config_entries
from homeassistant.const import CONF_PORT, CONF_PROTOCOL
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_ACCOUNT,
@ -62,7 +61,7 @@ ACCOUNT_SCHEMA = vol.Schema(
DEFAULT_OPTIONS = {CONF_IGNORE_TIMESTAMPS: False, CONF_ZONES: None}
def validate_input(data: ConfigType) -> dict[str, str] | None:
def validate_input(data: dict[str, Any]) -> dict[str, str] | None:
"""Validate the input by the user."""
try:
SIAAccount.validate_account(data[CONF_ACCOUNT], data.get(CONF_ENCRYPTION_KEY))
@ -82,7 +81,7 @@ def validate_input(data: ConfigType) -> dict[str, str] | None:
return validate_zones(data)
def validate_zones(data: ConfigType) -> dict[str, str] | None:
def validate_zones(data: dict[str, Any]) -> dict[str, str] | None:
"""Validate the zones field."""
if data[CONF_ZONES] == 0:
return {"base": "invalid_zones"}
@ -102,10 +101,10 @@ class SIAConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
def __init__(self):
"""Initialize the config flow."""
self._data: ConfigType = {}
self._data: dict[str, Any] = {}
self._options: Mapping[str, Any] = {CONF_ACCOUNTS: {}}
async def async_step_user(self, user_input: ConfigType = None) -> FlowResult:
async def async_step_user(self, user_input: dict[str, Any] = None) -> FlowResult:
"""Handle the initial user step."""
errors: dict[str, str] | None = None
if user_input is not None:
@ -116,7 +115,9 @@ class SIAConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
return await self.async_handle_data_and_route(user_input)
async def async_step_add_account(self, user_input: ConfigType = None) -> FlowResult:
async def async_step_add_account(
self, user_input: dict[str, Any] = None
) -> FlowResult:
"""Handle the additional accounts steps."""
errors: dict[str, str] | None = None
if user_input is not None:
@ -127,7 +128,9 @@ class SIAConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
)
return await self.async_handle_data_and_route(user_input)
async def async_handle_data_and_route(self, user_input: ConfigType) -> FlowResult:
async def async_handle_data_and_route(
self, user_input: dict[str, Any]
) -> FlowResult:
"""Handle the user_input, check if configured and route to the right next step or create entry."""
self._update_data(user_input)
@ -141,7 +144,7 @@ class SIAConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
options=self._options,
)
def _update_data(self, user_input: ConfigType) -> None:
def _update_data(self, user_input: dict[str, Any]) -> None:
"""Parse the user_input and store in data and options attributes.
If there is a port in the input or no data, assume it is fully new and overwrite.
@ -175,7 +178,7 @@ class SIAOptionsFlowHandler(config_entries.OptionsFlow):
self.hub: SIAHub | None = None
self.accounts_todo: list = []
async def async_step_init(self, user_input: ConfigType = None) -> FlowResult:
async def async_step_init(self, user_input: dict[str, Any] = None) -> FlowResult:
"""Manage the SIA options."""
self.hub = self.hass.data[DOMAIN][self.config_entry.entry_id]
assert self.hub is not None
@ -183,7 +186,7 @@ class SIAOptionsFlowHandler(config_entries.OptionsFlow):
self.accounts_todo = [a.account_id for a in self.hub.sia_accounts]
return await self.async_step_options()
async def async_step_options(self, user_input: ConfigType = None) -> FlowResult:
async def async_step_options(self, user_input: dict[str, Any] = None) -> FlowResult:
"""Create the options step for a account."""
errors: dict[str, str] | None = None
if user_input is not None:

11
homeassistant/components/sonarr/config_flow.py
View file

@ -18,7 +18,6 @@ from homeassistant.const import (
from homeassistant.core import HomeAssistant, callback
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import (
CONF_BASE_PATH,
@ -75,7 +74,7 @@ class SonarrConfigFlow(ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler."""
return SonarrOptionsFlowHandler(config_entry)
async def async_step_reauth(self, data: ConfigType | None = None) -> FlowResult:
async def async_step_reauth(self, data: dict[str, Any] | None = None) -> FlowResult:
"""Handle configuration by re-auth."""
self._reauth = True
self._entry_data = dict(data)
@ -85,7 +84,7 @@ class SonarrConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_reauth_confirm()
async def async_step_reauth_confirm(
self, user_input: ConfigType | None = None
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Confirm reauth dialog."""
if user_input is None:
@ -98,7 +97,9 @@ class SonarrConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_user()
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult:
async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initiated by the user."""
errors = {}
@ -171,7 +172,7 @@ class SonarrOptionsFlowHandler(OptionsFlow):
"""Initialize options flow."""
self.config_entry = config_entry
async def async_step_init(self, user_input: ConfigType | None = None):
async def async_step_init(self, user_input: dict[str, Any] | None = None):
"""Manage Sonarr options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)

4
homeassistant/components/system_bridge/config_flow.py
View file

@ -17,7 +17,7 @@ from homeassistant.const import CONF_API_KEY, CONF_HOST, CONF_PORT
from homeassistant.core import HomeAssistant
from homeassistant.data_entry_flow import FlowResult
from homeassistant.helpers import aiohttp_client, config_validation as cv
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import BRIDGE_CONNECTION_ERRORS, DOMAIN
@ -167,7 +167,7 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_authenticate()
async def async_step_reauth(self, entry_data: ConfigType) -> FlowResult:
async def async_step_reauth(self, entry_data: dict[str, Any]) -> FlowResult:
"""Perform reauth upon an API authentication error."""
self._name = entry_data[CONF_HOST]
self._input = {

4
homeassistant/components/yamaha_musiccast/config_flow.py
View file

@ -2,6 +2,7 @@
from __future__ import annotations
import logging
from typing import Any
from urllib.parse import urlparse
from aiohttp import ClientConnectorError
@ -13,7 +14,6 @@ from homeassistant.components import ssdp
from homeassistant.config_entries import ConfigFlow
from homeassistant.const import CONF_HOST
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from .const import DOMAIN
@ -29,7 +29,7 @@ class MusicCastFlowHandler(ConfigFlow, domain=DOMAIN):
host: str
async def async_step_user(
self, user_input: ConfigType | None = None
self, user_input: dict[str, Any] | None = None
) -> data_entry_flow.FlowResult:
"""Handle a flow initiated by the user."""
# Request user input, unless we are preparing discovery flow