From a70f9b8995fc65d6009c96c5a974576c09e521b2 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Wed, 19 Oct 2022 14:35:23 +0200 Subject: [PATCH] Add type hints to rest integration (#80546) --- homeassistant/components/rest/__init__.py | 60 ++++++++++++++--------- homeassistant/components/rest/data.py | 35 +++++++------ 2 files changed, 55 insertions(+), 40 deletions(-) diff --git a/homeassistant/components/rest/__init__.py b/homeassistant/components/rest/__init__.py index 96ed1ada7ae3..5549abc21433 100644 --- a/homeassistant/components/rest/__init__.py +++ b/homeassistant/components/rest/__init__.py @@ -1,7 +1,9 @@ """The rest component.""" +from __future__ import annotations import asyncio import contextlib +from datetime import timedelta import logging import httpx @@ -34,7 +36,7 @@ from homeassistant.helpers.reload import ( async_integration_yaml_config, async_reload_integration_platforms, ) -from homeassistant.helpers.typing import ConfigType +from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from .const import COORDINATOR, DOMAIN, PLATFORM_IDX, REST, REST_DATA, REST_IDX @@ -76,21 +78,22 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool: @callback -def _async_setup_shared_data(hass: HomeAssistant): +def _async_setup_shared_data(hass: HomeAssistant) -> None: """Create shared data for platform config and rest coordinators.""" hass.data[DOMAIN] = {key: [] for key in (REST_DATA, *COORDINATOR_AWARE_PLATFORMS)} -async def _async_process_config(hass, config) -> bool: +async def _async_process_config(hass: HomeAssistant, config: ConfigType) -> bool: """Process rest configuration.""" if DOMAIN not in config: return True refresh_tasks = [] load_tasks = [] - for rest_idx, conf in enumerate(config[DOMAIN]): - scan_interval = conf.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL) - resource_template = conf.get(CONF_RESOURCE_TEMPLATE) + rest_config: list[ConfigType] = config[DOMAIN] + for rest_idx, conf in enumerate(rest_config): + scan_interval: timedelta = conf.get(CONF_SCAN_INTERVAL, DEFAULT_SCAN_INTERVAL) + resource_template: template.Template | None = conf.get(CONF_RESOURCE_TEMPLATE) rest = create_rest_data_from_config(hass, conf) coordinator = _rest_coordinator(hass, rest, resource_template, scan_interval) refresh_tasks.append(coordinator.async_refresh()) @@ -122,18 +125,25 @@ async def _async_process_config(hass, config) -> bool: return True -async def async_get_config_and_coordinator(hass, platform_domain, discovery_info): +async def async_get_config_and_coordinator( + hass: HomeAssistant, platform_domain: str, discovery_info: DiscoveryInfoType +) -> tuple[ConfigType, DataUpdateCoordinator[None], RestData]: """Get the config and coordinator for the platform from discovery.""" shared_data = hass.data[DOMAIN][REST_DATA][discovery_info[REST_IDX]] - conf = hass.data[DOMAIN][platform_domain][discovery_info[PLATFORM_IDX]] - coordinator = shared_data[COORDINATOR] - rest = shared_data[REST] + conf: ConfigType = hass.data[DOMAIN][platform_domain][discovery_info[PLATFORM_IDX]] + coordinator: DataUpdateCoordinator[None] = shared_data[COORDINATOR] + rest: RestData = shared_data[REST] if rest.data is None: await coordinator.async_request_refresh() return conf, coordinator, rest -def _rest_coordinator(hass, rest, resource_template, update_interval): +def _rest_coordinator( + hass: HomeAssistant, + rest: RestData, + resource_template: template.Template | None, + update_interval: timedelta, +) -> DataUpdateCoordinator[None]: """Wrap a DataUpdateCoordinator around the rest object.""" if resource_template: @@ -154,33 +164,35 @@ def _rest_coordinator(hass, rest, resource_template, update_interval): ) -def create_rest_data_from_config(hass, config): +def create_rest_data_from_config(hass: HomeAssistant, config: ConfigType) -> RestData: """Create RestData from config.""" - resource = config.get(CONF_RESOURCE) - resource_template = config.get(CONF_RESOURCE_TEMPLATE) - method = config.get(CONF_METHOD) - payload = config.get(CONF_PAYLOAD) - verify_ssl = config.get(CONF_VERIFY_SSL) - username = config.get(CONF_USERNAME) - password = config.get(CONF_PASSWORD) - headers = config.get(CONF_HEADERS) - params = config.get(CONF_PARAMS) - timeout = config.get(CONF_TIMEOUT) + resource: str | None = config.get(CONF_RESOURCE) + resource_template: template.Template | None = config.get(CONF_RESOURCE_TEMPLATE) + method: str = config[CONF_METHOD] + payload: str | None = config.get(CONF_PAYLOAD) + verify_ssl: bool = config[CONF_VERIFY_SSL] + username: str | None = config.get(CONF_USERNAME) + password: str | None = config.get(CONF_PASSWORD) + headers: dict[str, str] | None = config.get(CONF_HEADERS) + params: dict[str, str] | None = config.get(CONF_PARAMS) + timeout: int = config[CONF_TIMEOUT] if resource_template is not None: resource_template.hass = hass resource = resource_template.async_render(parse_result=False) + if not resource: + raise HomeAssistantError("Resource not set for RestData") + template.attach(hass, headers) template.attach(hass, params) + auth: httpx.DigestAuth | tuple[str, str] | None = None if username and password: if config.get(CONF_AUTHENTICATION) == HTTP_DIGEST_AUTHENTICATION: auth = httpx.DigestAuth(username, password) else: auth = (username, password) - else: - auth = None return RestData( hass, method, resource, auth, headers, params, payload, verify_ssl, timeout diff --git a/homeassistant/components/rest/data.py b/homeassistant/components/rest/data.py index 351c59a9f52a..c1990b283368 100644 --- a/homeassistant/components/rest/data.py +++ b/homeassistant/components/rest/data.py @@ -1,8 +1,11 @@ """Support for RESTful API.""" +from __future__ import annotations + import logging import httpx +from homeassistant.core import HomeAssistant from homeassistant.helpers import template from homeassistant.helpers.httpx_client import get_async_client @@ -16,16 +19,16 @@ class RestData: def __init__( self, - hass, - method, - resource, - auth, - headers, - params, - data, - verify_ssl, - timeout=DEFAULT_TIMEOUT, - ): + hass: HomeAssistant, + method: str, + resource: str, + auth: httpx.DigestAuth | tuple[str, str] | None, + headers: dict[str, str] | None, + params: dict[str, str] | None, + data: str | None, + verify_ssl: bool, + timeout: int = DEFAULT_TIMEOUT, + ) -> None: """Initialize the data object.""" self._hass = hass self._method = method @@ -36,16 +39,16 @@ class RestData: self._request_data = data self._timeout = timeout self._verify_ssl = verify_ssl - self._async_client = None - self.data = None - self.last_exception = None - self.headers = None + self._async_client: httpx.AsyncClient | None = None + self.data: str | None = None + self.last_exception: Exception | None = None + self.headers: httpx.Headers | None = None - def set_url(self, url): + def set_url(self, url: str) -> None: """Set url.""" self._resource = url - async def async_update(self, log_errors=True): + async def async_update(self, log_errors: bool = True) -> None: """Get the latest data from REST service with provided method.""" if not self._async_client: self._async_client = get_async_client(