evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Correct use of ConfigType in MQTT config flow code (#79934)

Browse source 
Correct use of ConfigType
This commit is contained in:
Jan Bouwhuis 2022-10-09 14:41:30 +02:00 • committed by GitHub
parent 5b0a37a447
commit 6eb2c96d32
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 17 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
homeassistant/components/mqtt/config_flow.py
View file

@ -58,7 +58,9 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
"""Get the options flow for this handler.""" """Get the options flow for this handler."""
return MQTTOptionsFlowHandler(config_entry) return MQTTOptionsFlowHandler(config_entry)
async def async_step_user(self, user_input: ConfigType | None = None) -> FlowResult: async def async_step_user(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle a flow initialized by the user.""" """Handle a flow initialized by the user."""
if self._async_current_entries(): if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed") return self.async_abort(reason="single_instance_allowed")
@ -66,13 +68,13 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_broker() return await self.async_step_broker()
async def async_step_broker( async def async_step_broker(
self, user_input: ConfigType | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Confirm the setup.""" """Confirm the setup."""
yaml_config: ConfigType = get_mqtt_data(self.hass, True).config or {} yaml_config: ConfigType = get_mqtt_data(self.hass, True).config or {}
errors: dict[str, str] = {} errors: dict[str, str] = {}
fields: OrderedDict[Any, Any] = OrderedDict() fields: OrderedDict[Any, Any] = OrderedDict()
validated_user_input: ConfigType = {} validated_user_input: dict[str, Any] = {}
if await async_get_broker_settings( if await async_get_broker_settings(
self.hass, self.hass,
fields, fields,
@ -82,7 +84,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
validated_user_input, validated_user_input,
errors, errors,
): ):
test_config: ConfigType = yaml_config.copy() test_config: dict[str, Any] = yaml_config.copy()
test_config.update(validated_user_input) test_config.update(validated_user_input)
can_connect = await self.hass.async_add_executor_job( can_connect = await self.hass.async_add_executor_job(
try_connection, try_connection,
@ -118,7 +120,7 @@ class FlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
assert self._hassio_discovery assert self._hassio_discovery
if user_input is not None: if user_input is not None:
data: ConfigType = self._hassio_discovery.copy() data: dict[str, Any] = self._hassio_discovery.copy()
data[CONF_BROKER] = data.pop(CONF_HOST) data[CONF_BROKER] = data.pop(CONF_HOST)
can_connect = await self.hass.async_add_executor_job( can_connect = await self.hass.async_add_executor_job(
try_connection, try_connection,
@ -166,7 +168,7 @@ class MQTTOptionsFlowHandler(config_entries.OptionsFlow):
errors: dict[str, str] = {} errors: dict[str, str] = {}
yaml_config: ConfigType = get_mqtt_data(self.hass, True).config or {} yaml_config: ConfigType = get_mqtt_data(self.hass, True).config or {}
fields: OrderedDict[Any, Any] = OrderedDict() fields: OrderedDict[Any, Any] = OrderedDict()
validated_user_input: ConfigType = {} validated_user_input: dict[str, Any] = {}
if await async_get_broker_settings( if await async_get_broker_settings(
self.hass, self.hass,
fields, fields,
@ -176,7 +178,7 @@ class MQTTOptionsFlowHandler(config_entries.OptionsFlow):
validated_user_input, validated_user_input,
errors, errors,
): ):
test_config: ConfigType = yaml_config.copy() test_config: dict[str, Any] = yaml_config.copy()
test_config.update(validated_user_input) test_config.update(validated_user_input)
can_connect = await self.hass.async_add_executor_job( can_connect = await self.hass.async_add_executor_job(
try_connection, try_connection,
@ -197,13 +199,13 @@ class MQTTOptionsFlowHandler(config_entries.OptionsFlow):
) )
async def async_step_options( async def async_step_options(
self, user_input: ConfigType | None = None self, user_input: dict[str, Any] | None = None
) -> FlowResult: ) -> FlowResult:
"""Manage the MQTT options.""" """Manage the MQTT options."""
errors = {} errors = {}
current_config = self.config_entry.data current_config = self.config_entry.data
yaml_config = get_mqtt_data(self.hass, True).config or {} yaml_config = get_mqtt_data(self.hass, True).config or {}
options_config: ConfigType = {} options_config: dict[str, Any] = {}
bad_input: bool = False bad_input: bool = False
def _birth_will(birt_or_will: str) -> dict: def _birth_will(birt_or_will: str) -> dict:
@ -217,7 +219,7 @@ class MQTTOptionsFlowHandler(config_entries.OptionsFlow):
} }
def _validate( def _validate(
field: str, values: ConfigType, error_code: str, schema: Callable field: str, values: dict[str, Any], error_code: str, schema: Callable
): ):
"""Validate the user input.""" """Validate the user input."""
nonlocal bad_input nonlocal bad_input
@ -337,16 +339,16 @@ async def async_get_broker_settings(
fields: OrderedDict[Any, Any], fields: OrderedDict[Any, Any],
yaml_config: ConfigType, yaml_config: ConfigType,
entry_config: MappingProxyType[str, Any] | None, entry_config: MappingProxyType[str, Any] | None,
user_input: ConfigType | None, user_input: dict[str, Any] | None,
validated_user_input: ConfigType, validated_user_input: dict[str, Any],
errors: dict[str, str], errors: dict[str, str],
) -> bool: ) -> bool:
"""Build the config flow schema to collect the broker settings. """Build the config flow schema to collect the broker settings.
Returns True when settings are collected successfully. Returns True when settings are collected successfully.
""" """
user_input_basic: ConfigType = ConfigType() user_input_basic: dict[str, Any] = {}
current_config = entry_config.copy() if entry_config is not None else ConfigType() current_config = entry_config.copy() if entry_config is not None else {}
if user_input is not None: if user_input is not None:
validated_user_input.update(user_input) validated_user_input.update(user_input)
@ -384,7 +386,7 @@ async def async_get_broker_settings(
def try_connection( def try_connection(
user_input: ConfigType, user_input: dict[str, Any],
) -> bool: ) -> bool:
"""Test if we can connect to an MQTT broker.""" """Test if we can connect to an MQTT broker."""
# We don't import on the top because some integrations # We don't import on the top because some integrations