starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-06 02:35:01 -05:00
Code Issues Packages Projects Releases Wiki Activity

Improve config flow type hints (n-p) (#124909)

Browse Source
This commit is contained in:
epenet
2024-08-30 11:04:58 +02:00
committed by GitHub
parent 4940968cd5
commit 6833af6286
10 changed files with 81 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/netgear/config_flow.py
+8 -2
View File
@@ -67,7 +67,9 @@ class OptionsFlowHandler(OptionsFlow):
"""Init object."""
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:
"""Manage the options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
@@ -109,7 +111,11 @@ class NetgearFlowHandler(ConfigFlow, domain=DOMAIN):
"""Get the options flow."""
return OptionsFlowHandler(config_entry)
async def _show_setup_form(self, user_input=None, errors=None):
async def _show_setup_form(
self,
user_input: dict[str, Any] | None = None,
errors: dict[str, str] | None = None,
) -> ConfigFlowResult:
"""Show the setup form to the user."""
if not user_input:
user_input = {}
homeassistant/components/nuki/config_flow.py
+8 -3
View File
@@ -12,6 +12,7 @@ import voluptuous as vol
from homeassistant.components import dhcp
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_TOKEN
from homeassistant.core import HomeAssistant
from .const import CONF_ENCRYPT_TOKEN, DEFAULT_PORT, DEFAULT_TIMEOUT, DOMAIN
from .helpers import CannotConnect, InvalidAuth, parse_id
@@ -34,7 +35,7 @@ REAUTH_SCHEMA = vol.Schema(
)
async def validate_input(hass, data):
async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
"""Validate the user input allows us to connect.
Data has the keys from USER_SCHEMA with values provided by the user.
@@ -99,7 +100,9 @@ class NukiConfigFlow(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, Any] | None = None
) -> ConfigFlowResult:
"""Dialog that inform the user that reauth is required."""
errors = {}
if user_input is None:
@@ -140,7 +143,9 @@ class NukiConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="reauth_confirm", data_schema=REAUTH_SCHEMA, errors=errors
)
async def async_step_validate(self, user_input=None):
async def async_step_validate(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle init step of a flow."""
data_schema = self.discovery_schema or USER_SCHEMA
homeassistant/components/octoprint/config_flow.py
+11 -7
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import asyncio
from collections.abc import Mapping
import logging
from typing import Any
from typing import TYPE_CHECKING, Any
import aiohttp
from pyoctoprintapi import ApiError, OctoprintClient, OctoprintException
@@ -104,7 +104,9 @@ class OctoPrintConfigFlow(ConfigFlow, domain=DOMAIN):
self._user_input = user_input
return await self.async_step_get_api_key()
async def async_step_get_api_key(self, user_input=None):
async def async_step_get_api_key(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Get an Application Api Key."""
if not self.api_key_task:
self.api_key_task = self.hass.async_create_task(
@@ -130,7 +132,7 @@ class OctoPrintConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_show_progress_done(next_step_id="user")
async def _finish_config(self, user_input: dict):
async def _finish_config(self, user_input: dict[str, Any]) -> ConfigFlowResult:
"""Finish the configuration setup."""
existing_entry = await self.async_set_unique_id(self.unique_id)
if existing_entry is not None:
@@ -156,7 +158,7 @@ class OctoPrintConfigFlow(ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=user_input[CONF_HOST], data=user_input)
async def async_step_auth_failed(self, user_input):
async def async_step_auth_failed(self, user_input: None) -> ConfigFlowResult:
"""Handle api fetch failure."""
return self.async_abort(reason="auth_failed")
@@ -252,15 +254,17 @@ class OctoPrintConfigFlow(ConfigFlow, domain=DOMAIN):
self._user_input = self._reauth_data
return await self.async_step_get_api_key()
async def _async_get_auth_key(self):
async def _async_get_auth_key(self) -> None:
"""Get application api key."""
if TYPE_CHECKING:
assert self._user_input is not None
octoprint = self._get_octoprint_client(self._user_input)
self._user_input[CONF_API_KEY] = await octoprint.request_app_key(
"Home Assistant", self._user_input[CONF_USERNAME], 300
)
def _get_octoprint_client(self, user_input: dict) -> OctoprintClient:
def _get_octoprint_client(self, user_input: dict[str, Any]) -> OctoprintClient:
"""Build an octoprint client from the user_input."""
verify_ssl = user_input.get(CONF_VERIFY_SSL, True)
@@ -281,7 +285,7 @@ class OctoPrintConfigFlow(ConfigFlow, domain=DOMAIN):
path=user_input[CONF_PATH],
)
def async_remove(self):
def async_remove(self) -> None:
"""Detach the session."""
for session in self._sessions:
session.detach()
homeassistant/components/omnilogic/config_flow.py
+3 -1
View File
@@ -88,7 +88,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, Any] | None = None
) -> ConfigFlowResult:
"""Manage options."""
if user_input is not None:
homeassistant/components/onvif/config_flow.py
+7 -3
View File
@@ -198,7 +198,9 @@ class OnvifFlowHandler(ConfigFlow, domain=DOMAIN):
hass.async_create_task(self.hass.config_entries.async_reload(entry_id))
return self.async_abort(reason="already_configured")
async def async_step_device(self, user_input=None):
async def async_step_device(
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Handle WS-Discovery.
Let user choose between discovered devices and manual configuration.
@@ -395,11 +397,13 @@ class OnvifOptionsFlowHandler(OptionsFlow):
self.config_entry = config_entry
self.options = dict(config_entry.options)
async def async_step_init(self, user_input=None):
async def async_step_init(self, user_input: None = None) -> ConfigFlowResult:
"""Manage the ONVIF options."""
return await self.async_step_onvif_devices()
async def async_step_onvif_devices(self, user_input=None):
async def async_step_onvif_devices(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the ONVIF devices options."""
if user_input is not None:
self.options[CONF_EXTRA_ARGUMENTS] = user_input[CONF_EXTRA_ARGUMENTS]
homeassistant/components/opentherm_gw/config_flow.py
+7 -3
View File
@@ -50,7 +50,9 @@ class OpenThermGwConfigFlow(ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler."""
return OpenThermGwOptionsFlow(config_entry)
async def async_step_init(self, info=None):
async def async_step_init(
self, info: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle config flow initiation."""
if info:
name = info[CONF_NAME]
@@ -104,7 +106,7 @@ class OpenThermGwConfigFlow(ConfigFlow, domain=DOMAIN):
}
return await self.async_step_init(info=formatted_config)
def _show_form(self, errors=None):
def _show_form(self, errors: dict[str, str] | None = None) -> ConfigFlowResult:
"""Show the config flow form with possible errors."""
return self.async_show_form(
step_id="init",
@@ -132,7 +134,9 @@ class OpenThermGwOptionsFlow(OptionsFlow):
"""Initialize the 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, Any] | None = None
) -> ConfigFlowResult:
"""Manage the opentherm_gw options."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
homeassistant/components/plaato/config_flow.py
+12 -6
View File
@@ -71,7 +71,9 @@ class PlaatoConfigFlow(ConfigFlow, domain=DOMAIN):
),
)
async def async_step_api_method(self, user_input=None):
async def async_step_api_method(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Handle device type step."""
device_type = self._init_info[CONF_DEVICE_TYPE]
@@ -90,7 +92,9 @@ class PlaatoConfigFlow(ConfigFlow, domain=DOMAIN):
return await self._show_api_method_form(device_type)
async def async_step_webhook(self, user_input=None):
async def async_step_webhook(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Validate config step."""
use_webhook = self._init_info[CONF_USE_WEBHOOK]
@@ -136,8 +140,8 @@ class PlaatoConfigFlow(ConfigFlow, domain=DOMAIN):
)
async def _show_api_method_form(
self, device_type: PlaatoDeviceType, errors: dict | None = None
):
self, device_type: PlaatoDeviceType, errors: dict[str, str] | None = None
) -> ConfigFlowResult:
data_schema = vol.Schema({vol.Optional(CONF_TOKEN, default=""): str})
if device_type == PlaatoDeviceType.Airlock:
@@ -186,7 +190,7 @@ class PlaatoOptionsFlowHandler(OptionsFlow):
self._config_entry = config_entry
async def async_step_init(self, user_input=None):
async def async_step_init(self, user_input: None = None) -> ConfigFlowResult:
"""Manage the options."""
use_webhook = self._config_entry.data.get(CONF_USE_WEBHOOK, False)
if use_webhook:
@@ -215,7 +219,9 @@ class PlaatoOptionsFlowHandler(OptionsFlow):
),
)
async def async_step_webhook(self, user_input=None):
async def async_step_webhook(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options for webhook device."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
homeassistant/components/progettihwsw/config_flow.py
+7 -3
View File
@@ -1,6 +1,6 @@
"""Config flow for ProgettiHWSW Automation integration."""
from typing import Any
from typing import TYPE_CHECKING, Any
from ProgettiHWSW.ProgettiHWSWAPI import ProgettiHWSWAPI
import voluptuous as vol
@@ -42,9 +42,13 @@ class ProgettiHWSWConfigFlow(ConfigFlow, domain=DOMAIN):
"""Initialize class variables."""
self.s1_in: dict[str, Any] | None = None
async def async_step_relay_modes(self, user_input=None):
async def async_step_relay_modes(
self, user_input: dict[str, str] | None = None
) -> ConfigFlowResult:
"""Manage relay modes step."""
errors = {}
errors: dict[str, str] = {}
if TYPE_CHECKING:
assert self.s1_in is not None
if user_input is not None:
whole_data = user_input
whole_data.update(self.s1_in)
homeassistant/components/prosegur/config_flow.py
+4 -2
View File
@@ -116,9 +116,11 @@ class ProsegurConfigFlow(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:
"""Handle re-authentication with Prosegur."""
errors = {}
errors: dict[str, str] = {}
if user_input:
try:
homeassistant/components/ps4/config_flow.py
+14 -8
View File
@@ -48,13 +48,13 @@ class PlayStation4FlowHandler(ConfigFlow, domain=DOMAIN):
def __init__(self) -> None:
"""Initialize the config flow."""
self.helper = Helper()
self.creds = None
self.creds: str | None = None
self.name = None
self.host = None
self.region = None
self.pin = None
self.pin: str | None = None
self.m_device = None
self.location = None
self.location: location.LocationInfo | None = None
self.device_list: list[str] = []
async def async_step_user(
@@ -69,7 +69,9 @@ class PlayStation4FlowHandler(ConfigFlow, domain=DOMAIN):
return self.async_abort(reason=reason)
return await self.async_step_creds()
async def async_step_creds(self, user_input=None):
async def async_step_creds(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Return PS4 credentials from 2nd Screen App."""
errors = {}
if user_input is not None:
@@ -85,7 +87,9 @@ class PlayStation4FlowHandler(ConfigFlow, domain=DOMAIN):
return self.async_show_form(step_id="creds", errors=errors)
async def async_step_mode(self, user_input=None):
async def async_step_mode(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Prompt for mode."""
errors = {}
mode = [CONF_AUTO, CONF_MANUAL]
@@ -100,7 +104,7 @@ class PlayStation4FlowHandler(ConfigFlow, domain=DOMAIN):
if not errors:
return await self.async_step_link()
mode_schema = OrderedDict()
mode_schema = OrderedDict[vol.Marker, Any]()
mode_schema[vol.Required(CONF_MODE, default=CONF_AUTO)] = vol.In(list(mode))
mode_schema[vol.Optional(CONF_IP_ADDRESS)] = str
@@ -108,7 +112,9 @@ class PlayStation4FlowHandler(ConfigFlow, domain=DOMAIN):
step_id="mode", data_schema=vol.Schema(mode_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:
"""Prompt user input. Create or edit entry."""
regions = sorted(COUNTRIES.keys())
default_region = None
@@ -193,7 +199,7 @@ class PlayStation4FlowHandler(ConfigFlow, domain=DOMAIN):
default_region = country
# Show User Input form.
link_schema = OrderedDict()
link_schema = OrderedDict[vol.Marker, Any]()
link_schema[vol.Required(CONF_IP_ADDRESS)] = vol.In(list(self.device_list))
link_schema[vol.Required(CONF_REGION, default=default_region)] = vol.In(
list(regions)