Improve config flow type hints (part 3) (#124346)

pull/124582/head
epenet 2024-08-25 18:36:03 +02:00 committed by GitHub
parent 909dfcc436
commit 58b7711bdd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 69 additions and 42 deletions

View File

@ -1,9 +1,11 @@
"""Config flow for growatt server integration.""" """Config flow for growatt server integration."""
from typing import Any
import growattServer import growattServer
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigFlow from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_URL, CONF_USERNAME from homeassistant.const import CONF_NAME, CONF_PASSWORD, CONF_URL, CONF_USERNAME
from homeassistant.core import callback from homeassistant.core import callback
@ -21,11 +23,11 @@ class GrowattServerConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialise growatt server flow.""" """Initialise growatt server flow."""
self.api = None self.api: growattServer.GrowattApi | None = None
self.user_id = None self.user_id = None
self.data = {} self.data: dict[str, Any] = {}
@callback @callback
def _async_show_user_form(self, errors=None): def _async_show_user_form(self, errors=None):
@ -42,7 +44,9 @@ class GrowattServerConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="user", data_schema=data_schema, errors=errors step_id="user", data_schema=data_schema, 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 start of the config flow.""" """Handle the start of the config flow."""
if not user_input: if not user_input:
return self._async_show_user_form() return self._async_show_user_form()

View File

@ -9,7 +9,12 @@ from pygti.auth import GTI_DEFAULT_HOST
from pygti.exceptions import CannotConnect, InvalidAuth from pygti.exceptions import CannotConnect, InvalidAuth
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigEntry, ConfigFlow, OptionsFlow from homeassistant.config_entries import (
ConfigEntry,
ConfigFlow,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.const import CONF_HOST, CONF_OFFSET, CONF_PASSWORD, CONF_USERNAME from homeassistant.const import CONF_HOST, CONF_OFFSET, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import callback from homeassistant.core import callback
from homeassistant.helpers import aiohttp_client from homeassistant.helpers import aiohttp_client
@ -44,13 +49,15 @@ class HVVDeparturesConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize component.""" """Initialize component."""
self.hub = None self.hub: GTIHub | None = None
self.data = None self.data: dict[str, Any] | None = None
self.stations = {} self.stations: 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 the initial step.""" """Handle the initial step."""
errors = {} errors = {}

View File

@ -43,7 +43,7 @@ class IcloudFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize iCloud config flow.""" """Initialize iCloud config flow."""
self.api = None self.api = None
self._username = None self._username = None
@ -55,8 +55,8 @@ class IcloudFlowHandler(ConfigFlow, domain=DOMAIN):
self._trusted_device = None self._trusted_device = None
self._verification_code = None self._verification_code = None
self._existing_entry_data = None self._existing_entry_data: dict[str, Any] | None = None
self._description_placeholders = None self._description_placeholders: dict[str, str] | None = None
def _show_setup_form(self, user_input=None, errors=None, step_id="user"): def _show_setup_form(self, user_input=None, errors=None, step_id="user"):
"""Show the setup form to the user.""" """Show the setup form to the user."""
@ -164,11 +164,13 @@ class IcloudFlowHandler(ConfigFlow, domain=DOMAIN):
await self.hass.config_entries.async_reload(entry.entry_id) await self.hass.config_entries.async_reload(entry.entry_id)
return self.async_abort(reason="reauth_successful") return self.async_abort(reason="reauth_successful")
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.""" """Handle a flow initiated by the user."""
errors = {} errors: dict[str, str] = {}
icloud_dir = Store(self.hass, STORAGE_VERSION, STORAGE_KEY) icloud_dir = Store[Any](self.hass, STORAGE_VERSION, STORAGE_KEY)
if not os.path.exists(icloud_dir.path): if not os.path.exists(icloud_dir.path):
await self.hass.async_add_executor_job(os.makedirs, icloud_dir.path) await self.hass.async_add_executor_job(os.makedirs, icloud_dir.path)

View File

@ -3,11 +3,12 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
from iotawattpy.iotawatt import Iotawatt from iotawattpy.iotawatt import Iotawatt
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigFlow from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME from homeassistant.const import CONF_HOST, CONF_PASSWORD, CONF_USERNAME
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
@ -46,11 +47,13 @@ class IOTaWattConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize.""" """Initialize."""
self._data = {} self._data: 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 the initial step.""" """Handle the initial step."""
if user_input is None: if user_input is None:
user_input = {} user_input = {}

View File

@ -1,12 +1,13 @@
"""Config flow for Kostal Plenticore Solar Inverter integration.""" """Config flow for Kostal Plenticore Solar Inverter integration."""
import logging import logging
from typing import Any
from aiohttp.client_exceptions import ClientError from aiohttp.client_exceptions import ClientError
from pykoplenti import ApiClient, AuthenticationException from pykoplenti import ApiClient, AuthenticationException
import voluptuous as vol import voluptuous as vol
from homeassistant.config_entries import ConfigFlow from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_BASE, CONF_HOST, CONF_PASSWORD from homeassistant.const import CONF_BASE, CONF_HOST, CONF_PASSWORD
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -44,10 +45,11 @@ class KostalPlenticoreConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
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.""" """Handle the initial step."""
errors = {} errors = {}
hostname = None
if user_input is not None: if user_input is not None:
self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]}) self._async_abort_entries_match({CONF_HOST: user_input[CONF_HOST]})
@ -62,8 +64,7 @@ class KostalPlenticoreConfigFlow(ConfigFlow, domain=DOMAIN):
except Exception: except Exception:
_LOGGER.exception("Unexpected exception") _LOGGER.exception("Unexpected exception")
errors[CONF_BASE] = "unknown" errors[CONF_BASE] = "unknown"
else:
if not errors:
return self.async_create_entry(title=hostname, data=user_input) return self.async_create_entry(title=hostname, data=user_input)
return self.async_show_form( return self.async_show_form(

View File

@ -6,6 +6,7 @@ import asyncio
import logging import logging
import os import os
import ssl import ssl
from typing import Any
from pylutron_caseta.pairing import PAIR_CA, PAIR_CERT, PAIR_KEY, async_pair from pylutron_caseta.pairing import PAIR_CA, PAIR_CERT, PAIR_KEY, async_pair
from pylutron_caseta.smartbridge import Smartbridge from pylutron_caseta.smartbridge import Smartbridge
@ -50,14 +51,16 @@ class LutronCasetaFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize a Lutron Caseta flow.""" """Initialize a Lutron Caseta flow."""
self.data = {} self.data: dict[str, Any] = {}
self.lutron_id = None self.lutron_id: str | None = None
self.tls_assets_validated = False self.tls_assets_validated = False
self.attempted_tls_validation = False self.attempted_tls_validation = False
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 initialized by the user.""" """Handle a flow initialized by the user."""
if user_input is not None: if user_input is not None:
self.data[CONF_HOST] = user_input[CONF_HOST] self.data[CONF_HOST] = user_input[CONF_HOST]

View File

@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
import socket import socket
from typing import Any from typing import TYPE_CHECKING, Any
from pynobo import nobo from pynobo import nobo
import voluptuous as vol import voluptuous as vol
@ -36,10 +36,10 @@ class NoboHubConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize the config flow.""" """Initialize the config flow."""
self._discovered_hubs = None self._discovered_hubs: dict[str, Any] | None = None
self._hub = None self._hub: str | None = None
async def async_step_user( async def async_step_user(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
@ -75,6 +75,9 @@ class NoboHubConfigFlow(ConfigFlow, domain=DOMAIN):
) -> ConfigFlowResult: ) -> ConfigFlowResult:
"""Handle configuration of a selected discovered device.""" """Handle configuration of a selected discovered device."""
errors = {} errors = {}
if TYPE_CHECKING:
assert self._discovered_hubs
assert self._hub
if user_input is not None: if user_input is not None:
serial_prefix = self._discovered_hubs[self._hub] serial_prefix = self._discovered_hubs[self._hub]
serial_suffix = user_input["serial_suffix"] serial_suffix = user_input["serial_suffix"]

View File

@ -62,12 +62,14 @@ async def validate_input(hass, data):
class NukiConfigFlow(ConfigFlow, domain=DOMAIN): class NukiConfigFlow(ConfigFlow, domain=DOMAIN):
"""Nuki config flow.""" """Nuki config flow."""
def __init__(self): def __init__(self) -> None:
"""Initialize the Nuki config flow.""" """Initialize the Nuki config flow."""
self.discovery_schema = {} self.discovery_schema: vol.Schema | None = None
self._data = {} self._data: Mapping[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 a flow initiated by the user.""" """Handle a flow initiated by the user."""
return await self.async_step_validate(user_input) return await self.async_step_validate(user_input)

View File

@ -112,13 +112,15 @@ class OnvifFlowHandler(ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler.""" """Get the options flow for this handler."""
return OnvifOptionsFlowHandler(config_entry) return OnvifOptionsFlowHandler(config_entry)
def __init__(self): def __init__(self) -> None:
"""Initialize the ONVIF config flow.""" """Initialize the ONVIF config flow."""
self.device_id = None self.device_id = None
self.devices = [] self.devices: list[dict[str, Any]] = []
self.onvif_config = {} self.onvif_config: 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 flow.""" """Handle user flow."""
if user_input: if user_input:
if user_input["auto"]: if user_input["auto"]: