mirror of
https://github.com/home-assistant/core
synced 2024-07-15 09:42:11 +02:00
Add typing to config flow A (#105721)
This commit is contained in:
parent
8cd0644035
commit
9944047b35
@ -1,6 +1,8 @@
|
||||
"""Config flow for AEMET OpenData."""
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from aemet_opendata.exceptions import AuthError
|
||||
from aemet_opendata.interface import AEMET, ConnectionOptions
|
||||
import voluptuous as vol
|
||||
@ -8,6 +10,7 @@ import voluptuous as vol
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_API_KEY, CONF_LATITUDE, CONF_LONGITUDE, CONF_NAME
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers import aiohttp_client, config_validation as cv
|
||||
from homeassistant.helpers.schema_config_entry_flow import (
|
||||
SchemaFlowFormStep,
|
||||
@ -29,7 +32,9 @@ OPTIONS_FLOW = {
|
||||
class AemetConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Config flow for AEMET OpenData."""
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initialized by the user."""
|
||||
errors = {}
|
||||
|
||||
|
@ -1,5 +1,6 @@
|
||||
"""Config flow to configure Agent devices."""
|
||||
from contextlib import suppress
|
||||
from typing import Any
|
||||
|
||||
from agent import AgentConnectionError, AgentError
|
||||
from agent.a import Agent
|
||||
@ -7,6 +8,7 @@ import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
|
||||
from .const import DOMAIN, SERVER_URL
|
||||
@ -18,11 +20,9 @@ DEFAULT_PORT = 8090
|
||||
class AgentFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Handle an Agent config flow."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the Agent config flow."""
|
||||
self.device_config = {}
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle an Agent config flow."""
|
||||
errors = {}
|
||||
|
||||
@ -49,13 +49,15 @@ class AgentFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
}
|
||||
)
|
||||
|
||||
self.device_config = {
|
||||
device_config = {
|
||||
CONF_HOST: host,
|
||||
CONF_PORT: port,
|
||||
SERVER_URL: server_origin,
|
||||
}
|
||||
|
||||
return await self._create_entry(agent_client.name)
|
||||
return self.async_create_entry(
|
||||
title=agent_client.name, data=device_config
|
||||
)
|
||||
|
||||
errors["base"] = "cannot_connect"
|
||||
|
||||
@ -66,11 +68,6 @@ class AgentFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
description_placeholders=self.device_config,
|
||||
data_schema=vol.Schema(data),
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def _create_entry(self, server_name):
|
||||
"""Create entry for device."""
|
||||
return self.async_create_entry(title=server_name, data=self.device_config)
|
||||
|
@ -2,6 +2,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from adext import AdExt
|
||||
from alarmdecoder.devices import SerialDevice, SocketDevice
|
||||
@ -12,8 +13,10 @@ from homeassistant import config_entries
|
||||
from homeassistant.components.binary_sensor import (
|
||||
DEVICE_CLASSES_SCHEMA as BINARY_SENSOR_DEVICE_CLASSES_SCHEMA,
|
||||
)
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT, CONF_PROTOCOL
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import (
|
||||
CONF_ALT_NIGHT_MODE,
|
||||
@ -66,7 +69,9 @@ class AlarmDecoderFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Get the options flow for AlarmDecoder."""
|
||||
return AlarmDecoderOptionsFlowHandler(config_entry)
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initialized by the user."""
|
||||
if user_input is not None:
|
||||
self.protocol = user_input[CONF_PROTOCOL]
|
||||
@ -83,7 +88,9 @@ class AlarmDecoderFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
),
|
||||
)
|
||||
|
||||
async def async_step_protocol(self, user_input=None):
|
||||
async def async_step_protocol(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle AlarmDecoder protocol setup."""
|
||||
errors = {}
|
||||
if user_input is not None:
|
||||
@ -146,15 +153,18 @@ class AlarmDecoderFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
class AlarmDecoderOptionsFlowHandler(config_entries.OptionsFlow):
|
||||
"""Handle AlarmDecoder options."""
|
||||
|
||||
selected_zone: str | None = None
|
||||
|
||||
def __init__(self, config_entry: config_entries.ConfigEntry) -> None:
|
||||
"""Initialize AlarmDecoder options flow."""
|
||||
self.arm_options = config_entry.options.get(OPTIONS_ARM, DEFAULT_ARM_OPTIONS)
|
||||
self.zone_options = config_entry.options.get(
|
||||
OPTIONS_ZONES, DEFAULT_ZONE_OPTIONS
|
||||
)
|
||||
self.selected_zone = None
|
||||
|
||||
async def async_step_init(self, user_input=None):
|
||||
async def async_step_init(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Manage the options."""
|
||||
if user_input is not None:
|
||||
if user_input[EDIT_KEY] == EDIT_SETTINGS:
|
||||
@ -173,7 +183,9 @@ class AlarmDecoderOptionsFlowHandler(config_entries.OptionsFlow):
|
||||
),
|
||||
)
|
||||
|
||||
async def async_step_arm_settings(self, user_input=None):
|
||||
async def async_step_arm_settings(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Arming options form."""
|
||||
if user_input is not None:
|
||||
return self.async_create_entry(
|
||||
@ -200,7 +212,9 @@ class AlarmDecoderOptionsFlowHandler(config_entries.OptionsFlow):
|
||||
),
|
||||
)
|
||||
|
||||
async def async_step_zone_select(self, user_input=None):
|
||||
async def async_step_zone_select(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Zone selection form."""
|
||||
errors = _validate_zone_input(user_input)
|
||||
|
||||
@ -216,7 +230,9 @@ class AlarmDecoderOptionsFlowHandler(config_entries.OptionsFlow):
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_zone_details(self, user_input=None):
|
||||
async def async_step_zone_details(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Zone details form."""
|
||||
errors = _validate_zone_input(user_input)
|
||||
|
||||
@ -293,7 +309,7 @@ class AlarmDecoderOptionsFlowHandler(config_entries.OptionsFlow):
|
||||
)
|
||||
|
||||
|
||||
def _validate_zone_input(zone_input):
|
||||
def _validate_zone_input(zone_input: dict[str, Any] | None) -> dict[str, str]:
|
||||
if not zone_input:
|
||||
return {}
|
||||
errors = {}
|
||||
@ -327,7 +343,7 @@ def _validate_zone_input(zone_input):
|
||||
return errors
|
||||
|
||||
|
||||
def _fix_input_types(zone_input):
|
||||
def _fix_input_types(zone_input: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Convert necessary keys to int.
|
||||
|
||||
Since ConfigFlow inputs of type int cannot default to an empty string, we collect the values below as
|
||||
@ -341,7 +357,9 @@ def _fix_input_types(zone_input):
|
||||
return zone_input
|
||||
|
||||
|
||||
def _device_already_added(current_entries, user_input, protocol):
|
||||
def _device_already_added(
|
||||
current_entries: list[ConfigEntry], user_input: dict[str, Any], protocol: str | None
|
||||
) -> bool:
|
||||
"""Determine if entry has already been added to HA."""
|
||||
user_host = user_input.get(CONF_HOST)
|
||||
user_port = user_input.get(CONF_PORT)
|
||||
|
@ -1,5 +1,6 @@
|
||||
"""Config flow for Ambiclimate."""
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from aiohttp import web
|
||||
import ambiclimate
|
||||
@ -7,7 +8,8 @@ import ambiclimate
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.components.http import HomeAssistantView
|
||||
from homeassistant.const import CONF_CLIENT_ID, CONF_CLIENT_SECRET
|
||||
from homeassistant.core import callback
|
||||
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.network import get_url
|
||||
from homeassistant.helpers.storage import Store
|
||||
@ -26,7 +28,9 @@ _LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@callback
|
||||
def register_flow_implementation(hass, client_id, client_secret):
|
||||
def register_flow_implementation(
|
||||
hass: HomeAssistant, client_id: str, client_secret: str
|
||||
) -> None:
|
||||
"""Register a ambiclimate implementation.
|
||||
|
||||
client_id: Client id.
|
||||
@ -50,7 +54,9 @@ class AmbiclimateFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
self._registered_view = False
|
||||
self._oauth = None
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle external yaml configuration."""
|
||||
self._async_abort_entries_match()
|
||||
|
||||
@ -62,7 +68,9 @@ class AmbiclimateFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
return await self.async_step_auth()
|
||||
|
||||
async def async_step_auth(self, user_input=None):
|
||||
async def async_step_auth(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow start."""
|
||||
self._async_abort_entries_match()
|
||||
|
||||
@ -83,7 +91,7 @@ class AmbiclimateFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
errors=errors,
|
||||
)
|
||||
|
||||
async def async_step_code(self, code=None):
|
||||
async def async_step_code(self, code: str | None = None) -> FlowResult:
|
||||
"""Received code for authentication."""
|
||||
self._async_abort_entries_match()
|
||||
|
||||
@ -95,7 +103,7 @@ class AmbiclimateFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
return self.async_create_entry(title="Ambiclimate", data=config)
|
||||
|
||||
async def _get_token_info(self, code):
|
||||
async def _get_token_info(self, code: str | None) -> dict[str, Any] | None:
|
||||
oauth = self._generate_oauth()
|
||||
try:
|
||||
token_info = await oauth.get_access_token(code)
|
||||
@ -103,16 +111,16 @@ class AmbiclimateFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
_LOGGER.exception("Failed to get access token")
|
||||
return None
|
||||
|
||||
store = Store(self.hass, STORAGE_VERSION, STORAGE_KEY)
|
||||
store = Store[dict[str, Any]](self.hass, STORAGE_VERSION, STORAGE_KEY)
|
||||
await store.async_save(token_info)
|
||||
|
||||
return token_info
|
||||
|
||||
def _generate_view(self):
|
||||
def _generate_view(self) -> None:
|
||||
self.hass.http.register_view(AmbiclimateAuthCallbackView())
|
||||
self._registered_view = True
|
||||
|
||||
def _generate_oauth(self):
|
||||
def _generate_oauth(self) -> ambiclimate.AmbiclimateOAuth:
|
||||
config = self.hass.data[DATA_AMBICLIMATE_IMPL]
|
||||
clientsession = async_get_clientsession(self.hass)
|
||||
callback_url = self._cb_url()
|
||||
|
@ -1,9 +1,12 @@
|
||||
"""Config flow for the Atag component."""
|
||||
from typing import Any
|
||||
|
||||
import pyatag
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
|
||||
from . import DOMAIN
|
||||
@ -19,7 +22,9 @@ class AtagConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle a flow initialized by the user."""
|
||||
|
||||
if not user_input:
|
||||
@ -39,7 +44,7 @@ class AtagConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
return self.async_create_entry(title=atag.id, data=user_input)
|
||||
|
||||
async def _show_form(self, errors=None):
|
||||
async def _show_form(self, errors: dict[str, str] | None = None) -> FlowResult:
|
||||
"""Show the form to the user."""
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
|
@ -80,20 +80,24 @@ class AugustConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
VERSION = 1
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self) -> None:
|
||||
"""Store an AugustGateway()."""
|
||||
self._august_gateway: AugustGateway | None = None
|
||||
self._aiohttp_session: aiohttp.ClientSession | None = None
|
||||
self._user_auth_details: dict[str, Any] = {}
|
||||
self._needs_reset = True
|
||||
self._mode = None
|
||||
self._mode: str | None = None
|
||||
super().__init__()
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
return await self.async_step_user_validate()
|
||||
|
||||
async def async_step_user_validate(self, user_input=None):
|
||||
async def async_step_user_validate(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle authentication."""
|
||||
errors: dict[str, str] = {}
|
||||
description_placeholders: dict[str, str] = {}
|
||||
@ -177,7 +181,9 @@ class AugustConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
self._needs_reset = True
|
||||
return await self.async_step_reauth_validate()
|
||||
|
||||
async def async_step_reauth_validate(self, user_input=None):
|
||||
async def async_step_reauth_validate(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle reauth and validation."""
|
||||
errors: dict[str, str] = {}
|
||||
description_placeholders: dict[str, str] = {}
|
||||
|
@ -2,6 +2,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from aiohttp import ClientError
|
||||
from auroranoaa import AuroraForecast
|
||||
@ -10,6 +11,7 @@ import voluptuous as vol
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.const import CONF_LATITUDE, CONF_LONGITUDE
|
||||
from homeassistant.core import callback
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
from homeassistant.helpers import aiohttp_client, config_validation as cv
|
||||
from homeassistant.helpers.schema_config_entry_flow import (
|
||||
SchemaFlowFormStep,
|
||||
@ -45,7 +47,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
"""Get the options flow for this handler."""
|
||||
return SchemaOptionsFlowHandler(config_entry, OPTIONS_FLOW)
|
||||
|
||||
async def async_step_user(self, user_input=None):
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> FlowResult:
|
||||
"""Handle the initial step."""
|
||||
errors = {}
|
||||
|
||||
|
@ -1,6 +1,10 @@
|
||||
"""Config flow for AWS component."""
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any
|
||||
|
||||
from homeassistant import config_entries
|
||||
from homeassistant.data_entry_flow import FlowResult
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
@ -10,7 +14,7 @@ class AWSFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
|
||||
|
||||
VERSION = 1
|
||||
|
||||
async def async_step_import(self, user_input):
|
||||
async def async_step_import(self, user_input: Mapping[str, Any]) -> FlowResult:
|
||||
"""Import a config entry."""
|
||||
if self._async_current_entries():
|
||||
return self.async_abort(reason="single_instance_allowed")
|
||||
|
Loading…
Reference in New Issue
Block a user