Type frontend strictly (#52148)

This commit is contained in:
Martin Hjelmare 2021-06-24 16:01:28 +02:00 committed by GitHub
parent afa00b7626
commit 09b3882a5b
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 95 additions and 55 deletions

View file

@ -1,13 +1,14 @@
"""Handle the frontend for Home Assistant."""
from __future__ import annotations
from collections.abc import Iterator
from functools import lru_cache
import json
import logging
import mimetypes
import os
import pathlib
from typing import Any
from typing import Any, TypedDict, cast
from aiohttp import hdrs, web, web_urldispatcher
import jinja2
@ -16,18 +17,18 @@ from yarl import URL
from homeassistant.components import websocket_api
from homeassistant.components.http.view import HomeAssistantView
from homeassistant.components.websocket_api.connection import ActiveConnection
from homeassistant.config import async_hass_config_yaml
from homeassistant.const import CONF_MODE, CONF_NAME, EVENT_THEMES_UPDATED
from homeassistant.core import callback
from homeassistant.core import HomeAssistant, ServiceCall, callback
from homeassistant.helpers import service
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.translation import async_get_translations
from homeassistant.helpers.typing import ConfigType
from homeassistant.loader import async_get_integration, bind_hass
from .storage import async_setup_frontend_storage
# mypy: allow-untyped-defs, no-check-untyped-defs
# Fix mimetypes for borked Windows machines
# https://github.com/home-assistant/frontend/issues/3336
mimetypes.add_type("text/css", ".css")
@ -191,15 +192,15 @@ class UrlManager:
on hass.data
"""
def __init__(self, urls):
def __init__(self, urls: list[str]) -> None:
"""Init the url manager."""
self.urls = frozenset(urls)
def add(self, url):
def add(self, url: str) -> None:
"""Add a url to the set."""
self.urls = frozenset([*self.urls, url])
def remove(self, url):
def remove(self, url: str) -> None:
"""Remove a url from the set."""
self.urls = self.urls - {url}
@ -208,7 +209,7 @@ class Panel:
"""Abstract class for panels."""
# Name of the webcomponent
component_name: str | None = None
component_name: str
# Icon to show in the sidebar
sidebar_icon: str | None = None
@ -227,13 +228,13 @@ class Panel:
def __init__(
self,
component_name,
sidebar_title,
sidebar_icon,
frontend_url_path,
config,
require_admin,
):
component_name: str,
sidebar_title: str | None,
sidebar_icon: str | None,
frontend_url_path: str | None,
config: dict[str, Any] | None,
require_admin: bool,
) -> None:
"""Initialize a built-in panel."""
self.component_name = component_name
self.sidebar_title = sidebar_title
@ -243,7 +244,7 @@ class Panel:
self.require_admin = require_admin
@callback
def to_response(self):
def to_response(self) -> PanelRespons:
"""Panel as dictionary."""
return {
"component_name": self.component_name,
@ -258,16 +259,16 @@ class Panel:
@bind_hass
@callback
def async_register_built_in_panel(
hass,
component_name,
sidebar_title=None,
sidebar_icon=None,
frontend_url_path=None,
config=None,
require_admin=False,
hass: HomeAssistant,
component_name: str,
sidebar_title: str | None = None,
sidebar_icon: str | None = None,
frontend_url_path: str | None = None,
config: dict[str, Any] | None = None,
require_admin: bool = False,
*,
update=False,
):
update: bool = False,
) -> None:
"""Register a built-in panel."""
panel = Panel(
component_name,
@ -290,7 +291,7 @@ def async_register_built_in_panel(
@bind_hass
@callback
def async_remove_panel(hass, frontend_url_path):
def async_remove_panel(hass: HomeAssistant, frontend_url_path: str) -> None:
"""Remove a built-in panel."""
panel = hass.data.get(DATA_PANELS, {}).pop(frontend_url_path, None)
@ -300,18 +301,18 @@ def async_remove_panel(hass, frontend_url_path):
hass.bus.async_fire(EVENT_PANELS_UPDATED)
def add_extra_js_url(hass, url, es5=False):
def add_extra_js_url(hass: HomeAssistant, url: str, es5: bool = False) -> None:
"""Register extra js or module url to load."""
key = DATA_EXTRA_JS_URL_ES5 if es5 else DATA_EXTRA_MODULE_URL
hass.data[key].add(url)
def add_manifest_json_key(key, val):
def add_manifest_json_key(key: str, val: Any) -> None:
"""Add a keyval to the manifest.json."""
MANIFEST_JSON.update_key(key, val)
def _frontend_root(dev_repo_path):
def _frontend_root(dev_repo_path: str | None) -> pathlib.Path:
"""Return root path to the frontend files."""
if dev_repo_path is not None:
return pathlib.Path(dev_repo_path) / "hass_frontend"
@ -319,17 +320,17 @@ def _frontend_root(dev_repo_path):
# pylint: disable=import-outside-toplevel
import hass_frontend
return hass_frontend.where()
return cast(pathlib.Path, hass_frontend.where())
async def async_setup(hass, config):
async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
"""Set up the serving of the frontend."""
await async_setup_frontend_storage(hass)
hass.components.websocket_api.async_register_command(websocket_get_panels)
hass.components.websocket_api.async_register_command(websocket_get_themes)
hass.components.websocket_api.async_register_command(websocket_get_translations)
hass.components.websocket_api.async_register_command(websocket_get_version)
hass.http.register_view(ManifestJSONView)
hass.http.register_view(ManifestJSONView())
conf = config.get(DOMAIN, {})
@ -396,7 +397,9 @@ async def async_setup(hass, config):
return True
async def _async_setup_themes(hass, themes):
async def _async_setup_themes(
hass: HomeAssistant, themes: dict[str, Any] | None
) -> None:
"""Set up themes data and services."""
hass.data[DATA_THEMES] = themes or {}
@ -417,7 +420,7 @@ async def _async_setup_themes(hass, themes):
hass.data[DATA_DEFAULT_DARK_THEME] = dark_theme_name
@callback
def update_theme_and_fire_event():
def update_theme_and_fire_event() -> None:
"""Update theme_color in manifest."""
name = hass.data[DATA_DEFAULT_THEME]
themes = hass.data[DATA_THEMES]
@ -434,7 +437,7 @@ async def _async_setup_themes(hass, themes):
hass.bus.async_fire(EVENT_THEMES_UPDATED)
@callback
def set_theme(call):
def set_theme(call: ServiceCall) -> None:
"""Set backend-preferred theme."""
name = call.data[CONF_NAME]
mode = call.data.get("mode", "light")
@ -466,7 +469,7 @@ async def _async_setup_themes(hass, themes):
)
update_theme_and_fire_event()
async def reload_themes(_):
async def reload_themes(_: ServiceCall) -> None:
"""Reload themes."""
config = await async_hass_config_yaml(hass)
new_themes = config[DOMAIN].get(CONF_THEMES, {})
@ -500,19 +503,19 @@ async def _async_setup_themes(hass, themes):
@callback
@lru_cache(maxsize=1)
def _async_render_index_cached(template, **kwargs):
def _async_render_index_cached(template: jinja2.Template, **kwargs: Any) -> str:
return template.render(**kwargs)
class IndexView(web_urldispatcher.AbstractResource):
"""Serve the frontend."""
def __init__(self, repo_path, hass):
def __init__(self, repo_path: str | None, hass: HomeAssistant) -> None:
"""Initialize the frontend view."""
super().__init__(name="frontend:index")
self.repo_path = repo_path
self.hass = hass
self._template_cache = None
self._template_cache: jinja2.Template | None = None
@property
def canonical(self) -> str:
@ -520,7 +523,7 @@ class IndexView(web_urldispatcher.AbstractResource):
return "/"
@property
def _route(self):
def _route(self) -> web_urldispatcher.ResourceRoute:
"""Return the index route."""
return web_urldispatcher.ResourceRoute("GET", self.get, self)
@ -552,7 +555,7 @@ class IndexView(web_urldispatcher.AbstractResource):
Required for subapplications support.
"""
def get_info(self):
def get_info(self) -> dict[str, list[str]]: # type: ignore[override]
"""Return a dict with additional info useful for introspection."""
return {"panels": list(self.hass.data[DATA_PANELS])}
@ -562,7 +565,7 @@ class IndexView(web_urldispatcher.AbstractResource):
def raw_match(self, path: str) -> bool:
"""Perform a raw match against path."""
def get_template(self):
def get_template(self) -> jinja2.Template:
"""Get template."""
tpl = self._template_cache
if tpl is None:
@ -600,7 +603,7 @@ class IndexView(web_urldispatcher.AbstractResource):
"""Return length of resource."""
return 1
def __iter__(self):
def __iter__(self) -> Iterator[web_urldispatcher.ResourceRoute]:
"""Iterate over routes."""
return iter([self._route])
@ -613,7 +616,7 @@ class ManifestJSONView(HomeAssistantView):
name = "manifestjson"
@callback
def get(self, request): # pylint: disable=no-self-use
def get(self, request: web.Request) -> web.Response: # pylint: disable=no-self-use
"""Return the manifest.json."""
return web.Response(
text=MANIFEST_JSON.json, content_type="application/manifest+json"
@ -622,7 +625,9 @@ class ManifestJSONView(HomeAssistantView):
@callback
@websocket_api.websocket_command({"type": "get_panels"})
def websocket_get_panels(hass, connection, msg):
def websocket_get_panels(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Handle get panels command."""
user_is_admin = connection.user.is_admin
panels = {
@ -636,7 +641,9 @@ def websocket_get_panels(hass, connection, msg):
@callback
@websocket_api.websocket_command({"type": "frontend/get_themes"})
def websocket_get_themes(hass, connection, msg):
def websocket_get_themes(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Handle get themes command."""
if hass.config.safe_mode:
connection.send_message(
@ -677,7 +684,9 @@ def websocket_get_themes(hass, connection, msg):
}
)
@websocket_api.async_response
async def websocket_get_translations(hass, connection, msg):
async def websocket_get_translations(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Handle get translations command."""
resources = await async_get_translations(
hass,
@ -693,7 +702,9 @@ async def websocket_get_translations(hass, connection, msg):
@websocket_api.websocket_command({"type": "frontend/get_version"})
@websocket_api.async_response
async def websocket_get_version(hass, connection, msg):
async def websocket_get_version(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Handle get version command."""
integration = await async_get_integration(hass, "frontend")
@ -707,3 +718,14 @@ async def websocket_get_version(hass, connection, msg):
connection.send_error(msg["id"], "unknown_version", "Version not found")
else:
connection.send_result(msg["id"], {"version": frontend})
class PanelRespons(TypedDict):
"""Represent the panel response type."""
component_name: str
icon: str | None
title: str | None
config: dict[str, Any] | None
url_path: str | None
require_admin: bool

View file

@ -1,28 +1,34 @@
"""API for persistent storage for the frontend."""
from __future__ import annotations
from functools import wraps
from typing import Any, Callable
import voluptuous as vol
from homeassistant.components import websocket_api
# mypy: allow-untyped-calls, allow-untyped-defs
from homeassistant.components.websocket_api.connection import ActiveConnection
from homeassistant.core import HomeAssistant
from homeassistant.helpers.storage import Store
DATA_STORAGE = "frontend_storage"
STORAGE_VERSION_USER_DATA = 1
async def async_setup_frontend_storage(hass):
async def async_setup_frontend_storage(hass: HomeAssistant) -> None:
"""Set up frontend storage."""
hass.data[DATA_STORAGE] = ({}, {})
hass.components.websocket_api.async_register_command(websocket_set_user_data)
hass.components.websocket_api.async_register_command(websocket_get_user_data)
def with_store(orig_func):
def with_store(orig_func: Callable) -> Callable:
"""Decorate function to provide data."""
@wraps(orig_func)
async def with_store_func(hass, connection, msg):
async def with_store_func(
hass: HomeAssistant, connection: ActiveConnection, msg: dict
) -> None:
"""Provide user specific data and store to function."""
stores, data = hass.data[DATA_STORAGE]
user_id = connection.user.id
@ -50,7 +56,13 @@ def with_store(orig_func):
)
@websocket_api.async_response
@with_store
async def websocket_set_user_data(hass, connection, msg, store, data):
async def websocket_set_user_data(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
store: Store,
data: dict[str, Any],
) -> None:
"""Handle set global data command.
Async friendly.
@ -65,7 +77,13 @@ async def websocket_set_user_data(hass, connection, msg, store, data):
)
@websocket_api.async_response
@with_store
async def websocket_get_user_data(hass, connection, msg, store, data):
async def websocket_get_user_data(
hass: HomeAssistant,
connection: ActiveConnection,
msg: dict,
store: Store,
data: dict[str, Any],
) -> None:
"""Handle get global data command.
Async friendly.