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

Improve typing for synology_dsm (#49656)

Browse source
This commit is contained in:
Michael 2021-05-09 22:44:55 +02:00 • committed by GitHub
parent 717f4e69d5
commit 042822e35e
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 241 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

70
homeassistant/components/synology_dsm/config_flow.py
View file

@ -1,5 +1,8 @@
"""Config flow to configure the Synology DSM integration."""
from __future__ import annotations
import logging
from typing import Any
from urllib.parse import urlparse
from synology_dsm import SynologyDSM
@ -12,8 +15,14 @@ from synology_dsm.exceptions import (
)
import voluptuous as vol
from homeassistant import config_entries, exceptions
from homeassistant import exceptions
from homeassistant.components import ssdp
from homeassistant.config_entries import (
CONN_CLASS_CLOUD_POLL,
ConfigEntry,
ConfigFlow,
OptionsFlow,
)
from homeassistant.const import (
CONF_DISKS,
CONF_HOST,
@ -28,7 +37,9 @@ from homeassistant.const import (
CONF_VERIFY_SSL,
)
from homeassistant.core import callback
from homeassistant.data_entry_flow import FlowResult
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import DiscoveryInfoType
from .const import (
CONF_DEVICE_TOKEN,
@ -47,11 +58,11 @@ _LOGGER = logging.getLogger(__name__)
CONF_OTP_CODE = "otp_code"
def _discovery_schema_with_defaults(discovery_info):
def _discovery_schema_with_defaults(discovery_info: DiscoveryInfoType) -> vol.Schema:
return vol.Schema(_ordered_shared_schema(discovery_info))
def _user_schema_with_defaults(user_input):
def _user_schema_with_defaults(user_input: dict[str, Any]) -> vol.Schema:
user_schema = {
vol.Required(CONF_HOST, default=user_input.get(CONF_HOST, "")): str,
}
@ -60,7 +71,9 @@ def _user_schema_with_defaults(user_input):
return vol.Schema(user_schema)
def _ordered_shared_schema(schema_input):
def _ordered_shared_schema(
schema_input: dict[str, Any]
) -> dict[vol.Required | vol.Optional, Any]:
return {
vol.Required(CONF_USERNAME, default=schema_input.get(CONF_USERNAME, "")): str,
vol.Required(CONF_PASSWORD, default=schema_input.get(CONF_PASSWORD, "")): str,
@ -75,23 +88,30 @@ def _ordered_shared_schema(schema_input):
}
class SynologyDSMFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
class SynologyDSMFlowHandler(ConfigFlow, domain=DOMAIN):
"""Handle a config flow."""
VERSION = 1
CONNECTION_CLASS = CONN_CLASS_CLOUD_POLL
@staticmethod
@callback
def async_get_options_flow(config_entry):
def async_get_options_flow(
config_entry: ConfigEntry,
) -> SynologyDSMOptionsFlowHandler:
"""Get the options flow for this handler."""
return SynologyDSMOptionsFlowHandler(config_entry)
def __init__(self):
def __init__(self) -> None:
"""Initialize the synology_dsm config flow."""
self.saved_user_input = {}
self.discovered_conf = {}
self.saved_user_input: dict[str, Any] = {}
self.discovered_conf: dict[str, Any] = {}
async def _show_setup_form(self, user_input=None, errors=None):
async def _show_setup_form(
self,
user_input: dict[str, Any] | None = None,
errors: dict[str, str] | None = None,
) -> FlowResult:
"""Show the setup form to the user."""
if not user_input:
user_input = {}
@ -111,7 +131,9 @@ class SynologyDSMFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
description_placeholders=self.discovered_conf or {},
)
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 initiated by the user."""
errors = {}
@ -188,7 +210,7 @@ class SynologyDSMFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return self.async_create_entry(title=host, data=config_data)
async def async_step_ssdp(self, discovery_info):
async def async_step_ssdp(self, discovery_info: DiscoveryInfoType) -> FlowResult:
"""Handle a discovered synology_dsm."""
parsed_url = urlparse(discovery_info[ssdp.ATTR_SSDP_LOCATION])
friendly_name = (
@ -211,15 +233,19 @@ class SynologyDSMFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
self.context["title_placeholders"] = self.discovered_conf
return await self.async_step_user()
async def async_step_import(self, user_input=None):
async def async_step_import(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Import a config entry."""
return await self.async_step_user(user_input)
async def async_step_link(self, user_input):
async def async_step_link(self, user_input: dict[str, Any]) -> FlowResult:
"""Link a config entry from discovery."""
return await self.async_step_user(user_input)
async def async_step_2sa(self, user_input, errors=None):
async def async_step_2sa(
self, user_input: dict[str, Any], errors: dict[str, str] | None = None
) -> FlowResult:
"""Enter 2SA code to anthenticate."""
if not self.saved_user_input:
self.saved_user_input = user_input
@ -236,7 +262,7 @@ class SynologyDSMFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return await self.async_step_user(user_input)
def _mac_already_configured(self, mac):
def _mac_already_configured(self, mac: str) -> bool:
"""See if we already have configured a NAS with this MAC address."""
existing_macs = [
mac.replace("-", "")
@ -246,14 +272,16 @@ class SynologyDSMFlowHandler(config_entries.ConfigFlow, domain=DOMAIN):
return mac in existing_macs
class SynologyDSMOptionsFlowHandler(config_entries.OptionsFlow):
class SynologyDSMOptionsFlowHandler(OptionsFlow):
"""Handle a option flow."""
def __init__(self, config_entry: config_entries.ConfigEntry):
def __init__(self, config_entry: ConfigEntry):
"""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, Any] | None = None
) -> FlowResult:
"""Handle options flow."""
if user_input is not None:
return self.async_create_entry(title="", data=user_input)
@ -277,7 +305,7 @@ class SynologyDSMOptionsFlowHandler(config_entries.OptionsFlow):
return self.async_show_form(step_id="init", data_schema=data_schema)
def _login_and_fetch_syno_info(api, otp_code):
def _login_and_fetch_syno_info(api: SynologyDSM, otp_code: str) -> str:
"""Login to the NAS and fetch basic data."""
# These do i/o
api.login(otp_code)
@ -293,7 +321,7 @@ def _login_and_fetch_syno_info(api, otp_code):
):
raise InvalidData
return api.information.serial
return api.information.serial # type: ignore[no-any-return]
class InvalidData(exceptions.HomeAssistantError):