Improve config flow type hints (q-s) (#125198)
* Improve config flow type hints (q-s) * Revert screenlogic * Revert starlinepull/125246/head
parent
643fd34478
commit
0fb1fbf0d1
|
@ -118,7 +118,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, int] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle options flow."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(title="", data=user_input)
|
||||
|
|
|
@ -60,7 +60,9 @@ class RadioThermConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
self.discovered_ip = discovery_info.ip
|
||||
return await self.async_step_confirm()
|
||||
|
||||
async def async_step_confirm(self, user_input=None):
|
||||
async def async_step_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Attempt to confirm."""
|
||||
ip_address = self.discovered_ip
|
||||
init_data = self.discovered_init_data
|
||||
|
|
|
@ -48,7 +48,7 @@ DEFAULT_OPTIONS = {CONF_PROTOCOL: DEFAULT_PROTOCOL}
|
|||
class ReolinkOptionsFlowHandler(OptionsFlow):
|
||||
"""Handle Reolink options."""
|
||||
|
||||
def __init__(self, config_entry):
|
||||
def __init__(self, config_entry: ConfigEntry) -> None:
|
||||
"""Initialize ReolinkOptionsFlowHandler."""
|
||||
self.config_entry = config_entry
|
||||
|
||||
|
|
|
@ -142,9 +142,11 @@ class RoonConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
return await self.async_step_fallback()
|
||||
|
||||
async def async_step_fallback(self, user_input=None):
|
||||
async def async_step_fallback(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Get host and port details from the user."""
|
||||
errors = {}
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
if user_input is not None:
|
||||
self._host = user_input["host"]
|
||||
|
@ -155,7 +157,9 @@ class RoonConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
step_id="fallback", data_schema=DATA_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:
|
||||
"""Handle linking and authenticating with the roon server."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
|
|
|
@ -3,7 +3,7 @@
|
|||
from collections.abc import Mapping
|
||||
from functools import partial
|
||||
import logging
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from typing import Any
|
||||
|
||||
from sense_energy import (
|
||||
ASyncSenseable,
|
||||
|
@ -34,9 +34,10 @@ class SenseConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
VERSION = 1
|
||||
|
||||
_gateway: ASyncSenseable
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Init Config ."""
|
||||
self._gateway: ASyncSenseable | None = None
|
||||
self._auth_data: dict[str, Any] = {}
|
||||
|
||||
async def validate_input(self, data: Mapping[str, Any]) -> None:
|
||||
|
@ -58,14 +59,12 @@ class SenseConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
client_session=client_session,
|
||||
)
|
||||
)
|
||||
if TYPE_CHECKING:
|
||||
assert self._gateway
|
||||
self._gateway.rate_limit = ACTIVE_UPDATE_RATE
|
||||
await self._gateway.authenticate(
|
||||
self._auth_data[CONF_EMAIL], self._auth_data[CONF_PASSWORD]
|
||||
)
|
||||
|
||||
async def create_entry_from_data(self):
|
||||
async def create_entry_from_data(self) -> ConfigFlowResult:
|
||||
"""Create the entry from the config data."""
|
||||
self._auth_data["access_token"] = self._gateway.sense_access_token
|
||||
self._auth_data["user_id"] = self._gateway.sense_user_id
|
||||
|
@ -99,7 +98,9 @@ class SenseConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
return await self.create_entry_from_data()
|
||||
return None
|
||||
|
||||
async def async_step_validation(self, user_input=None):
|
||||
async def async_step_validation(
|
||||
self, user_input: dict[str, str] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle validation (2fa) step."""
|
||||
errors = {}
|
||||
if user_input:
|
||||
|
|
|
@ -69,9 +69,11 @@ class SmappeeFlowHandler(
|
|||
|
||||
return await self.async_step_zeroconf_confirm()
|
||||
|
||||
async def async_step_zeroconf_confirm(self, user_input=None):
|
||||
async def async_step_zeroconf_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Confirm zeroconf flow."""
|
||||
errors = {}
|
||||
errors: dict[str, str] = {}
|
||||
|
||||
# Check if already configured (cloud)
|
||||
if self.is_cloud_device_already_added():
|
||||
|
@ -118,7 +120,9 @@ class SmappeeFlowHandler(
|
|||
|
||||
return await self.async_step_environment()
|
||||
|
||||
async def async_step_environment(self, user_input=None):
|
||||
async def async_step_environment(
|
||||
self, user_input: dict[str, str] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Decide environment, cloud or local."""
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
|
@ -144,7 +148,9 @@ class SmappeeFlowHandler(
|
|||
|
||||
return await self.async_step_pick_implementation()
|
||||
|
||||
async def async_step_local(self, user_input=None):
|
||||
async def async_step_local(
|
||||
self, user_input: dict[str, str] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle local flow."""
|
||||
if user_input is None:
|
||||
return self.async_show_form(
|
||||
|
|
|
@ -42,16 +42,17 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
VERSION = 2
|
||||
|
||||
api: SmartThings
|
||||
app_id: str
|
||||
location_id: str
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Create a new instance of the flow handler."""
|
||||
self.access_token = None
|
||||
self.app_id = None
|
||||
self.api = None
|
||||
self.access_token: str | None = None
|
||||
self.oauth_client_secret = None
|
||||
self.oauth_client_id = None
|
||||
self.installed_app_id = None
|
||||
self.refresh_token = None
|
||||
self.location_id = None
|
||||
self.endpoints_initialized = False
|
||||
|
||||
async def async_step_import(self, import_data: None) -> ConfigFlowResult:
|
||||
|
@ -91,9 +92,11 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
# Show the next screen
|
||||
return await self.async_step_pat()
|
||||
|
||||
async def async_step_pat(self, user_input=None):
|
||||
async def async_step_pat(
|
||||
self, user_input: dict[str, str] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Get the Personal Access Token and validate it."""
|
||||
errors = {}
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is None or CONF_ACCESS_TOKEN not in user_input:
|
||||
return self._show_step_pat(errors)
|
||||
|
||||
|
@ -169,7 +172,9 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
return await self.async_step_select_location()
|
||||
|
||||
async def async_step_select_location(self, user_input=None):
|
||||
async def async_step_select_location(
|
||||
self, user_input: dict[str, str] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Ask user to select the location to setup."""
|
||||
if user_input is None or CONF_LOCATION_ID not in user_input:
|
||||
# Get available locations
|
||||
|
@ -196,7 +201,9 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
await self.async_set_unique_id(format_unique_id(self.app_id, self.location_id))
|
||||
return await self.async_step_authorize()
|
||||
|
||||
async def async_step_authorize(self, user_input=None):
|
||||
async def async_step_authorize(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Wait for the user to authorize the app installation."""
|
||||
user_input = {} if user_input is None else user_input
|
||||
self.installed_app_id = user_input.get(CONF_INSTALLED_APP_ID)
|
||||
|
@ -233,7 +240,9 @@ class SmartThingsFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
},
|
||||
)
|
||||
|
||||
async def async_step_install(self, data=None):
|
||||
async def async_step_install(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Create a config entry at completion of a flow and authorization of the app."""
|
||||
data = {
|
||||
CONF_ACCESS_TOKEN: self.access_token,
|
||||
|
|
|
@ -81,9 +81,13 @@ class SmartTubConfigFlow(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:
|
||||
"""Dialog that informs the user that reauth is required."""
|
||||
if user_input is None:
|
||||
if TYPE_CHECKING:
|
||||
assert self._reauth_input is not None
|
||||
# same as DATA_SCHEMA but with default email
|
||||
data_schema = vol.Schema(
|
||||
{
|
||||
|
|
|
@ -39,7 +39,7 @@ class SomaFlowHandler(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
return await self.async_step_creation(user_input)
|
||||
|
||||
async def async_step_creation(self, user_input=None):
|
||||
async def async_step_creation(self, user_input: dict[str, Any]) -> ConfigFlowResult:
|
||||
"""Finish config flow."""
|
||||
try:
|
||||
api = await self.hass.async_add_executor_job(
|
||||
|
|
|
@ -132,7 +132,7 @@ class OptionsFlowHandler(OptionsFlow):
|
|||
"""Initialize options flow."""
|
||||
self.config_entry = config_entry
|
||||
self.options = deepcopy(dict(config_entry.options))
|
||||
self._target_id = None
|
||||
self._target_id: str | None = None
|
||||
|
||||
@callback
|
||||
def _async_callback_targets(self):
|
||||
|
@ -150,7 +150,9 @@ class OptionsFlowHandler(OptionsFlow):
|
|||
return cover["name"]
|
||||
raise KeyError
|
||||
|
||||
async def async_step_init(self, user_input=None):
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle options flow."""
|
||||
|
||||
if self.config_entry.state is not ConfigEntryState.LOADED:
|
||||
|
@ -173,9 +175,13 @@ class OptionsFlowHandler(OptionsFlow):
|
|||
|
||||
return self.async_show_form(step_id="init", data_schema=data_schema, errors={})
|
||||
|
||||
async def async_step_target_config(self, user_input=None, target_id=None):
|
||||
async def async_step_target_config(
|
||||
self, user_input: dict[str, bool] | None = None, target_id: str | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle options flow for target."""
|
||||
reversed_target_ids = self.options.setdefault(CONF_REVERSED_TARGET_IDS, {})
|
||||
reversed_target_ids: dict[str | None, bool] = self.options.setdefault(
|
||||
CONF_REVERSED_TARGET_IDS, {}
|
||||
)
|
||||
|
||||
if user_input is not None:
|
||||
if user_input[CONF_REVERSE] != reversed_target_ids.get(self._target_id):
|
||||
|
|
|
@ -3,7 +3,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from songpal import Device, SongpalException
|
||||
|
@ -21,7 +21,7 @@ _LOGGER = logging.getLogger(__name__)
|
|||
class SongpalConfig:
|
||||
"""Device Configuration."""
|
||||
|
||||
def __init__(self, name, host, endpoint):
|
||||
def __init__(self, name: str, host: str | None, endpoint: str) -> None:
|
||||
"""Initialize Configuration."""
|
||||
self.name = name
|
||||
self.host = host
|
||||
|
@ -33,12 +33,10 @@ class SongpalConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Initialize the flow."""
|
||||
self.conf: SongpalConfig | None = None
|
||||
conf: SongpalConfig
|
||||
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
self, user_input: dict[str, str] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle a flow initiated by the user."""
|
||||
if user_input is None:
|
||||
|
@ -75,7 +73,9 @@ class SongpalConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
|
||||
return await self.async_step_init(user_input)
|
||||
|
||||
async def async_step_init(self, user_input=None):
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle a flow start."""
|
||||
# Check if already configured
|
||||
self._async_abort_entries_match({CONF_ENDPOINT: self.conf.endpoint})
|
||||
|
@ -122,14 +122,16 @@ class SongpalConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
CONF_HOST: parsed_url.hostname,
|
||||
}
|
||||
|
||||
if TYPE_CHECKING:
|
||||
assert isinstance(parsed_url.hostname, str)
|
||||
self.conf = SongpalConfig(friendly_name, parsed_url.hostname, endpoint)
|
||||
|
||||
return await self.async_step_init()
|
||||
|
||||
async def async_step_import(self, import_data: dict[str, Any]) -> ConfigFlowResult:
|
||||
async def async_step_import(self, import_data: dict[str, str]) -> ConfigFlowResult:
|
||||
"""Import a config entry."""
|
||||
name = import_data.get(CONF_NAME)
|
||||
endpoint = import_data.get(CONF_ENDPOINT)
|
||||
endpoint = import_data[CONF_ENDPOINT]
|
||||
parsed_url = urlparse(endpoint)
|
||||
|
||||
# Try to connect to test the endpoint
|
||||
|
|
|
@ -68,7 +68,9 @@ class SoundtouchConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
self.context["title_placeholders"] = {"name": self.name}
|
||||
return await self.async_step_zeroconf_confirm()
|
||||
|
||||
async def async_step_zeroconf_confirm(self, user_input=None):
|
||||
async def async_step_zeroconf_confirm(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle user-confirmation of discovered node."""
|
||||
if user_input is not None:
|
||||
return await self._async_create_soundtouch_entry()
|
||||
|
|
|
@ -64,7 +64,9 @@ class SyncThruConfigFlow(ConfigFlow, domain=DOMAIN):
|
|||
self.context["title_placeholders"] = {CONF_NAME: self.name}
|
||||
return await self.async_step_confirm()
|
||||
|
||||
async def async_step_confirm(self, user_input=None):
|
||||
async def async_step_confirm(
|
||||
self, user_input: dict[str, str] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle discovery confirmation by user."""
|
||||
if user_input is not None:
|
||||
return await self._async_check_and_create("confirm", user_input)
|
||||
|
|
Loading…
Reference in New Issue