Improve hassio typing (#107292)

This commit is contained in:
Marc Mueller 2024-01-08 10:08:09 +01:00 committed by GitHub
parent 78752264b3
commit a6fc4c2bd5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
9 changed files with 102 additions and 82 deletions

View File

@ -25,6 +25,7 @@ from homeassistant.const import (
) )
from homeassistant.core import ( from homeassistant.core import (
CALLBACK_TYPE, CALLBACK_TYPE,
Event,
HassJob, HassJob,
HomeAssistant, HomeAssistant,
ServiceCall, ServiceCall,
@ -332,7 +333,7 @@ def get_addons_info(hass: HomeAssistant) -> dict[str, dict[str, Any]] | None:
@callback @callback
@bind_hass @bind_hass
def get_addons_stats(hass): def get_addons_stats(hass: HomeAssistant) -> dict[str, Any]:
"""Return Addons stats. """Return Addons stats.
Async friendly. Async friendly.
@ -342,7 +343,7 @@ def get_addons_stats(hass):
@callback @callback
@bind_hass @bind_hass
def get_core_stats(hass): def get_core_stats(hass: HomeAssistant) -> dict[str, Any]:
"""Return core stats. """Return core stats.
Async friendly. Async friendly.
@ -352,7 +353,7 @@ def get_core_stats(hass):
@callback @callback
@bind_hass @bind_hass
def get_supervisor_stats(hass): def get_supervisor_stats(hass: HomeAssistant) -> dict[str, Any]:
"""Return supervisor stats. """Return supervisor stats.
Async friendly. Async friendly.
@ -362,7 +363,7 @@ def get_supervisor_stats(hass):
@callback @callback
@bind_hass @bind_hass
def get_addons_changelogs(hass): def get_addons_changelogs(hass: HomeAssistant):
"""Return Addons changelogs. """Return Addons changelogs.
Async friendly. Async friendly.
@ -488,7 +489,7 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: # noqa:
last_timezone = None last_timezone = None
async def push_config(_): async def push_config(_: Event | None) -> None:
"""Push core config to Hass.io.""" """Push core config to Hass.io."""
nonlocal last_timezone nonlocal last_timezone
@ -986,7 +987,7 @@ class HassioDataUpdateCoordinator(DataUpdateCoordinator):
enabled_updates[key].add(entity_id) enabled_updates[key].add(entity_id)
@callback @callback
def _remove(): def _remove() -> None:
for key in types: for key in types:
enabled_updates[key].remove(entity_id) enabled_updates[key].remove(entity_id)

View File

@ -2,6 +2,7 @@
import asyncio import asyncio
from http import HTTPStatus from http import HTTPStatus
import logging import logging
from typing import Any
from aiohttp import web from aiohttp import web
@ -11,12 +12,12 @@ from homeassistant.const import ATTR_ICON
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from .const import ATTR_ADMIN, ATTR_ENABLE, ATTR_PANELS, ATTR_TITLE from .const import ATTR_ADMIN, ATTR_ENABLE, ATTR_PANELS, ATTR_TITLE
from .handler import HassioAPIError from .handler import HassIO, HassioAPIError
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
async def async_setup_addon_panel(hass: HomeAssistant, hassio): async def async_setup_addon_panel(hass: HomeAssistant, hassio: HassIO) -> None:
"""Add-on Ingress Panel setup.""" """Add-on Ingress Panel setup."""
hassio_addon_panel = HassIOAddonPanel(hass, hassio) hassio_addon_panel = HassIOAddonPanel(hass, hassio)
hass.http.register_view(hassio_addon_panel) hass.http.register_view(hassio_addon_panel)
@ -26,7 +27,7 @@ async def async_setup_addon_panel(hass: HomeAssistant, hassio):
return return
# Register available panels # Register available panels
jobs = [] jobs: list[asyncio.Task[None]] = []
for addon, data in panels.items(): for addon, data in panels.items():
if not data[ATTR_ENABLE]: if not data[ATTR_ENABLE]:
continue continue
@ -46,12 +47,12 @@ class HassIOAddonPanel(HomeAssistantView):
name = "api:hassio_push:panel" name = "api:hassio_push:panel"
url = "/api/hassio_push/panel/{addon}" url = "/api/hassio_push/panel/{addon}"
def __init__(self, hass, hassio): def __init__(self, hass: HomeAssistant, hassio: HassIO) -> None:
"""Initialize WebView.""" """Initialize WebView."""
self.hass = hass self.hass = hass
self.hassio = hassio self.hassio = hassio
async def post(self, request, addon): async def post(self, request: web.Request, addon: str) -> web.Response:
"""Handle new add-on panel requests.""" """Handle new add-on panel requests."""
panels = await self.get_panels() panels = await self.get_panels()
@ -65,12 +66,12 @@ class HassIOAddonPanel(HomeAssistantView):
await _register_panel(self.hass, addon, data) await _register_panel(self.hass, addon, data)
return web.Response() return web.Response()
async def delete(self, request, addon): async def delete(self, request: web.Request, addon: str) -> web.Response:
"""Handle remove add-on panel requests.""" """Handle remove add-on panel requests."""
frontend.async_remove_panel(self.hass, addon) frontend.async_remove_panel(self.hass, addon)
return web.Response() return web.Response()
async def get_panels(self): async def get_panels(self) -> dict:
"""Return panels add-on info data.""" """Return panels add-on info data."""
try: try:
data = await self.hassio.get_ingress_panels() data = await self.hassio.get_ingress_panels()
@ -80,7 +81,9 @@ class HassIOAddonPanel(HomeAssistantView):
return {} return {}
async def _register_panel(hass, addon, data): async def _register_panel(
hass: HomeAssistant, addon: str, data: dict[str, Any]
) -> None:
"""Init coroutine to register the panel.""" """Init coroutine to register the panel."""
await panel_custom.async_register_panel( await panel_custom.async_register_panel(
hass, hass,

View File

@ -21,7 +21,7 @@ _LOGGER = logging.getLogger(__name__)
@callback @callback
def async_setup_auth_view(hass: HomeAssistant, user: User): def async_setup_auth_view(hass: HomeAssistant, user: User) -> None:
"""Auth setup.""" """Auth setup."""
hassio_auth = HassIOAuth(hass, user) hassio_auth = HassIOAuth(hass, user)
hassio_password_reset = HassIOPasswordReset(hass, user) hassio_password_reset = HassIOPasswordReset(hass, user)
@ -38,7 +38,7 @@ class HassIOBaseAuth(HomeAssistantView):
self.hass = hass self.hass = hass
self.user = user self.user = user
def _check_access(self, request: web.Request): def _check_access(self, request: web.Request) -> None:
"""Check if this call is from Supervisor.""" """Check if this call is from Supervisor."""
# Check caller IP # Check caller IP
hassio_ip = os.environ["SUPERVISOR"].split(":")[0] hassio_ip = os.environ["SUPERVISOR"].split(":")[0]
@ -71,7 +71,7 @@ class HassIOAuth(HassIOBaseAuth):
extra=vol.ALLOW_EXTRA, extra=vol.ALLOW_EXTRA,
) )
) )
async def post(self, request, data): async def post(self, request: web.Request, data: dict[str, str]) -> web.Response:
"""Handle auth requests.""" """Handle auth requests."""
self._check_access(request) self._check_access(request)
provider = auth_ha.async_get_provider(request.app["hass"]) provider = auth_ha.async_get_provider(request.app["hass"])
@ -101,7 +101,7 @@ class HassIOPasswordReset(HassIOBaseAuth):
extra=vol.ALLOW_EXTRA, extra=vol.ALLOW_EXTRA,
) )
) )
async def post(self, request, data): async def post(self, request: web.Request, data: dict[str, str]) -> web.Response:
"""Handle password reset requests.""" """Handle password reset requests."""
self._check_access(request) self._check_access(request)
provider = auth_ha.async_get_provider(request.app["hass"]) provider = auth_ha.async_get_provider(request.app["hass"])

View File

@ -1,7 +1,11 @@
"""Config flow for Home Assistant Supervisor integration.""" """Config flow for Home Assistant Supervisor integration."""
from __future__ import annotations
import logging import logging
from typing import Any
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.data_entry_flow import FlowResult
from . import DOMAIN from . import DOMAIN
@ -13,7 +17,9 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
VERSION = 1 VERSION = 1
async def async_step_system(self, user_input=None): async def async_step_system(
self, user_input: dict[str, Any] | None = None
) -> FlowResult:
"""Handle the initial step.""" """Handle the initial step."""
# We only need one Hass.io config entry # We only need one Hass.io config entry
await self.async_set_unique_id(DOMAIN) await self.async_set_unique_id(DOMAIN)

View File

@ -12,7 +12,7 @@ from aiohttp.web_exceptions import HTTPServiceUnavailable
from homeassistant import config_entries from homeassistant import config_entries
from homeassistant.components.http import HomeAssistantView from homeassistant.components.http import HomeAssistantView
from homeassistant.const import ATTR_NAME, ATTR_SERVICE, EVENT_HOMEASSISTANT_START from homeassistant.const import ATTR_NAME, ATTR_SERVICE, EVENT_HOMEASSISTANT_START
from homeassistant.core import HomeAssistant, callback from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.data_entry_flow import BaseServiceInfo from homeassistant.data_entry_flow import BaseServiceInfo
from homeassistant.helpers import discovery_flow from homeassistant.helpers import discovery_flow
@ -33,13 +33,13 @@ class HassioServiceInfo(BaseServiceInfo):
@callback @callback
def async_setup_discovery_view(hass: HomeAssistant, hassio): def async_setup_discovery_view(hass: HomeAssistant, hassio: HassIO) -> None:
"""Discovery setup.""" """Discovery setup."""
hassio_discovery = HassIODiscovery(hass, hassio) hassio_discovery = HassIODiscovery(hass, hassio)
hass.http.register_view(hassio_discovery) hass.http.register_view(hassio_discovery)
# Handle exists discovery messages # Handle exists discovery messages
async def _async_discovery_start_handler(event): async def _async_discovery_start_handler(event: Event) -> None:
"""Process all exists discovery on startup.""" """Process all exists discovery on startup."""
try: try:
data = await hassio.retrieve_discovery_messages() data = await hassio.retrieve_discovery_messages()
@ -70,7 +70,7 @@ class HassIODiscovery(HomeAssistantView):
self.hass = hass self.hass = hass
self.hassio = hassio self.hassio = hassio
async def post(self, request, uuid): async def post(self, request: web.Request, uuid: str) -> web.Response:
"""Handle new discovery requests.""" """Handle new discovery requests."""
# Fetch discovery data and prevent injections # Fetch discovery data and prevent injections
try: try:
@ -82,9 +82,9 @@ class HassIODiscovery(HomeAssistantView):
await self.async_process_new(data) await self.async_process_new(data)
return web.Response() return web.Response()
async def delete(self, request, uuid): async def delete(self, request: web.Request, uuid: str) -> web.Response:
"""Handle remove discovery requests.""" """Handle remove discovery requests."""
data = await request.json() data: dict[str, Any] = await request.json()
await self.async_process_del(data) await self.async_process_del(data)
return web.Response() return web.Response()
@ -114,7 +114,7 @@ class HassIODiscovery(HomeAssistantView):
data=HassioServiceInfo(config=config_data, name=name, slug=slug, uuid=uuid), data=HassioServiceInfo(config=config_data, name=name, slug=slug, uuid=uuid),
) )
async def async_process_del(self, data): async def async_process_del(self, data: dict[str, Any]) -> None:
"""Process remove discovery entry.""" """Process remove discovery entry."""
service = data[ATTR_SERVICE] service = data[ATTR_SERVICE]
uuid = data[ATTR_UUID] uuid = data[ATTR_UUID]

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
from collections.abc import Coroutine
from http import HTTPStatus from http import HTTPStatus
import logging import logging
import os import os
@ -10,6 +11,7 @@ from typing import Any
import aiohttp import aiohttp
from yarl import URL from yarl import URL
from homeassistant.auth.models import RefreshToken
from homeassistant.components.http import ( from homeassistant.components.http import (
CONF_SERVER_HOST, CONF_SERVER_HOST,
CONF_SERVER_PORT, CONF_SERVER_PORT,
@ -62,7 +64,7 @@ async def async_get_addon_info(hass: HomeAssistant, slug: str) -> dict:
The add-on must be installed. The add-on must be installed.
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
return await hassio.get_addon_info(slug) return await hassio.get_addon_info(slug)
@ -83,7 +85,7 @@ async def async_update_diagnostics(hass: HomeAssistant, diagnostics: bool) -> di
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
return await hassio.update_diagnostics(diagnostics) return await hassio.update_diagnostics(diagnostics)
@ -94,7 +96,7 @@ async def async_install_addon(hass: HomeAssistant, slug: str) -> dict:
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = f"/addons/{slug}/install" command = f"/addons/{slug}/install"
return await hassio.send_command(command, timeout=None) return await hassio.send_command(command, timeout=None)
@ -106,7 +108,7 @@ async def async_uninstall_addon(hass: HomeAssistant, slug: str) -> dict:
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = f"/addons/{slug}/uninstall" command = f"/addons/{slug}/uninstall"
return await hassio.send_command(command, timeout=60) return await hassio.send_command(command, timeout=60)
@ -122,7 +124,7 @@ async def async_update_addon(
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = f"/addons/{slug}/update" command = f"/addons/{slug}/update"
return await hassio.send_command( return await hassio.send_command(
command, command,
@ -138,7 +140,7 @@ async def async_start_addon(hass: HomeAssistant, slug: str) -> dict:
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = f"/addons/{slug}/start" command = f"/addons/{slug}/start"
return await hassio.send_command(command, timeout=60) return await hassio.send_command(command, timeout=60)
@ -150,7 +152,7 @@ async def async_restart_addon(hass: HomeAssistant, slug: str) -> dict:
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = f"/addons/{slug}/restart" command = f"/addons/{slug}/restart"
return await hassio.send_command(command, timeout=None) return await hassio.send_command(command, timeout=None)
@ -162,7 +164,7 @@ async def async_stop_addon(hass: HomeAssistant, slug: str) -> dict:
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = f"/addons/{slug}/stop" command = f"/addons/{slug}/stop"
return await hassio.send_command(command, timeout=60) return await hassio.send_command(command, timeout=60)
@ -176,7 +178,7 @@ async def async_set_addon_options(
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = f"/addons/{slug}/options" command = f"/addons/{slug}/options"
return await hassio.send_command(command, payload=options) return await hassio.send_command(command, payload=options)
@ -184,7 +186,7 @@ async def async_set_addon_options(
@bind_hass @bind_hass
async def async_get_addon_discovery_info(hass: HomeAssistant, slug: str) -> dict | None: async def async_get_addon_discovery_info(hass: HomeAssistant, slug: str) -> dict | None:
"""Return discovery data for an add-on.""" """Return discovery data for an add-on."""
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
data = await hassio.retrieve_discovery_messages() data = await hassio.retrieve_discovery_messages()
discovered_addons = data[ATTR_DISCOVERY] discovered_addons = data[ATTR_DISCOVERY]
return next((addon for addon in discovered_addons if addon["addon"] == slug), None) return next((addon for addon in discovered_addons if addon["addon"] == slug), None)
@ -199,7 +201,7 @@ async def async_create_backup(
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
backup_type = "partial" if partial else "full" backup_type = "partial" if partial else "full"
command = f"/backups/new/{backup_type}" command = f"/backups/new/{backup_type}"
return await hassio.send_command(command, payload=payload, timeout=None) return await hassio.send_command(command, payload=payload, timeout=None)
@ -212,7 +214,7 @@ async def async_update_os(hass: HomeAssistant, version: str | None = None) -> di
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = "/os/update" command = "/os/update"
return await hassio.send_command( return await hassio.send_command(
command, command,
@ -228,7 +230,7 @@ async def async_update_supervisor(hass: HomeAssistant) -> dict:
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = "/supervisor/update" command = "/supervisor/update"
return await hassio.send_command(command, timeout=None) return await hassio.send_command(command, timeout=None)
@ -242,7 +244,7 @@ async def async_update_core(
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = "/core/update" command = "/core/update"
return await hassio.send_command( return await hassio.send_command(
command, command,
@ -258,7 +260,7 @@ async def async_apply_suggestion(hass: HomeAssistant, suggestion_uuid: str) -> b
The caller of the function should handle HassioAPIError. The caller of the function should handle HassioAPIError.
""" """
hassio = hass.data[DOMAIN] hassio: HassIO = hass.data[DOMAIN]
command = f"/resolution/suggestion/{suggestion_uuid}" command = f"/resolution/suggestion/{suggestion_uuid}"
return await hassio.send_command(command, timeout=None) return await hassio.send_command(command, timeout=None)
@ -330,7 +332,7 @@ class HassIO:
self._ip = ip self._ip = ip
@_api_bool @_api_bool
def is_connected(self): def is_connected(self) -> Coroutine:
"""Return true if it connected to Hass.io supervisor. """Return true if it connected to Hass.io supervisor.
This method returns a coroutine. This method returns a coroutine.
@ -338,7 +340,7 @@ class HassIO:
return self.send_command("/supervisor/ping", method="get", timeout=15) return self.send_command("/supervisor/ping", method="get", timeout=15)
@api_data @api_data
def get_info(self): def get_info(self) -> Coroutine:
"""Return generic Supervisor information. """Return generic Supervisor information.
This method returns a coroutine. This method returns a coroutine.
@ -346,7 +348,7 @@ class HassIO:
return self.send_command("/info", method="get") return self.send_command("/info", method="get")
@api_data @api_data
def get_host_info(self): def get_host_info(self) -> Coroutine:
"""Return data for Host. """Return data for Host.
This method returns a coroutine. This method returns a coroutine.
@ -354,7 +356,7 @@ class HassIO:
return self.send_command("/host/info", method="get") return self.send_command("/host/info", method="get")
@api_data @api_data
def get_os_info(self): def get_os_info(self) -> Coroutine:
"""Return data for the OS. """Return data for the OS.
This method returns a coroutine. This method returns a coroutine.
@ -362,7 +364,7 @@ class HassIO:
return self.send_command("/os/info", method="get") return self.send_command("/os/info", method="get")
@api_data @api_data
def get_core_info(self): def get_core_info(self) -> Coroutine:
"""Return data for Home Asssistant Core. """Return data for Home Asssistant Core.
This method returns a coroutine. This method returns a coroutine.
@ -370,7 +372,7 @@ class HassIO:
return self.send_command("/core/info", method="get") return self.send_command("/core/info", method="get")
@api_data @api_data
def get_supervisor_info(self): def get_supervisor_info(self) -> Coroutine:
"""Return data for the Supervisor. """Return data for the Supervisor.
This method returns a coroutine. This method returns a coroutine.
@ -378,7 +380,7 @@ class HassIO:
return self.send_command("/supervisor/info", method="get") return self.send_command("/supervisor/info", method="get")
@api_data @api_data
def get_addon_info(self, addon): def get_addon_info(self, addon: str) -> Coroutine:
"""Return data for a Add-on. """Return data for a Add-on.
This method returns a coroutine. This method returns a coroutine.
@ -386,7 +388,7 @@ class HassIO:
return self.send_command(f"/addons/{addon}/info", method="get") return self.send_command(f"/addons/{addon}/info", method="get")
@api_data @api_data
def get_core_stats(self): def get_core_stats(self) -> Coroutine:
"""Return stats for the core. """Return stats for the core.
This method returns a coroutine. This method returns a coroutine.
@ -394,7 +396,7 @@ class HassIO:
return self.send_command("/core/stats", method="get") return self.send_command("/core/stats", method="get")
@api_data @api_data
def get_addon_stats(self, addon): def get_addon_stats(self, addon: str) -> Coroutine:
"""Return stats for an Add-on. """Return stats for an Add-on.
This method returns a coroutine. This method returns a coroutine.
@ -402,14 +404,14 @@ class HassIO:
return self.send_command(f"/addons/{addon}/stats", method="get") return self.send_command(f"/addons/{addon}/stats", method="get")
@api_data @api_data
def get_supervisor_stats(self): def get_supervisor_stats(self) -> Coroutine:
"""Return stats for the supervisor. """Return stats for the supervisor.
This method returns a coroutine. This method returns a coroutine.
""" """
return self.send_command("/supervisor/stats", method="get") return self.send_command("/supervisor/stats", method="get")
def get_addon_changelog(self, addon): def get_addon_changelog(self, addon: str) -> Coroutine:
"""Return changelog for an Add-on. """Return changelog for an Add-on.
This method returns a coroutine. This method returns a coroutine.
@ -419,7 +421,7 @@ class HassIO:
) )
@api_data @api_data
def get_store(self): def get_store(self) -> Coroutine:
"""Return data from the store. """Return data from the store.
This method returns a coroutine. This method returns a coroutine.
@ -427,7 +429,7 @@ class HassIO:
return self.send_command("/store", method="get") return self.send_command("/store", method="get")
@api_data @api_data
def get_ingress_panels(self): def get_ingress_panels(self) -> Coroutine:
"""Return data for Add-on ingress panels. """Return data for Add-on ingress panels.
This method returns a coroutine. This method returns a coroutine.
@ -435,7 +437,7 @@ class HassIO:
return self.send_command("/ingress/panels", method="get") return self.send_command("/ingress/panels", method="get")
@_api_bool @_api_bool
def restart_homeassistant(self): def restart_homeassistant(self) -> Coroutine:
"""Restart Home-Assistant container. """Restart Home-Assistant container.
This method returns a coroutine. This method returns a coroutine.
@ -443,7 +445,7 @@ class HassIO:
return self.send_command("/homeassistant/restart") return self.send_command("/homeassistant/restart")
@_api_bool @_api_bool
def stop_homeassistant(self): def stop_homeassistant(self) -> Coroutine:
"""Stop Home-Assistant container. """Stop Home-Assistant container.
This method returns a coroutine. This method returns a coroutine.
@ -451,7 +453,7 @@ class HassIO:
return self.send_command("/homeassistant/stop") return self.send_command("/homeassistant/stop")
@_api_bool @_api_bool
def refresh_updates(self): def refresh_updates(self) -> Coroutine:
"""Refresh available updates. """Refresh available updates.
This method returns a coroutine. This method returns a coroutine.
@ -459,7 +461,7 @@ class HassIO:
return self.send_command("/refresh_updates", timeout=None) return self.send_command("/refresh_updates", timeout=None)
@api_data @api_data
def retrieve_discovery_messages(self): def retrieve_discovery_messages(self) -> Coroutine:
"""Return all discovery data from Hass.io API. """Return all discovery data from Hass.io API.
This method returns a coroutine. This method returns a coroutine.
@ -467,7 +469,7 @@ class HassIO:
return self.send_command("/discovery", method="get", timeout=60) return self.send_command("/discovery", method="get", timeout=60)
@api_data @api_data
def get_discovery_message(self, uuid): def get_discovery_message(self, uuid: str) -> Coroutine:
"""Return a single discovery data message. """Return a single discovery data message.
This method returns a coroutine. This method returns a coroutine.
@ -475,7 +477,7 @@ class HassIO:
return self.send_command(f"/discovery/{uuid}", method="get") return self.send_command(f"/discovery/{uuid}", method="get")
@api_data @api_data
def get_resolution_info(self): def get_resolution_info(self) -> Coroutine:
"""Return data for Supervisor resolution center. """Return data for Supervisor resolution center.
This method returns a coroutine. This method returns a coroutine.
@ -483,7 +485,9 @@ class HassIO:
return self.send_command("/resolution/info", method="get") return self.send_command("/resolution/info", method="get")
@api_data @api_data
def get_suggestions_for_issue(self, issue_id: str) -> dict[str, Any]: def get_suggestions_for_issue(
self, issue_id: str
) -> Coroutine[Any, Any, dict[str, Any]]:
"""Return suggestions for issue from Supervisor resolution center. """Return suggestions for issue from Supervisor resolution center.
This method returns a coroutine. This method returns a coroutine.
@ -493,7 +497,9 @@ class HassIO:
) )
@_api_bool @_api_bool
async def update_hass_api(self, http_config, refresh_token): async def update_hass_api(
self, http_config: dict[str, Any], refresh_token: RefreshToken
):
"""Update Home Assistant API data on Hass.io.""" """Update Home Assistant API data on Hass.io."""
port = http_config.get(CONF_SERVER_PORT) or SERVER_PORT port = http_config.get(CONF_SERVER_PORT) or SERVER_PORT
options = { options = {
@ -513,7 +519,7 @@ class HassIO:
return await self.send_command("/homeassistant/options", payload=options) return await self.send_command("/homeassistant/options", payload=options)
@_api_bool @_api_bool
def update_hass_timezone(self, timezone): def update_hass_timezone(self, timezone: str) -> Coroutine:
"""Update Home-Assistant timezone data on Hass.io. """Update Home-Assistant timezone data on Hass.io.
This method returns a coroutine. This method returns a coroutine.
@ -521,7 +527,7 @@ class HassIO:
return self.send_command("/supervisor/options", payload={"timezone": timezone}) return self.send_command("/supervisor/options", payload={"timezone": timezone})
@_api_bool @_api_bool
def update_diagnostics(self, diagnostics: bool): def update_diagnostics(self, diagnostics: bool) -> Coroutine:
"""Update Supervisor diagnostics setting. """Update Supervisor diagnostics setting.
This method returns a coroutine. This method returns a coroutine.
@ -531,7 +537,7 @@ class HassIO:
) )
@_api_bool @_api_bool
def apply_suggestion(self, suggestion_uuid: str): def apply_suggestion(self, suggestion_uuid: str) -> Coroutine:
"""Apply a suggestion from supervisor's resolution center. """Apply a suggestion from supervisor's resolution center.
This method returns a coroutine. This method returns a coroutine.
@ -540,14 +546,14 @@ class HassIO:
async def send_command( async def send_command(
self, self,
command, command: str,
method="post", method: str = "post",
payload=None, payload: Any | None = None,
timeout=10, timeout: int | None = 10,
return_text=False, return_text: bool = False,
*, *,
source="core.handler", source: str = "core.handler",
): ) -> Any:
"""Send API command to Hass.io. """Send API command to Hass.io.
This method is a coroutine. This method is a coroutine.

View File

@ -9,7 +9,7 @@ import logging
from urllib.parse import quote from urllib.parse import quote
import aiohttp import aiohttp
from aiohttp import ClientTimeout, hdrs, web from aiohttp import ClientTimeout, ClientWebSocketResponse, hdrs, web
from aiohttp.web_exceptions import HTTPBadGateway, HTTPBadRequest from aiohttp.web_exceptions import HTTPBadGateway, HTTPBadRequest
from multidict import CIMultiDict from multidict import CIMultiDict
from yarl import URL from yarl import URL
@ -46,7 +46,7 @@ MAX_SIMPLE_RESPONSE_SIZE = 4194000
@callback @callback
def async_setup_ingress_view(hass: HomeAssistant, host: str): def async_setup_ingress_view(hass: HomeAssistant, host: str) -> None:
"""Auth setup.""" """Auth setup."""
websession = async_get_clientsession(hass) websession = async_get_clientsession(hass)
@ -281,7 +281,10 @@ def _is_websocket(request: web.Request) -> bool:
) )
async def _websocket_forward(ws_from, ws_to): async def _websocket_forward(
ws_from: web.WebSocketResponse | ClientWebSocketResponse,
ws_to: web.WebSocketResponse | ClientWebSocketResponse,
) -> None:
"""Handle websocket message directly.""" """Handle websocket message directly."""
try: try:
async for msg in ws_from: async for msg in ws_from:
@ -294,7 +297,7 @@ async def _websocket_forward(ws_from, ws_to):
elif msg.type == aiohttp.WSMsgType.PONG: elif msg.type == aiohttp.WSMsgType.PONG:
await ws_to.pong() await ws_to.pong()
elif ws_to.closed: elif ws_to.closed:
await ws_to.close(code=ws_to.close_code, message=msg.extra) await ws_to.close(code=ws_to.close_code, message=msg.extra) # type: ignore[arg-type]
except RuntimeError: except RuntimeError:
_LOGGER.debug("Ingress Websocket runtime error") _LOGGER.debug("Ingress Websocket runtime error")
except ConnectionResetError: except ConnectionResetError:

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import os import os
from typing import Any
from homeassistant.components import system_health from homeassistant.components import system_health
from homeassistant.core import HomeAssistant, callback from homeassistant.core import HomeAssistant, callback
@ -20,7 +21,7 @@ def async_register(
register.async_register_info(system_health_info) register.async_register_info(system_health_info)
async def system_health_info(hass: HomeAssistant): async def system_health_info(hass: HomeAssistant) -> dict[str, Any]:
"""Get info for the info page.""" """Get info for the info page."""
info = get_info(hass) or {} info = get_info(hass) or {}
host_info = get_host_info(hass) or {} host_info = get_host_info(hass) or {}

View File

@ -54,7 +54,7 @@ _LOGGER: logging.Logger = logging.getLogger(__package__)
@callback @callback
def async_load_websocket_api(hass: HomeAssistant): def async_load_websocket_api(hass: HomeAssistant) -> None:
"""Set up the websocket API.""" """Set up the websocket API."""
websocket_api.async_register_command(hass, websocket_supervisor_event) websocket_api.async_register_command(hass, websocket_supervisor_event)
websocket_api.async_register_command(hass, websocket_supervisor_api) websocket_api.async_register_command(hass, websocket_supervisor_api)
@ -66,11 +66,11 @@ def async_load_websocket_api(hass: HomeAssistant):
@websocket_api.async_response @websocket_api.async_response
async def websocket_subscribe( async def websocket_subscribe(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any] hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
): ) -> None:
"""Subscribe to supervisor events.""" """Subscribe to supervisor events."""
@callback @callback
def forward_messages(data): def forward_messages(data: dict[str, str]) -> None:
"""Forward events to websocket.""" """Forward events to websocket."""
connection.send_message(websocket_api.event_message(msg[WS_ID], data)) connection.send_message(websocket_api.event_message(msg[WS_ID], data))
@ -89,7 +89,7 @@ async def websocket_subscribe(
@websocket_api.async_response @websocket_api.async_response
async def websocket_supervisor_event( async def websocket_supervisor_event(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any] hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
): ) -> None:
"""Publish events from the Supervisor.""" """Publish events from the Supervisor."""
connection.send_result(msg[WS_ID]) connection.send_result(msg[WS_ID])
async_dispatcher_send(hass, EVENT_SUPERVISOR_EVENT, msg[ATTR_DATA]) async_dispatcher_send(hass, EVENT_SUPERVISOR_EVENT, msg[ATTR_DATA])
@ -107,7 +107,7 @@ async def websocket_supervisor_event(
@websocket_api.async_response @websocket_api.async_response
async def websocket_supervisor_api( async def websocket_supervisor_api(
hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any] hass: HomeAssistant, connection: ActiveConnection, msg: dict[str, Any]
): ) -> None:
"""Websocket handler to call Supervisor API.""" """Websocket handler to call Supervisor API."""
if not connection.user.is_admin and not WS_NO_ADMIN_ENDPOINTS.match( if not connection.user.is_admin and not WS_NO_ADMIN_ENDPOINTS.match(
msg[ATTR_ENDPOINT] msg[ATTR_ENDPOINT]