Improve config flow type hints (a-f) (#124859)

This commit is contained in:
epenet 2024-08-29 17:24:04 +02:00 committed by GitHub
parent 34680becaa
commit 681fe3485d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 53 additions and 32 deletions

View File

@ -5,7 +5,7 @@ import errno
from functools import partial from functools import partial
import logging import logging
import socket import socket
from typing import TYPE_CHECKING, Any from typing import Any
import broadlink as blk import broadlink as blk
from broadlink.exceptions import ( from broadlink.exceptions import (
@ -37,9 +37,7 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self) -> None: device: blk.Device
"""Initialize the Broadlink flow."""
self.device: blk.Device | None = None
async def async_set_device( async def async_set_device(
self, device: blk.Device, raise_on_progress: bool = True self, device: blk.Device, raise_on_progress: bool = True
@ -131,8 +129,6 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
) )
return await self.async_step_auth() return await self.async_step_auth()
if TYPE_CHECKING:
assert self.device
if device.mac == self.device.mac: if device.mac == self.device.mac:
await self.async_set_device(device, raise_on_progress=False) await self.async_set_device(device, raise_on_progress=False)
return await self.async_step_auth() return await self.async_step_auth()
@ -158,10 +154,10 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
errors=errors, errors=errors,
) )
async def async_step_auth(self): async def async_step_auth(self) -> ConfigFlowResult:
"""Authenticate to the device.""" """Authenticate to the device."""
device = self.device device = self.device
errors = {} errors: dict[str, str] = {}
try: try:
await self.hass.async_add_executor_job(device.auth) await self.hass.async_add_executor_job(device.auth)
@ -211,7 +207,11 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
) )
return self.async_show_form(step_id="auth", errors=errors) return self.async_show_form(step_id="auth", errors=errors)
async def async_step_reset(self, user_input=None, errors=None): async def async_step_reset(
self,
user_input: dict[str, Any] | None = None,
errors: dict[str, str] | None = None,
) -> ConfigFlowResult:
"""Guide the user to unlock the device manually. """Guide the user to unlock the device manually.
We are unable to authenticate because the device is locked. We are unable to authenticate because the device is locked.
@ -234,7 +234,9 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
{CONF_HOST: device.host[0], CONF_TIMEOUT: device.timeout} {CONF_HOST: device.host[0], CONF_TIMEOUT: device.timeout}
) )
async def async_step_unlock(self, user_input=None): async def async_step_unlock(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Unlock the device. """Unlock the device.
The authentication succeeded, but the device is locked. The authentication succeeded, but the device is locked.
@ -288,10 +290,12 @@ class BroadlinkFlowHandler(ConfigFlow, domain=DOMAIN):
}, },
) )
async def async_step_finish(self, user_input=None): async def async_step_finish(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Choose a name for the device and create config entry.""" """Choose a name for the device and create config entry."""
device = self.device device = self.device
errors = {} errors: dict[str, str] = {}
# Abort reauthentication flow. # Abort reauthentication flow.
self._abort_if_unique_id_configured( self._abort_if_unique_id_configured(

View File

@ -3,7 +3,7 @@
from __future__ import annotations from __future__ import annotations
import logging import logging
from typing import Any from typing import TYPE_CHECKING, Any
from aiohttp.client_exceptions import ClientError from aiohttp.client_exceptions import ClientError
from pyControl4.account import C4Account from pyControl4.account import C4Account
@ -23,7 +23,7 @@ from homeassistant.const import (
CONF_SCAN_INTERVAL, CONF_SCAN_INTERVAL,
CONF_USERNAME, CONF_USERNAME,
) )
from homeassistant.core import callback from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import aiohttp_client, config_validation as cv from homeassistant.helpers import aiohttp_client, config_validation as cv
from homeassistant.helpers.device_registry import format_mac from homeassistant.helpers.device_registry import format_mac
@ -49,7 +49,9 @@ DATA_SCHEMA = vol.Schema(
class Control4Validator: class Control4Validator:
"""Validates that config details can be used to authenticate and communicate with Control4.""" """Validates that config details can be used to authenticate and communicate with Control4."""
def __init__(self, host, username, password, hass): def __init__(
self, host: str, username: str, password: str, hass: HomeAssistant
) -> None:
"""Initialize.""" """Initialize."""
self.host = host self.host = host
self.username = username self.username = username
@ -126,6 +128,8 @@ class Control4ConfigFlow(ConfigFlow, domain=DOMAIN):
if not errors: if not errors:
controller_unique_id = hub.controller_unique_id controller_unique_id = hub.controller_unique_id
if TYPE_CHECKING:
assert hub.controller_unique_id
mac = (controller_unique_id.split("_", 3))[2] mac = (controller_unique_id.split("_", 3))[2]
formatted_mac = format_mac(mac) formatted_mac = format_mac(mac)
await self.async_set_unique_id(formatted_mac) await self.async_set_unique_id(formatted_mac)
@ -160,7 +164,9 @@ class OptionsFlowHandler(OptionsFlow):
"""Initialize options flow.""" """Initialize options flow."""
self.config_entry = config_entry self.config_entry = config_entry
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.""" """Handle options flow."""
if user_input is not None: if user_input is not None:
return self.async_create_entry(title="", data=user_input) return self.async_create_entry(title="", data=user_input)

View File

@ -79,7 +79,9 @@ class DexcomOptionsFlowHandler(OptionsFlow):
"""Initialize options flow.""" """Initialize options flow."""
self.config_entry = config_entry self.config_entry = config_entry
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.""" """Handle options flow."""
if user_input is not None: if user_input is not None:
return self.async_create_entry(title="", data=user_input) return self.async_create_entry(title="", data=user_input)

View File

@ -23,9 +23,7 @@ class EcobeeFlowHandler(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
def __init__(self) -> None: _ecobee: Ecobee
"""Initialize the ecobee flow."""
self._ecobee: Ecobee | None = None
async def async_step_user( async def async_step_user(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
@ -59,7 +57,9 @@ class EcobeeFlowHandler(ConfigFlow, domain=DOMAIN):
errors=errors, errors=errors,
) )
async def async_step_authorize(self, user_input=None): async def async_step_authorize(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Present the user with the PIN so that the app can be authorized on ecobee.com.""" """Present the user with the PIN so that the app can be authorized on ecobee.com."""
errors = {} errors = {}

View File

@ -34,10 +34,11 @@ class EmonitorConfigFlow(ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
discovered_info: dict[str, str]
def __init__(self) -> None: def __init__(self) -> None:
"""Initialize Emonitor ConfigFlow.""" """Initialize Emonitor ConfigFlow."""
self.discovered_ip: str | None = None self.discovered_ip: str | None = None
self.discovered_info: dict[str, str] | None = None
async def async_step_user( async def async_step_user(
self, user_input: dict[str, Any] | None = None self, user_input: dict[str, Any] | None = None
@ -87,7 +88,9 @@ class EmonitorConfigFlow(ConfigFlow, domain=DOMAIN):
return await self.async_step_user() return await self.async_step_user()
return await self.async_step_confirm() 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.""" """Attempt to confirm."""
if user_input is not None: if user_input is not None:
return self.async_create_entry( return self.async_create_entry(

View File

@ -6,13 +6,13 @@ import voluptuous as vol
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
from homeassistant.const import CONF_NAME from homeassistant.const import CONF_NAME
from homeassistant.core import callback from homeassistant.core import HomeAssistant, callback
from .const import CONF_LISTEN_PORT, DEFAULT_NAME, DEFAULT_PORT, DOMAIN from .const import CONF_LISTEN_PORT, DEFAULT_NAME, DEFAULT_PORT, DOMAIN
@callback @callback
def configured_servers(hass): def configured_servers(hass: HomeAssistant) -> set[str]:
"""Return a set of the configured servers.""" """Return a set of the configured servers."""
return { return {
entry.data[CONF_NAME] for entry in hass.config_entries.async_entries(DOMAIN) entry.data[CONF_NAME] for entry in hass.config_entries.async_entries(DOMAIN)

View File

@ -43,12 +43,14 @@ class EnOceanFlowHandler(ConfigFlow, domain=DOMAIN):
return await self.async_step_detect() return await self.async_step_detect()
async def async_step_detect(self, user_input=None): async def async_step_detect(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Propose a list of detected dongles.""" """Propose a list of detected dongles."""
errors = {} errors = {}
if user_input is not None: if user_input is not None:
if user_input[CONF_DEVICE] == self.MANUAL_PATH_VALUE: if user_input[CONF_DEVICE] == self.MANUAL_PATH_VALUE:
return await self.async_step_manual(None) return await self.async_step_manual()
if await self.validate_enocean_conf(user_input): if await self.validate_enocean_conf(user_input):
return self.create_enocean_entry(user_input) return self.create_enocean_entry(user_input)
errors = {CONF_DEVICE: ERROR_INVALID_DONGLE_PATH} errors = {CONF_DEVICE: ERROR_INVALID_DONGLE_PATH}
@ -64,7 +66,9 @@ class EnOceanFlowHandler(ConfigFlow, domain=DOMAIN):
errors=errors, errors=errors,
) )
async def async_step_manual(self, user_input=None): async def async_step_manual(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Request manual USB dongle path.""" """Request manual USB dongle path."""
default_value = None default_value = None
errors = {} errors = {}

View File

@ -56,7 +56,9 @@ class ForkedDaapdOptionsFlowHandler(OptionsFlow):
"""Initialize.""" """Initialize."""
self.config_entry = config_entry self.config_entry = config_entry
async def async_step_init(self, user_input=None): async def async_step_init(
self, user_input: dict[str, Any] | None = None
) -> ConfigFlowResult:
"""Manage the options.""" """Manage the options."""
if user_input is not None: if user_input is not None:
return self.async_create_entry(title="options", data=user_input) return self.async_create_entry(title="options", data=user_input)

View File

@ -19,19 +19,19 @@ STEP_USER_DATA_SCHEMA = vol.Schema({vol.Required(CONF_API_KEY): str})
class Hub: class Hub:
"""Freedompro Hub class.""" """Freedompro Hub class."""
def __init__(self, hass, api_key): def __init__(self, hass: HomeAssistant, api_key: str) -> None:
"""Freedompro Hub class init.""" """Freedompro Hub class init."""
self._hass = hass self._hass = hass
self._api_key = api_key self._api_key = api_key
async def authenticate(self): async def authenticate(self) -> dict[str, Any]:
"""Freedompro Hub class authenticate.""" """Freedompro Hub class authenticate."""
return await get_list( return await get_list(
aiohttp_client.async_get_clientsession(self._hass), self._api_key aiohttp_client.async_get_clientsession(self._hass), self._api_key
) )
async def validate_input(hass: HomeAssistant, api_key): async def validate_input(hass: HomeAssistant, api_key: str) -> None:
"""Validate api key.""" """Validate api key."""
hub = Hub(hass, api_key) hub = Hub(hass, api_key)
result = await hub.authenticate() result = await hub.authenticate()