github
/
ha-core
mirror of https://github.com/home-assistant/core
1
Fork 0
Code Releases Activity

Enable strict typing for config (#108023)

This commit is contained in:
Marc Mueller 2024-01-18 09:20:19 +01:00 committed by GitHub
parent 26cc6a5bb4
commit afcb7a26cd
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 178 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.strict-typing
View File

@ -122,6 +122,7 @@ homeassistant.components.climate.*
homeassistant.components.cloud.*
homeassistant.components.co2signal.*
homeassistant.components.command_line.*
homeassistant.components.config.*
homeassistant.components.configurator.*
homeassistant.components.counter.*
homeassistant.components.cover.*

110
homeassistant/components/config/__init__.py
View File

@ -1,9 +1,14 @@
"""Component to configure Home Assistant via an API."""
from __future__ import annotations
import asyncio
from collections.abc import Callable, Coroutine
from http import HTTPStatus
import importlib
import os
from typing import Any, Generic, TypeVar, cast
from aiohttp import web
import voluptuous as vol
from homeassistant.components import frontend
@ -16,6 +21,9 @@ from homeassistant.helpers.typing import ConfigType
from homeassistant.setup import ATTR_COMPONENT
from homeassistant.util.file import write_utf8_file_atomic
from homeassistant.util.yaml import dump, load_yaml
from homeassistant.util.yaml.loader import JSON_TYPE
_DataT = TypeVar("_DataT", dict[str, dict[str, Any]], list[dict[str, Any]])
DOMAIN = "config"
SECTIONS = (
@ -42,7 +50,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
hass, "config", "config", "hass:cog", require_admin=True
)
async def setup_panel(panel_name):
async def setup_panel(panel_name: str) -> None:
"""Set up a panel."""
panel = importlib.import_module(f".{panel_name}", __name__)
@ -63,20 +71,24 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
return True
class BaseEditConfigView(HomeAssistantView):
class BaseEditConfigView(HomeAssistantView, Generic[_DataT]):
"""Configure a Group endpoint."""
def __init__(
self,
component,
config_type,
path,
key_schema,
data_schema,
component: str,
config_type: str,
path: str,
key_schema: Callable[[Any], str],
data_schema: Callable[[dict[str, Any]], Any],
*,
post_write_hook=None,
data_validator=None,
):
post_write_hook: Callable[[str, str], Coroutine[Any, Any, None]] | None = None,
data_validator: Callable[
[HomeAssistant, str, dict[str, Any]],
Coroutine[Any, Any, dict[str, Any] | None],
]
| None = None,
) -> None:
"""Initialize a config view."""
self.url = f"/api/config/{component}/{config_type}/{{config_key}}"
self.name = f"api:config:{component}:{config_type}"
@ -87,26 +99,36 @@ class BaseEditConfigView(HomeAssistantView):
self.data_validator = data_validator
self.mutation_lock = asyncio.Lock()
def _empty_config(self):
def _empty_config(self) -> _DataT:
"""Empty config if file not found."""
raise NotImplementedError
def _get_value(self, hass, data, config_key):
def _get_value(
self, hass: HomeAssistant, data: _DataT, config_key: str
) -> dict[str, Any] | None:
"""Get value."""
raise NotImplementedError
def _write_value(self, hass, data, config_key, new_value):
def _write_value(
self,
hass: HomeAssistant,
data: _DataT,
config_key: str,
new_value: dict[str, Any],
) -> None:
"""Set value."""
raise NotImplementedError
def _delete_value(self, hass, data, config_key):
def _delete_value(
self, hass: HomeAssistant, data: _DataT, config_key: str
) -> dict[str, Any] | None:
"""Delete value."""
raise NotImplementedError
@require_admin
async def get(self, request, config_key):
async def get(self, request: web.Request, config_key: str) -> web.Response:
"""Fetch device specific config."""
hass = request.app["hass"]
hass: HomeAssistant = request.app["hass"]
async with self.mutation_lock:
current = await self.read_config(hass)
value = self._get_value(hass, current, config_key)
@ -117,7 +139,7 @@ class BaseEditConfigView(HomeAssistantView):
return self.json(value)
@require_admin
async def post(self, request, config_key):
async def post(self, request: web.Request, config_key: str) -> web.Response:
"""Validate config and return results."""
try:
data = await request.json()
@ -129,7 +151,7 @@ class BaseEditConfigView(HomeAssistantView):
except vol.Invalid as err:
return self.json_message(f"Key malformed: {err}", HTTPStatus.BAD_REQUEST)
hass = request.app["hass"]
hass: HomeAssistant = request.app["hass"]
try:
# We just validate, we don't store that data because
@ -159,9 +181,9 @@ class BaseEditConfigView(HomeAssistantView):
return self.json({"result": "ok"})
@require_admin
async def delete(self, request, config_key):
async def delete(self, request: web.Request, config_key: str) -> web.Response:
"""Remove an entry."""
hass = request.app["hass"]
hass: HomeAssistant = request.app["hass"]
async with self.mutation_lock:
current = await self.read_config(hass)
value = self._get_value(hass, current, config_key)
@ -178,46 +200,64 @@ class BaseEditConfigView(HomeAssistantView):
return self.json({"result": "ok"})
async def read_config(self, hass):
async def read_config(self, hass: HomeAssistant) -> _DataT:
"""Read the config."""
current = await hass.async_add_executor_job(_read, hass.config.path(self.path))
if not current:
current = self._empty_config()
return current
return cast(_DataT, current)
class EditKeyBasedConfigView(BaseEditConfigView):
class EditKeyBasedConfigView(BaseEditConfigView[dict[str, dict[str, Any]]]):
"""Configure a list of entries."""
def _empty_config(self):
def _empty_config(self) -> dict[str, Any]:
"""Return an empty config."""
return {}
def _get_value(self, hass, data, config_key):
def _get_value(
self, hass: HomeAssistant, data: dict[str, dict[str, Any]], config_key: str
) -> dict[str, Any] | None:
"""Get value."""
return data.get(config_key)
def _write_value(self, hass, data, config_key, new_value):
def _write_value(
self,
hass: HomeAssistant,
data: dict[str, dict[str, Any]],
config_key: str,
new_value: dict[str, Any],
) -> None:
"""Set value."""
data.setdefault(config_key, {}).update(new_value)
def _delete_value(self, hass, data, config_key):
def _delete_value(
self, hass: HomeAssistant, data: dict[str, dict[str, Any]], config_key: str
) -> dict[str, Any]:
"""Delete value."""
return data.pop(config_key)
class EditIdBasedConfigView(BaseEditConfigView):
class EditIdBasedConfigView(BaseEditConfigView[list[dict[str, Any]]]):
"""Configure key based config entries."""
def _empty_config(self):
def _empty_config(self) -> list[Any]:
"""Return an empty config."""
return []
def _get_value(self, hass, data, config_key):
def _get_value(
self, hass: HomeAssistant, data: list[dict[str, Any]], config_key: str
) -> dict[str, Any] | None:
"""Get value."""
return next((val for val in data if val.get(CONF_ID) == config_key), None)
def _write_value(self, hass, data, config_key, new_value):
def _write_value(
self,
hass: HomeAssistant,
data: list[dict[str, Any]],
config_key: str,
new_value: dict[str, Any],
) -> None:
"""Set value."""
if (value := self._get_value(hass, data, config_key)) is None:
value = {CONF_ID: config_key}
@ -225,7 +265,9 @@ class EditIdBasedConfigView(BaseEditConfigView):
value.update(new_value)
def _delete_value(self, hass, data, config_key):
def _delete_value(
self, hass: HomeAssistant, data: list[dict[str, Any]], config_key: str
) -> None:
"""Delete value."""
index = next(
idx for idx, val in enumerate(data) if val.get(CONF_ID) == config_key
@ -233,7 +275,7 @@ class EditIdBasedConfigView(BaseEditConfigView):
data.pop(index)
def _read(path):
def _read(path: str) -> JSON_TYPE | None:
"""Read YAML helper."""
if not os.path.isfile(path):
return None
@ -241,7 +283,7 @@ def _read(path):
return load_yaml(path)
def _write(path, data):
def _write(path: str, data: dict | list) -> None:
"""Write YAML helper."""
# Do it before opening file. If dump causes error it will now not
# truncate the file.

8
homeassistant/components/config/area_registry.py
View File

@ -1,14 +1,16 @@
"""HTTP views to interact with the area registry."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.area_registry import async_get
from homeassistant.helpers.area_registry import AreaEntry, async_get
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Enable the Area Registry views."""
websocket_api.async_register_command(hass, websocket_list_areas)
websocket_api.async_register_command(hass, websocket_create_area)
@ -126,7 +128,7 @@ def websocket_update_area(
@callback
def _entry_dict(entry):
def _entry_dict(entry: AreaEntry) -> dict[str, Any]:
"""Convert entry to API format."""
return {
"aliases": entry.aliases,

7
homeassistant/components/config/auth.py
View File

@ -1,8 +1,11 @@
"""Offer API to configure Home Assistant auth."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
from homeassistant.auth.models import User
from homeassistant.components import websocket_api
from homeassistant.core import HomeAssistant
@ -17,7 +20,7 @@ SCHEMA_WS_DELETE = websocket_api.BASE_COMMAND_MESSAGE_SCHEMA.extend(
)
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Enable the Home Assistant views."""
websocket_api.async_register_command(
hass, WS_TYPE_LIST, websocket_list, SCHEMA_WS_LIST
@ -151,7 +154,7 @@ async def websocket_update(
)
def _user_info(user):
def _user_info(user: User) -> dict[str, Any]:
"""Format a user."""
ha_username = next(

6
homeassistant/components/config/auth_provider_homeassistant.py
View File

@ -1,4 +1,6 @@
"""Offer API to configure the Home Assistant auth provider."""
from __future__ import annotations
from typing import Any
import voluptuous as vol
@ -9,7 +11,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import Unauthorized
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Enable the Home Assistant views."""
websocket_api.async_register_command(hass, websocket_create)
websocket_api.async_register_command(hass, websocket_delete)
@ -115,7 +117,7 @@ async def websocket_change_password(
) -> None:
"""Change current user password."""
if (user := connection.user) is None:
connection.send_error(msg["id"], "user_not_found", "User not found")
connection.send_error(msg["id"], "user_not_found", "User not found") # type: ignore[unreachable]
return
provider = auth_ha.async_get_provider(hass)

16
homeassistant/components/config/automation.py
View File

@ -1,4 +1,7 @@
"""Provide configuration end points for Automations."""
from __future__ import annotations
from typing import Any
import uuid
from homeassistant.components.automation.config import (
@ -8,15 +11,16 @@ from homeassistant.components.automation.config import (
)
from homeassistant.config import AUTOMATION_CONFIG_PATH
from homeassistant.const import CONF_ID, SERVICE_RELOAD
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_registry as er
from . import ACTION_DELETE, EditIdBasedConfigView
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Set up the Automation config API."""
async def hook(action, config_key):
async def hook(action: str, config_key: str) -> None:
"""post_write_hook for Config View that reloads automations."""
await hass.services.async_call(DOMAIN, SERVICE_RELOAD)
@ -49,7 +53,13 @@ async def async_setup(hass):
class EditAutomationConfigView(EditIdBasedConfigView):
"""Edit automation config."""
def _write_value(self, hass, data, config_key, new_value):
def _write_value(
self,
hass: HomeAssistant,
data: list[dict[str, Any]],
config_key: str,
new_value: dict[str, Any],
) -> None:
"""Set value."""
updated_value = {CONF_ID: config_key}

52
homeassistant/components/config/config_entries.py
View File

@ -1,8 +1,9 @@
"""Http views to control the config manager."""
from __future__ import annotations
from collections.abc import Callable
from http import HTTPStatus
from typing import Any
from typing import Any, NoReturn
from aiohttp import web
import aiohttp.web_exceptions
@ -29,7 +30,7 @@ from homeassistant.loader import (
)
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Enable the Home Assistant views."""
hass.http.register_view(ConfigManagerEntryIndexView)
hass.http.register_view(ConfigManagerEntryResourceView)
@ -58,7 +59,7 @@ class ConfigManagerEntryIndexView(HomeAssistantView):
url = "/api/config/config_entries/entry"
name = "api:config:config_entries:entry"
async def get(self, request):
async def get(self, request: web.Request) -> web.Response:
"""List available config entries."""
hass: HomeAssistant = request.app["hass"]
domain = None
@ -76,12 +77,12 @@ class ConfigManagerEntryResourceView(HomeAssistantView):
url = "/api/config/config_entries/entry/{entry_id}"
name = "api:config:config_entries:entry:resource"
async def delete(self, request, entry_id):
async def delete(self, request: web.Request, entry_id: str) -> web.Response:
"""Delete a config entry."""
if not request["hass_user"].is_admin:
raise Unauthorized(config_entry_id=entry_id, permission="remove")
hass = request.app["hass"]
hass: HomeAssistant = request.app["hass"]
try:
result = await hass.config_entries.async_remove(entry_id)
@ -97,12 +98,12 @@ class ConfigManagerEntryResourceReloadView(HomeAssistantView):
url = "/api/config/config_entries/entry/{entry_id}/reload"
name = "api:config:config_entries:entry:resource:reload"
async def post(self, request, entry_id):
async def post(self, request: web.Request, entry_id: str) -> web.Response:
"""Reload a config entry."""
if not request["hass_user"].is_admin:
raise Unauthorized(config_entry_id=entry_id, permission="remove")
hass = request.app["hass"]
hass: HomeAssistant = request.app["hass"]
entry = hass.config_entries.async_get_entry(entry_id)
if not entry:
return self.json_message("Invalid entry specified", HTTPStatus.NOT_FOUND)
@ -116,7 +117,12 @@ class ConfigManagerEntryResourceReloadView(HomeAssistantView):
return self.json({"require_restart": not entry.state.recoverable})
def _prepare_config_flow_result_json(result, prepare_result_json):
def _prepare_config_flow_result_json(
result: data_entry_flow.FlowResult,
prepare_result_json: Callable[
[data_entry_flow.FlowResult], data_entry_flow.FlowResult
],
) -> data_entry_flow.FlowResult:
"""Convert result to JSON."""
if result["type"] != data_entry_flow.FlowResultType.CREATE_ENTRY:
return prepare_result_json(result)
@ -134,14 +140,14 @@ class ConfigManagerFlowIndexView(FlowManagerIndexView):
url = "/api/config/config_entries/flow"
name = "api:config:config_entries:flow"
async def get(self, request):
async def get(self, request: web.Request) -> NoReturn:
"""Not implemented."""
raise aiohttp.web_exceptions.HTTPMethodNotAllowed("GET", ["POST"])
@require_admin(
error=Unauthorized(perm_category=CAT_CONFIG_ENTRIES, permission="add")
)
async def post(self, request):
async def post(self, request: web.Request) -> web.Response:
"""Handle a POST request."""
try:
return await super().post(request)
@ -151,7 +157,9 @@ class ConfigManagerFlowIndexView(FlowManagerIndexView):
status=HTTPStatus.BAD_REQUEST,
)
def _prepare_result_json(self, result):
def _prepare_result_json(
self, result: data_entry_flow.FlowResult
) -> data_entry_flow.FlowResult:
"""Convert result to JSON."""
return _prepare_config_flow_result_json(result, super()._prepare_result_json)
@ -165,18 +173,20 @@ class ConfigManagerFlowResourceView(FlowManagerResourceView):
@require_admin(
error=Unauthorized(perm_category=CAT_CONFIG_ENTRIES, permission="add")
)
async def get(self, request, /, flow_id):
async def get(self, request: web.Request, /, flow_id: str) -> web.Response:
"""Get the current state of a data_entry_flow."""
return await super().get(request, flow_id)
@require_admin(
error=Unauthorized(perm_category=CAT_CONFIG_ENTRIES, permission="add")
)
async def post(self, request, flow_id):
async def post(self, request: web.Request, flow_id: str) -> web.Response:
"""Handle a POST request."""
return await super().post(request, flow_id)
def _prepare_result_json(self, result):
def _prepare_result_json(
self, result: data_entry_flow.FlowResult
) -> data_entry_flow.FlowResult:
"""Convert result to JSON."""
return _prepare_config_flow_result_json(result, super()._prepare_result_json)
@ -187,10 +197,10 @@ class ConfigManagerAvailableFlowView(HomeAssistantView):
url = "/api/config/config_entries/flow_handlers"
name = "api:config:config_entries:flow_handlers"
async def get(self, request):
async def get(self, request: web.Request) -> web.Response:
"""List available flow handlers."""
hass = request.app["hass"]
kwargs = {}
hass: HomeAssistant = request.app["hass"]
kwargs: dict[str, Any] = {}
if "type" in request.query:
kwargs["type_filter"] = request.query["type"]
return self.json(await async_get_config_flows(hass, **kwargs))
@ -205,7 +215,7 @@ class OptionManagerFlowIndexView(FlowManagerIndexView):
@require_admin(
error=Unauthorized(perm_category=CAT_CONFIG_ENTRIES, permission=POLICY_EDIT)
)
async def post(self, request):
async def post(self, request: web.Request) -> web.Response:
"""Handle a POST request.
handler in request is entry_id.
@ -222,14 +232,14 @@ class OptionManagerFlowResourceView(FlowManagerResourceView):
@require_admin(
error=Unauthorized(perm_category=CAT_CONFIG_ENTRIES, permission=POLICY_EDIT)
)
async def get(self, request, /, flow_id):
async def get(self, request: web.Request, /, flow_id: str) -> web.Response:
"""Get the current state of a data_entry_flow."""
return await super().get(request, flow_id)
@require_admin(
error=Unauthorized(perm_category=CAT_CONFIG_ENTRIES, permission=POLICY_EDIT)
)
async def post(self, request, flow_id):
async def post(self, request: web.Request, flow_id: str) -> web.Response:
"""Handle a POST request."""
return await super().post(request, flow_id)
@ -535,7 +545,7 @@ async def async_matching_config_entries(
@callback
def entry_json(entry: config_entries.ConfigEntry) -> dict:
def entry_json(entry: config_entries.ConfigEntry) -> dict[str, Any]:
"""Return JSON value of a config entry."""
handler = config_entries.HANDLERS.get(entry.domain)
# work out if handler has support for options flow

6
homeassistant/components/config/core.py
View File

@ -1,7 +1,9 @@
"""Component to interact with Hassbian tools."""
from __future__ import annotations
from typing import Any
from aiohttp import web
import voluptuous as vol
from homeassistant.components import websocket_api
@ -13,7 +15,7 @@ from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.util import location, unit_system
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Set up the Hassbian config."""
hass.http.register_view(CheckConfigView)
websocket_api.async_register_command(hass, websocket_update_config)
@ -28,7 +30,7 @@ class CheckConfigView(HomeAssistantView):
name = "api:config:core:check_config"
@require_admin
async def post(self, request):
async def post(self, request: web.Request) -> web.Response:
"""Validate configuration and return results."""
res = await check_config.async_check_ha_config_file(request.app["hass"])

2
homeassistant/components/config/device_registry.py
View File

@ -17,7 +17,7 @@ from homeassistant.helpers.device_registry import (
)
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Enable the Device Registry views."""
websocket_api.async_register_command(hass, websocket_list_devices)

17
homeassistant/components/config/scene.py
View File

@ -1,19 +1,22 @@
"""Provide configuration end points for Scenes."""
from __future__ import annotations
from typing import Any
import uuid
from homeassistant.components.scene import DOMAIN, PLATFORM_SCHEMA
from homeassistant.config import SCENE_CONFIG_PATH
from homeassistant.const import CONF_ID, SERVICE_RELOAD
from homeassistant.core import DOMAIN as HA_DOMAIN
from homeassistant.core import DOMAIN as HA_DOMAIN, HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_registry as er
from . import ACTION_DELETE, EditIdBasedConfigView
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Set up the Scene config API."""
async def hook(action, config_key):
async def hook(action: str, config_key: str) -> None:
"""post_write_hook for Config View that reloads scenes."""
if action != ACTION_DELETE:
await hass.services.async_call(DOMAIN, SERVICE_RELOAD)
@ -44,7 +47,13 @@ async def async_setup(hass):
class EditSceneConfigView(EditIdBasedConfigView):
"""Edit scene config."""
def _write_value(self, hass, data, config_key, new_value):
def _write_value(
self,
hass: HomeAssistant,
data: list[dict[str, Any]],
config_key: str,
new_value: dict[str, Any],
) -> None:
"""Set value."""
updated_value = {CONF_ID: config_key}
# Iterate through some keys that we want to have ordered in the output

17
homeassistant/components/config/script.py
View File

@ -1,4 +1,8 @@
"""Provide configuration end points for scripts."""
from __future__ import annotations
from typing import Any
from homeassistant.components.script import DOMAIN
from homeassistant.components.script.config import (
SCRIPT_ENTITY_SCHEMA,
@ -6,15 +10,16 @@ from homeassistant.components.script.config import (
)
from homeassistant.config import SCRIPT_CONFIG_PATH
from homeassistant.const import SERVICE_RELOAD
from homeassistant.core import HomeAssistant
from homeassistant.helpers import config_validation as cv, entity_registry as er
from . import ACTION_DELETE, EditKeyBasedConfigView
async def async_setup(hass):
async def async_setup(hass: HomeAssistant) -> bool:
"""Set up the script config API."""
async def hook(action, config_key):
async def hook(action: str, config_key: str) -> None:
"""post_write_hook for Config View that reloads scripts."""
if action != ACTION_DELETE:
await hass.services.async_call(DOMAIN, SERVICE_RELOAD)
@ -46,6 +51,12 @@ async def async_setup(hass):
class EditScriptConfigView(EditKeyBasedConfigView):
"""Edit script config."""
def _write_value(self, hass, data, config_key, new_value):
def _write_value(
self,
hass: HomeAssistant,
data: dict[str, dict[str, Any]],
config_key: str,
new_value: dict[str, Any],
) -> None:
"""Set value."""
data[config_key] = new_value

2
homeassistant/util/yaml/dumper.py
View File

@ -17,7 +17,7 @@ except ImportError:
)
def dump(_dict: dict) -> str:
def dump(_dict: dict | list) -> str:
"""Dump YAML to a string and remove null."""
return yaml.dump(
_dict,

10
mypy.ini
View File

@ -980,6 +980,16 @@ disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.config.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.configurator.*]
check_untyped_defs = true
disallow_incomplete_defs = true