Improve config flow type hints (part 5) (#124349)

This commit is contained in:
epenet 2024-08-25 18:34:39 +02:00 committed by GitHub
parent 70fc8fa2eb
commit 909dfcc436
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 42 additions and 21 deletions

View File

@ -28,14 +28,16 @@ class TotalConnectConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize the config flow.""" """Initialize the config flow."""
self.username = None self.username = None
self.password = None self.password = None
self.usercodes = {} self.usercodes: dict[str, Any] = {}
self.client = None self.client = 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.""" """Handle a flow initiated by the user."""
errors = {} errors = {}

View File

@ -3,12 +3,13 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any
import aiohttp import aiohttp
from uasiren.client import Client from uasiren.client import Client
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_REGION from homeassistant.const import CONF_NAME, CONF_REGION
from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.aiohttp_client import async_get_clientsession
@ -22,12 +23,14 @@ class UkraineAlarmConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize a new UkraineAlarmConfigFlow.""" """Initialize a new UkraineAlarmConfigFlow."""
self.states = None self.states = None
self.selected_region = None self.selected_region: dict[str, Any] | 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 initialized by the user.""" """Handle a flow initialized by the user."""
if len(self._async_current_entries()) == 5: if len(self._async_current_entries()) == 5:
@ -91,7 +94,11 @@ class UkraineAlarmConfigFlow(ConfigFlow, domain=DOMAIN):
): ):
self.selected_region = _find(source, user_input[CONF_REGION]) self.selected_region = _find(source, user_input[CONF_REGION])
if next_step and self.selected_region["regionChildIds"]: if (
next_step
and self.selected_region
and self.selected_region["regionChildIds"]
):
return await getattr(self, f"async_step_{next_step}")() return await getattr(self, f"async_step_{next_step}")()
return await self._async_finish_flow() return await self._async_finish_flow()

View File

@ -24,13 +24,13 @@ class WiLightFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize the WiLight flow.""" """Initialize the WiLight flow."""
self._host = None self._host = None
self._serial_number = None self._serial_number = None
self._title = None self._title = None
self._model_name = None self._model_name = None
self._wilight_components = [] self._wilight_components: list[str] = []
self._components_text = "" self._components_text = ""
def _wilight_update(self, host, serial_number, model_name): def _wilight_update(self, host, serial_number, model_name):

View File

@ -2,6 +2,7 @@
import logging import logging
from socket import gaierror from socket import gaierror
from typing import TYPE_CHECKING, Any
import voluptuous as vol import voluptuous as vol
from xiaomi_gateway import MULTICAST_PORT, XiaomiGateway, XiaomiGatewayDiscovery from xiaomi_gateway import MULTICAST_PORT, XiaomiGateway, XiaomiGatewayDiscovery
@ -49,13 +50,13 @@ class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self): def __init__(self) -> None:
"""Initialize.""" """Initialize."""
self.host = None self.host: str | None = None
self.interface = DEFAULT_INTERFACE self.interface = DEFAULT_INTERFACE
self.sid = None self.sid: str | None = None
self.gateways = None self.gateways: dict[str, XiaomiGateway] | None = None
self.selected_gateway = None self.selected_gateway: XiaomiGateway | None = None
@callback @callback
def async_show_form_step_user(self, errors): def async_show_form_step_user(self, errors):
@ -66,9 +67,11 @@ class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
return self.async_show_form(step_id="user", data_schema=schema, errors=errors) return self.async_show_form(step_id="user", data_schema=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 a flow initialized by the user.""" """Handle a flow initialized by the user."""
errors = {} errors: dict[str, str] = {}
if user_input is None: if user_input is None:
return self.async_show_form_step_user(errors) return self.async_show_form_step_user(errors)
@ -96,6 +99,8 @@ class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
None, None,
) )
if TYPE_CHECKING:
assert self.selected_gateway
if self.selected_gateway.connection_error: if self.selected_gateway.connection_error:
errors[CONF_HOST] = "invalid_host" errors[CONF_HOST] = "invalid_host"
if self.selected_gateway.mac_error: if self.selected_gateway.mac_error:
@ -115,6 +120,8 @@ class XiaomiAqaraFlowHandler(ConfigFlow, domain=DOMAIN):
self.gateways = xiaomi.gateways self.gateways = xiaomi.gateways
if TYPE_CHECKING:
assert self.gateways is not None
if len(self.gateways) == 1: if len(self.gateways) == 1:
self.selected_gateway = list(self.gateways.values())[0] self.selected_gateway = list(self.gateways.values())[0]
self.sid = self.selected_gateway.sid self.sid = self.selected_gateway.sid

View File

@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import TYPE_CHECKING, Any
from urllib.parse import urlparse from urllib.parse import urlparse
import voluptuous as vol import voluptuous as vol
@ -57,11 +58,11 @@ class YeelightConfigFlow(ConfigFlow, domain=DOMAIN):
"""Return the options flow.""" """Return the options flow."""
return OptionsFlowHandler(config_entry) return OptionsFlowHandler(config_entry)
def __init__(self): def __init__(self) -> None:
"""Initialize the config flow.""" """Initialize the config flow."""
self._discovered_devices = {} self._discovered_devices: dict[str, Any] = {}
self._discovered_model = None self._discovered_model = None
self._discovered_ip = None self._discovered_ip: str | None = None
async def async_step_homekit( async def async_step_homekit(
self, discovery_info: zeroconf.ZeroconfServiceInfo self, discovery_info: zeroconf.ZeroconfServiceInfo
@ -162,7 +163,9 @@ class YeelightConfigFlow(ConfigFlow, domain=DOMAIN):
step_id="discovery_confirm", description_placeholders=placeholders step_id="discovery_confirm", description_placeholders=placeholders
) )
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 = {}
if user_input is not None: if user_input is not None:
@ -176,6 +179,8 @@ class YeelightConfigFlow(ConfigFlow, domain=DOMAIN):
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
else: else:
self._abort_if_unique_id_configured() self._abort_if_unique_id_configured()
if TYPE_CHECKING:
assert self.unique_id
return self.async_create_entry( return self.async_create_entry(
title=async_format_model_id(model, self.unique_id), title=async_format_model_id(model, self.unique_id),
data={ data={