mirror of
https://github.com/home-assistant/core
synced 2024-07-27 18:58:57 +02:00
Add type hints to rest integration (#80546)
This commit is contained in:
parent
dc2a87b9ae
commit
a70f9b8995
@ -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
|
||||
|
@ -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(
|
||||
|
Loading…
Reference in New Issue
Block a user