None optional hass typing in FlowHandler (#46462)

Co-authored-by: Martin Hjelmare <marhje52@gmail.com>
pull/46489/head
Franck Nijhof 2021-02-13 13:21:37 +01:00 committed by GitHub
parent 1244fb4152
commit bc1daf1802
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
58 changed files with 22 additions and 141 deletions

View File

@ -198,7 +198,7 @@ class TotpSetupFlow(SetupFlow):
errors: Dict[str, str] = {}
if user_input:
verified = await self.hass.async_add_executor_job( # type: ignore
verified = await self.hass.async_add_executor_job(
pyotp.TOTP(self._ota_secret).verify, user_input["code"]
)
if verified:

View File

@ -101,10 +101,7 @@ class AppleTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(info[CONF_IDENTIFIER])
self.target_device = info[CONF_IDENTIFIER]
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {"name": info[CONF_NAME]}
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["identifier"] = self.unique_id
return await self.async_step_reconfigure()
@ -170,7 +167,6 @@ class AppleTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(identifier)
self._abort_if_unique_id_configured()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["identifier"] = self.unique_id
self.context["title_placeholders"] = {"name": name}
self.target_device = identifier

View File

@ -71,7 +71,7 @@ class ArcamFmjFlowHandler(config_entries.ConfigFlow):
async def async_step_confirm(self, user_input=None):
"""Handle user-confirmation of discovered node."""
context = self.context # pylint: disable=no-member
context = self.context
placeholders = {
"host": context[CONF_HOST],
}
@ -94,7 +94,7 @@ class ArcamFmjFlowHandler(config_entries.ConfigFlow):
await self._async_set_unique_id_and_update(host, port, uuid)
context = self.context # pylint: disable=no-member
context = self.context
context[CONF_HOST] = host
context[CONF_PORT] = DEFAULT_PORT
return await self.async_step_confirm()

View File

@ -138,7 +138,6 @@ class AxisFlowHandler(config_entries.ConfigFlow, domain=AXIS_DOMAIN):
async def async_step_reauth(self, device_config: dict):
"""Trigger a reauthentication flow."""
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
CONF_NAME: device_config[CONF_NAME],
CONF_HOST: device_config[CONF_HOST],
@ -204,7 +203,6 @@ class AxisFlowHandler(config_entries.ConfigFlow, domain=AXIS_DOMAIN):
}
)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
CONF_NAME: device[CONF_NAME],
CONF_HOST: device[CONF_HOST],

View File

@ -95,7 +95,6 @@ class AzureDevOpsFlowHandler(ConfigFlow, domain=DOMAIN):
self._project = user_input[CONF_PROJECT]
self._pat = user_input[CONF_PAT]
# pylint: disable=no-member
self.context["title_placeholders"] = {
"project_url": f"{self._organization}/{self._project}",
}

View File

@ -73,7 +73,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
CONF_HOST: host,
CONF_BOND_ID: bond_id,
}
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": self._discovered})
return await self.async_step_confirm()

View File

@ -57,7 +57,6 @@ class BroadlinkFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
)
self.device = device
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
"name": device.name,
"model": device.model,

View File

@ -97,7 +97,6 @@ class BrotherConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(self.brother.serial.lower())
self._abort_if_unique_id_configured()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update(
{
"title_placeholders": {
@ -112,7 +111,6 @@ class BrotherConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Handle a flow initiated by zeroconf."""
if user_input is not None:
title = f"{self.brother.model} {self.brother.serial}"
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
return self.async_create_entry(
title=title,
data={CONF_HOST: self.host, CONF_TYPE: user_input[CONF_TYPE]},

View File

@ -63,9 +63,7 @@ class CertexpiryConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
title=title,
data={CONF_HOST: host, CONF_PORT: port},
)
if ( # pylint: disable=no-member
self.context["source"] == config_entries.SOURCE_IMPORT
):
if self.context["source"] == config_entries.SOURCE_IMPORT:
_LOGGER.error("Config import failed for %s", user_input[CONF_HOST])
return self.async_abort(reason="import_failed")
else:

View File

@ -97,7 +97,6 @@ class CloudflareConfigFlow(ConfigFlow, domain=DOMAIN):
if self._async_current_entries():
return self.async_abort(reason="single_instance_allowed")
assert self.hass
persistent_notification.async_dismiss(self.hass, "cloudflare_setup")
errors = {}

View File

@ -176,7 +176,6 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_reauth(self, config: dict):
"""Trigger a reauthentication flow."""
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {CONF_HOST: config[CONF_HOST]}
self.deconz_config = {
@ -207,7 +206,6 @@ class DeconzFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
updates={CONF_HOST: parsed_url.hostname, CONF_PORT: parsed_url.port}
)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {"host": parsed_url.hostname}
self.deconz_config = {

View File

@ -225,7 +225,6 @@ class DenonAvrFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured({CONF_HOST: self.host})
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update(
{
"title_placeholders": {

View File

@ -79,7 +79,6 @@ class DirecTVConfigFlow(ConfigFlow, domain=DOMAIN):
if discovery_info.get(ATTR_UPNP_SERIAL):
receiver_id = discovery_info[ATTR_UPNP_SERIAL][4:] # strips off RID-
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": {"name": host}})
self.discovery_info.update(

View File

@ -103,7 +103,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if friendly_hostname.endswith(chop_ending):
friendly_hostname = friendly_hostname[: -len(chop_ending)]
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
CONF_NAME: friendly_hostname,
CONF_HOST: discovery_info[CONF_HOST],

View File

@ -63,7 +63,6 @@ class ElgatoFlowHandler(ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(info.serial_number)
self._abort_if_unique_id_configured(updates={CONF_HOST: user_input[CONF_HOST]})
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update(
{
CONF_HOST: user_input[CONF_HOST],
@ -76,7 +75,6 @@ class ElgatoFlowHandler(ConfigFlow, domain=DOMAIN):
# Prepare configuration flow
return self._show_confirm_dialog()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
async def async_step_zeroconf_confirm(
self, user_input: ConfigType = None
) -> Dict[str, Any]:
@ -119,7 +117,6 @@ class ElgatoFlowHandler(ConfigFlow, domain=DOMAIN):
def _show_confirm_dialog(self) -> Dict[str, Any]:
"""Show the confirm dialog to the user."""
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
serial_number = self.context.get(CONF_SERIAL_NUMBER)
return self.async_show_form(
step_id="zeroconf_confirm",

View File

@ -48,12 +48,10 @@ class EsphomeFlowHandler(ConfigFlow, domain=DOMAIN):
@property
def _name(self):
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
return self.context.get(CONF_NAME)
@_name.setter
def _name(self, value):
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context[CONF_NAME] = value
self.context["title_placeholders"] = {"name": self._name}

View File

@ -188,6 +188,5 @@ class ForkedDaapdFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
CONF_NAME: discovery_info["properties"]["Machine Name"],
}
self.discovery_schema = vol.Schema(fill_in_schema_dict(zeroconf_data))
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": zeroconf_data})
return await self.async_step_user()

View File

@ -43,8 +43,6 @@ class FritzboxConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
def __init__(self):
"""Initialize flow."""
self._entry = None

View File

@ -165,9 +165,7 @@ class FritzBoxCallMonitorConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if result != RESULT_SUCCESS:
return self.async_abort(reason=result)
if ( # pylint: disable=no-member
self.context["source"] == config_entries.SOURCE_IMPORT
):
if self.context["source"] == config_entries.SOURCE_IMPORT:
self._phonebook_id = user_input[CONF_PHONEBOOK]
self._phonebook_name = user_input[CONF_NAME]

View File

@ -88,7 +88,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
pin = async_get_pin_from_discovery_hostname(discovery_info["hostname"])
await self._async_set_unique_id(pin)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context[CONF_IP_ADDRESS] = discovery_info["host"]
if any(

View File

@ -89,7 +89,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if self._host_already_configured(parsed_url.hostname):
return self.async_abort(reason="already_configured")
# pylint: disable=no-member
self.context["title_placeholders"] = {"name": friendly_name}
self.harmony_config = {

View File

@ -253,7 +253,6 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow):
await self.async_set_unique_id(normalize_hkid(hkid))
self._abort_if_unique_id_configured()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["hkid"] = hkid
if paired:
@ -392,7 +391,6 @@ class HomekitControllerFlowHandler(config_entries.ConfigFlow):
@callback
def _async_step_pair_show_form(self, errors=None):
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
placeholders = {"name": self.name}
self.context["title_placeholders"] = {"name": self.name}

View File

@ -70,10 +70,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
CONF_URL,
default=user_input.get(
CONF_URL,
# https://github.com/PyCQA/pylint/issues/3167
self.context.get( # pylint: disable=no-member
CONF_URL, ""
),
self.context.get(CONF_URL, ""),
),
),
str,
@ -192,7 +189,6 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
title = info.get("DeviceName")
return title or DEFAULT_DEVICE_NAME
assert self.hass is not None
try:
conn = await self.hass.async_add_executor_job(try_connect, user_input)
except LoginErrorUsernameWrongException:
@ -218,7 +214,6 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
user_input=user_input, errors=errors
)
# pylint: disable=no-member
title = self.context.get("title_placeholders", {}).get(
CONF_NAME
) or await self.hass.async_add_executor_job(get_router_title, conn)
@ -238,8 +233,7 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if "mobile" not in discovery_info.get(ssdp.ATTR_UPNP_FRIENDLY_NAME, "").lower():
return self.async_abort(reason="not_huawei_lte")
# https://github.com/PyCQA/pylint/issues/3167
url = self.context[CONF_URL] = url_normalize( # pylint: disable=no-member
url = self.context[CONF_URL] = url_normalize(
discovery_info.get(
ssdp.ATTR_UPNP_PRESENTATION_URL,
f"http://{urlparse(discovery_info[ssdp.ATTR_SSDP_LOCATION]).hostname}/",
@ -255,7 +249,6 @@ class ConfigFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
if self._already_configured(user_input):
return self.async_abort(reason="already_configured")
# pylint: disable=no-member
self.context["title_placeholders"] = {
CONF_NAME: discovery_info.get(ssdp.ATTR_UPNP_FRIENDLY_NAME)
}

View File

@ -270,7 +270,6 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
auth_resp = await hyperion_client.async_request_token(
comment=DEFAULT_ORIGIN, id=auth_id
)
assert self.hass
await self.hass.config_entries.flow.async_configure(
flow_id=self.flow_id, user_input=auth_resp
)
@ -344,7 +343,6 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
# Start a task in the background requesting a new token. The next step will
# wait on the response (which includes the user needing to visit the Hyperion
# UI to approve the request for a new token).
assert self.hass
assert self._auth_id is not None
self._request_token_task = self.hass.async_create_task(
self._request_token_task_func(self._auth_id)
@ -414,9 +412,7 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
entry = await self.async_set_unique_id(hyperion_id, raise_on_progress=False)
# pylint: disable=no-member
if self.context.get(CONF_SOURCE) == SOURCE_REAUTH and entry is not None:
assert self.hass
self.hass.config_entries.async_update_entry(entry, data=self._data)
# Need to manually reload, as the listener won't have been installed because
# the initial load did not succeed (the reauth flow will not be initiated if
@ -426,7 +422,6 @@ class HyperionConfigFlow(ConfigFlow, domain=DOMAIN):
self._abort_if_unique_id_configured()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
return self.async_create_entry(
title=f"{self._data[CONF_HOST]}:{self._data[CONF_PORT]}", data=self._data
)

View File

@ -106,7 +106,6 @@ class IPPFlowHandler(ConfigFlow, domain=DOMAIN):
tls = zctype == "_ipps._tcp.local."
base_path = discovery_info["properties"].get("rp", "ipp/print")
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": {"name": name}})
self.discovery_info.update(

View File

@ -168,7 +168,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
CONF_HOST: url,
}
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = self.discovered_conf
return await self.async_step_user()

View File

@ -119,7 +119,6 @@ class KodiConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
}
)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": {CONF_NAME: self._name}})
try:

View File

@ -169,8 +169,6 @@ class KonnectedFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
# class variable to store/share discovered host information
discovered_hosts = {}
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
def __init__(self):
"""Initialize the Konnected flow."""
self.data = {}

View File

@ -77,7 +77,6 @@ class LutronCasetaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self._abort_if_unique_id_configured({CONF_HOST: host})
self.data[CONF_HOST] = host
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
CONF_NAME: self.bridge_id,
CONF_HOST: host,
@ -201,8 +200,6 @@ class LutronCasetaFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
async def async_step_import_failed(self, user_input=None):
"""Make failed import surfaced to user."""
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {CONF_NAME: self.data[CONF_HOST]}
if user_input is None:

View File

@ -129,7 +129,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
"""Prepare configuration for a discovered nut device."""
self.discovery_info = discovery_info
await self._async_handle_discovery_without_unique_id()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
CONF_PORT: discovery_info.get(CONF_PORT, DEFAULT_PORT),
CONF_HOST: discovery_info[CONF_HOST],

View File

@ -62,7 +62,6 @@ class OVOEnergyFlowHandler(ConfigFlow, domain=DOMAIN):
if user_input and user_input.get(CONF_USERNAME):
self.username = user_input[CONF_USERNAME]
# pylint: disable=no-member
self.context["title_placeholders"] = {CONF_USERNAME: self.username}
if user_input is not None and user_input.get(CONF_PASSWORD) is not None:

View File

@ -230,10 +230,7 @@ class PlexFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
}
entry = await self.async_set_unique_id(server_id)
if (
self.context[CONF_SOURCE] # pylint: disable=no-member
== config_entries.SOURCE_REAUTH
):
if self.context[CONF_SOURCE] == config_entries.SOURCE_REAUTH:
self.hass.config_entries.async_update_entry(entry, data=data)
_LOGGER.debug("Updated config entry for %s", plex_server.friendly_name)
await self.hass.config_entries.async_reload(entry.entry_id)
@ -280,7 +277,7 @@ class PlexFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self._abort_if_unique_id_configured()
host = f"{discovery_info['from'][0]}:{discovery_info['data']['Port']}"
name = discovery_info["data"]["Name"]
self.context["title_placeholders"] = { # pylint: disable=no-member
self.context["title_placeholders"] = {
"host": host,
"name": name,
}

View File

@ -97,7 +97,6 @@ class PlugwiseConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
_version = _properties.get("version", "n/a")
_name = f"{ZEROCONF_MAP.get(_product, _product)} v{_version}"
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
CONF_HOST: discovery_info[CONF_HOST],
CONF_PORT: discovery_info.get(CONF_PORT, DEFAULT_PORT),

View File

@ -65,7 +65,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_abort(reason="already_configured")
self.ip_address = dhcp_discovery[IP_ADDRESS]
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {CONF_IP_ADDRESS: self.ip_address}
return await self.async_step_user()

View File

@ -109,7 +109,6 @@ class RokuConfigFlow(ConfigFlow, domain=DOMAIN):
updates={CONF_HOST: discovery_info[CONF_HOST]},
)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": {"name": info["title"]}})
self.discovery_info.update({CONF_NAME: info["title"]})
@ -126,7 +125,6 @@ class RokuConfigFlow(ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(serial_number)
self._abort_if_unique_id_configured(updates={CONF_HOST: host})
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": {"name": name}})
self.discovery_info.update({CONF_HOST: host, CONF_NAME: name})
@ -146,7 +144,6 @@ class RokuConfigFlow(ConfigFlow, domain=DOMAIN):
self, user_input: Optional[Dict] = None
) -> Dict[str, Any]:
"""Handle user-confirmation of discovered device."""
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
if user_input is None:
return self.async_show_form(
step_id="discovery_confirm",

View File

@ -91,7 +91,6 @@ class RoombaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.host = dhcp_discovery[IP_ADDRESS]
self.blid = blid
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {"host": self.host, "name": self.blid}
return await self.async_step_user()
@ -133,7 +132,6 @@ class RoombaConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
}
if self.host and self.host in self.discovered_robots:
# From discovery
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
"host": self.host,
"name": self.discovered_robots[self.host].robot_name,

View File

@ -51,8 +51,6 @@ class SamsungTVConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1
CONNECTION_CLASS = config_entries.CONN_CLASS_LOCAL_POLL
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
def __init__(self):
"""Initialize flow."""
self._host = None

View File

@ -166,7 +166,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
except HTTP_CONNECT_ERRORS:
return self.async_abort(reason="cannot_connect")
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
"name": zeroconf_info.get("name", "").split(".")[0]
}

View File

@ -56,7 +56,6 @@ class SmappeeFlowHandler(
if self.is_cloud_device_already_added():
return self.async_abort(reason="already_configured_device")
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update(
{
CONF_IP_ADDRESS: discovery_info["host"],
@ -76,7 +75,6 @@ class SmappeeFlowHandler(
return self.async_abort(reason="already_configured_device")
if user_input is None:
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
serialnumber = self.context.get(CONF_SERIALNUMBER)
return self.async_show_form(
step_id="zeroconf_confirm",
@ -84,7 +82,6 @@ class SmappeeFlowHandler(
errors=errors,
)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
ip_address = self.context.get(CONF_IP_ADDRESS)
serial_number = self.context.get(CONF_SERIALNUMBER)

View File

@ -27,7 +27,7 @@ async def get_imei_from_config(hass: core.HomeAssistant, data):
raise CannotConnect
try:
imei = await gateway.get_imei_async()
except gammu.GSMError as err: # pylint: disable=no-member
except gammu.GSMError as err:
raise CannotConnect from err
finally:
await gateway.terminate_async()

View File

@ -72,7 +72,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.host = dhcp_discovery[HOSTNAME]
self.mac = formatted_mac
self.ip_address = dhcp_discovery[IP_ADDRESS]
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {"ip": self.ip_address, "mac": self.mac}
return await self.async_step_user()

View File

@ -114,7 +114,6 @@ class SongpalConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if "videoScreen" in service_types:
return self.async_abort(reason="not_songpal_device")
# pylint: disable=no-member
self.context["title_placeholders"] = {
CONF_NAME: friendly_name,
CONF_HOST: parsed_url.hostname,

View File

@ -63,7 +63,6 @@ class SpotifyFlowHandler(
if entry:
self.entry = entry
assert self.hass
persistent_notification.async_create(
self.hass,
f"Spotify integration for account {entry['id']} needs to be re-authenticated. Please go to the integrations page to re-configure it.",
@ -85,7 +84,6 @@ class SpotifyFlowHandler(
errors={},
)
assert self.hass
persistent_notification.async_dismiss(self.hass, "spotify_reauth")
return await self.async_step_pick_implementation(

View File

@ -182,7 +182,6 @@ class SqueezeboxConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
# update schema with suggested values from discovery
self.data_schema = _base_schema(discovery_info)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": {"host": discovery_info[CONF_HOST]}})
return await self.async_step_edit()

View File

@ -63,9 +63,7 @@ class SyncThruConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.name = re.sub(r"\s+\([\d.]+\)\s*$", "", self.name)
# https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = { # pylint: disable=no-member
CONF_NAME: self.name
}
self.context["title_placeholders"] = {CONF_NAME: self.name}
return await self.async_step_confirm()
async def async_step_confirm(self, user_input=None):

View File

@ -208,7 +208,6 @@ class SynologyDSMFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
CONF_NAME: friendly_name,
CONF_HOST: parsed_url.hostname,
}
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = self.discovered_conf
return await self.async_step_user()

View File

@ -56,7 +56,6 @@ class ToonFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
"""
if config is not None and CONF_MIGRATE in config:
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({CONF_MIGRATE: config[CONF_MIGRATE]})
else:
await self._async_handle_discovery_without_unique_id()
@ -87,10 +86,7 @@ class ToonFlowHandler(AbstractOAuth2FlowHandler, domain=DOMAIN):
return await self._create_entry(self.agreements[agreement_index])
async def _create_entry(self, agreement: Agreement) -> Dict[str, Any]:
if ( # pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
CONF_MIGRATE in self.context
):
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
if CONF_MIGRATE in self.context:
await self.hass.config_entries.async_remove(self.context[CONF_MIGRATE])
await self.async_set_unique_id(agreement.agreement_id)

View File

@ -195,7 +195,6 @@ class UnifiFlowHandler(config_entries.ConfigFlow, domain=UNIFI_DOMAIN):
"""Trigger a reauthentication flow."""
self.reauth_config_entry = config_entry
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
CONF_HOST: config_entry.data[CONF_HOST],
CONF_SITE_ID: config_entry.title,
@ -229,7 +228,6 @@ class UnifiFlowHandler(config_entries.ConfigFlow, domain=UNIFI_DOMAIN):
await self.async_set_unique_id(mac_address)
self._abort_if_unique_id_configured(updates=self.config)
# pylint: disable=no-member
self.context["title_placeholders"] = {
CONF_HOST: self.config[CONF_HOST],
CONF_SITE_ID: DEFAULT_SITE_ID,

View File

@ -183,7 +183,6 @@ class UpnpFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self._discoveries = [discovery]
# Ensure user recognizable.
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {
"name": discovery[DISCOVERY_NAME],
}

View File

@ -206,7 +206,6 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self, user_input: Dict[str, Any] = None
) -> Dict[str, Any]:
"""Handle a flow initialized by the user."""
assert self.hass
errors = {}
if user_input is not None:
@ -232,7 +231,6 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors[CONF_HOST] = "existing_config_entry_found"
if not errors:
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
if self._must_show_form and self.context["source"] == SOURCE_ZEROCONF:
# Discovery should always display the config form before trying to
# create entry so that user can update default config options
@ -251,7 +249,6 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if not errors:
return await self._create_entry(user_input)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
elif self._must_show_form and self.context["source"] == SOURCE_IMPORT:
# Import should always display the config form if CONF_ACCESS_TOKEN
# wasn't included but is needed so that the user can choose to update
@ -271,7 +268,6 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
schema = self._user_schema or _get_config_schema()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
if errors and self.context["source"] == SOURCE_IMPORT:
# Log an error message if import config flow fails since otherwise failure is silent
_LOGGER.error(
@ -346,8 +342,6 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self, discovery_info: Optional[DiscoveryInfoType] = None
) -> Dict[str, Any]:
"""Handle zeroconf discovery."""
assert self.hass
# If host already has port, no need to add it again
if ":" not in discovery_info[CONF_HOST]:
discovery_info[
@ -432,7 +426,6 @@ class VizioConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self._data[CONF_ACCESS_TOKEN] = pair_data.auth_token
self._must_show_form = True
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
if self.context["source"] == SOURCE_IMPORT:
# If user is pairing via config import, show different message
return await self.async_step_pairing_complete_import()

View File

@ -84,7 +84,6 @@ class WiLightFlowHandler(ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(self._serial_number)
self._abort_if_unique_id_configured(updates={CONF_HOST: self._host})
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context["title_placeholders"] = {"name": self._title}
return await self.async_step_confirm()

View File

@ -48,7 +48,6 @@ class WithingsFlowHandler(
async def async_step_profile(self, data: dict) -> dict:
"""Prompt the user to select a user profile."""
errors = {}
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
reauth_profile = (
self.context.get(const.PROFILE)
if self.context.get("source") == "reauth"
@ -81,14 +80,12 @@ class WithingsFlowHandler(
if data is not None:
return await self.async_step_user()
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
placeholders = {const.PROFILE: self.context["profile"]}
self.context.update({"title_placeholders": placeholders})
return self.async_show_form(
step_id="reauth",
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
description_placeholders=placeholders,
)

View File

@ -39,7 +39,6 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN):
host = user_input["hostname"].rstrip(".")
name, _ = host.rsplit(".")
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update(
{
CONF_HOST: user_input["host"],
@ -62,7 +61,6 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN):
self, user_input: Optional[ConfigType] = None, prepare: bool = False
) -> Dict[str, Any]:
"""Config flow handler for WLED."""
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
source = self.context.get("source")
# Request user input, unless we are preparing discovery flow
@ -72,7 +70,6 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN):
return self._show_setup_form()
if source == SOURCE_ZEROCONF:
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
user_input[CONF_HOST] = self.context.get(CONF_HOST)
user_input[CONF_MAC] = self.context.get(CONF_MAC)
@ -93,7 +90,6 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN):
title = user_input[CONF_HOST]
if source == SOURCE_ZEROCONF:
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
title = self.context.get(CONF_NAME)
if prepare:
@ -114,7 +110,6 @@ class WLEDFlowHandler(ConfigFlow, domain=DOMAIN):
def _show_confirm_dialog(self, errors: Optional[Dict] = None) -> Dict[str, Any]:
"""Show the confirm dialog to the user."""
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
name = self.context.get(CONF_NAME)
return self.async_show_form(
step_id="zeroconf_confirm",

View File

@ -181,7 +181,6 @@ class XiaomiAqaraFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
{CONF_HOST: self.host, CONF_MAC: mac_address}
)
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update({"title_placeholders": {"name": self.host}})
return await self.async_step_user()

View File

@ -67,7 +67,6 @@ class XiaomiMiioFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
await self.async_set_unique_id(unique_id)
self._abort_if_unique_id_configured({CONF_HOST: self.host})
# pylint: disable=no-member # https://github.com/PyCQA/pylint/issues/3167
self.context.update(
{"title_placeholders": {"name": f"Gateway {self.host}"}}
)

View File

@ -84,7 +84,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Handle the initial step."""
assert self.hass # typing
if self.hass.components.hassio.is_hassio():
return await self.async_step_on_supervisor()
@ -101,7 +100,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
errors = {}
assert self.hass # typing
try:
version_info = await validate_input(self.hass, user_input)
except InvalidInput as err:
@ -128,7 +126,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
This flow is triggered by the Z-Wave JS add-on.
"""
assert self.hass
self.ws_address = f"ws://{discovery_info['host']}:{discovery_info['port']}"
try:
version_info = await async_get_version_info(self.hass, self.ws_address)
@ -182,7 +179,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self.ws_address = f"ws://{discovery_info['host']}:{discovery_info['port']}"
if not self.unique_id:
assert self.hass
try:
version_info = await async_get_version_info(
self.hass, self.ws_address
@ -208,14 +204,12 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
self, user_input: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""Install Z-Wave JS add-on."""
assert self.hass
if not self.install_task:
self.install_task = self.hass.async_create_task(self._async_install_addon())
return self.async_show_progress(
step_id="install_addon", progress_action="install_addon"
)
assert self.hass
try:
await self.install_task
except self.hass.components.hassio.HassioAPIError as err:
@ -253,7 +247,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
if new_addon_config != self.addon_config:
await self._async_set_addon_config(new_addon_config)
assert self.hass
try:
await self.hass.components.hassio.async_start_addon("core_zwave_js")
except self.hass.components.hassio.HassioAPIError as err:
@ -299,7 +292,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_get_addon_info(self) -> dict:
"""Return and cache Z-Wave JS add-on info."""
assert self.hass
try:
addon_info: dict = await self.hass.components.hassio.async_get_addon_info(
"core_zwave_js"
@ -327,7 +319,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_set_addon_config(self, config: dict) -> None:
"""Set Z-Wave JS add-on config."""
assert self.hass
options = {"options": config}
try:
await self.hass.components.hassio.async_set_addon_options(
@ -339,7 +330,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_install_addon(self) -> None:
"""Install the Z-Wave JS add-on."""
assert self.hass
try:
await self.hass.components.hassio.async_install_addon("core_zwave_js")
finally:
@ -350,7 +340,6 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
async def _async_get_addon_discovery_info(self) -> dict:
"""Return add-on discovery info."""
assert self.hass
try:
discovery_info: dict = (
await self.hass.components.hassio.async_get_addon_discovery_info(

View File

@ -903,7 +903,6 @@ class ConfigFlow(data_entry_flow.FlowHandler):
reload_on_update: bool = True,
) -> None:
"""Abort if the unique ID is already configured."""
assert self.hass
if self.unique_id is None:
return
@ -945,7 +944,6 @@ class ConfigFlow(data_entry_flow.FlowHandler):
self.context["unique_id"] = unique_id # pylint: disable=no-member
# Abort discoveries done using the default discovery unique id
assert self.hass is not None
if unique_id != DEFAULT_DISCOVERY_UNIQUE_ID:
for progress in self._async_in_progress():
if progress["context"].get("unique_id") == DEFAULT_DISCOVERY_UNIQUE_ID:
@ -963,7 +961,6 @@ class ConfigFlow(data_entry_flow.FlowHandler):
If the flow is user initiated, filter out ignored entries unless include_ignore is True.
"""
assert self.hass is not None
config_entries = self.hass.config_entries.async_entries(self.handler)
if include_ignore or self.source != SOURCE_USER:
@ -974,7 +971,6 @@ class ConfigFlow(data_entry_flow.FlowHandler):
@callback
def _async_current_ids(self, include_ignore: bool = True) -> Set[Optional[str]]:
"""Return current unique IDs."""
assert self.hass is not None
return {
entry.unique_id
for entry in self.hass.config_entries.async_entries(self.handler)
@ -984,7 +980,6 @@ class ConfigFlow(data_entry_flow.FlowHandler):
@callback
def _async_in_progress(self) -> List[Dict]:
"""Return other in progress flows for current domain."""
assert self.hass is not None
return [
flw
for flw in self.hass.config_entries.flow.async_progress()
@ -1027,7 +1022,6 @@ class ConfigFlow(data_entry_flow.FlowHandler):
self._abort_if_unique_id_configured()
# Abort if any other flow for this handler is already in progress
assert self.hass is not None
if self._async_in_progress():
raise data_entry_flow.AbortFlow("already_in_progress")
@ -1043,8 +1037,6 @@ class ConfigFlow(data_entry_flow.FlowHandler):
self, *, reason: str, description_placeholders: Optional[Dict] = None
) -> Dict[str, Any]:
"""Abort the config flow."""
assert self.hass
# Remove reauth notification if no reauth flows are in progress
if self.source == SOURCE_REAUTH and not any(
ent["context"]["source"] == SOURCE_REAUTH

View File

@ -3,7 +3,8 @@ from __future__ import annotations
import abc
import asyncio
from typing import Any, Dict, List, Optional, cast
from types import MappingProxyType
from typing import Any, Dict, List, Optional
import uuid
import voluptuous as vol
@ -264,11 +265,14 @@ class FlowHandler:
"""Handle the configuration flow of a component."""
# Set by flow manager
flow_id: str = None # type: ignore
hass: Optional[HomeAssistant] = None
handler: Optional[str] = None
cur_step: Optional[Dict[str, str]] = None
context: Dict
# Ignore types, pylint workaround: https://github.com/PyCQA/pylint/issues/3167
flow_id: str = None # type: ignore
hass: HomeAssistant = None # type: ignore
handler: str = None # type: ignore
# Pylint workaround: https://github.com/PyCQA/pylint/issues/3167
# Ensure the attribute has a subscriptable, but immutable, default value.
context: Dict = MappingProxyType({}) # type: ignore
# Set by _async_create_flow callback
init_step = "init"
@ -339,7 +343,7 @@ class FlowHandler:
) -> Dict[str, Any]:
"""Abort the config flow."""
return _create_abort_data(
self.flow_id, cast(str, self.handler), reason, description_placeholders
self.flow_id, self.handler, reason, description_placeholders
)
@callback