Enforce strict typing for IQVIA (#53408)

* Enforce strict typing for IQVIA

* Cleanup

* Code review

* Ignore untyped numpy function

.strict-typing

@@ -54,6 +54,7 @@ homeassistant.components.huawei_lte.*
 homeassistant.components.hyperion.*
 homeassistant.components.image_processing.*
 homeassistant.components.integration.*
+homeassistant.components.iqvia.*
 homeassistant.components.knx.*
 homeassistant.components.kraken.*
 homeassistant.components.lcn.*

homeassistant/components/iqvia/__init__.py

@@ -1,14 +1,19 @@
 """Support for IQVIA."""
+from __future__ import annotations
+
 import asyncio
+from collections.abc import Awaitable
 from datetime import timedelta
 from functools import partial
+from typing import Any, Callable, Dict, cast
 
 from pyiqvia import Client
 from pyiqvia.errors import IQVIAError
 
 from homeassistant.components.sensor import SensorEntity
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import ATTR_ATTRIBUTION
-from homeassistant.core import callback
+from homeassistant.core import HomeAssistant, callback
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import aiohttp_client
 from homeassistant.helpers.update_coordinator import (
@@ -37,7 +42,7 @@ DEFAULT_SCAN_INTERVAL = timedelta(minutes=30)
 PLATFORMS = ["sensor"]
 
 
-async def async_setup_entry(hass, entry):
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up IQVIA as config entry."""
     hass.data.setdefault(DOMAIN, {})
     coordinators = {}
@@ -51,13 +56,17 @@ async def async_setup_entry(hass, entry):
     websession = aiohttp_client.async_get_clientsession(hass)
     client = Client(entry.data[CONF_ZIP_CODE], session=websession)
 
-    async def async_get_data_from_api(api_coro):
+    async def async_get_data_from_api(
+        api_coro: Callable[..., Awaitable]
+    ) -> dict[str, Any]:
         """Get data from a particular API coroutine."""
         try:
-            return await api_coro()
+            data = await api_coro()
         except IQVIAError as err:
             raise UpdateFailed from err
 
+        return cast(Dict[str, Any], data)
+
     init_data_update_tasks = []
     for sensor_type, api_coro in (
         (TYPE_ALLERGY_FORECAST, client.allergens.extended),
@@ -90,7 +99,7 @@ async def async_setup_entry(hass, entry):
     return True
 
 
-async def async_unload_entry(hass, entry):
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Unload an OpenUV config entry."""
     unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
     if unload_ok:
@@ -101,7 +110,14 @@ async def async_unload_entry(hass, entry):
 class IQVIAEntity(CoordinatorEntity, SensorEntity):
     """Define a base IQVIA entity."""
 
-    def __init__(self, coordinator, entry, sensor_type, name, icon):
+    def __init__(
+        self,
+        coordinator: DataUpdateCoordinator,
+        entry: ConfigEntry,
+        sensor_type: str,
+        name: str,
+        icon: str,
+    ) -> None:
         """Initialize."""
         super().__init__(coordinator)
 
@@ -122,7 +138,7 @@ class IQVIAEntity(CoordinatorEntity, SensorEntity):
         self.update_from_latest_data()
         self.async_write_ha_state()
 
-    async def async_added_to_hass(self):
+    async def async_added_to_hass(self) -> None:
         """Register callbacks."""
         await super().async_added_to_hass()
 
@@ -136,6 +152,6 @@ class IQVIAEntity(CoordinatorEntity, SensorEntity):
         self.update_from_latest_data()
 
     @callback
-    def update_from_latest_data(self):
+    def update_from_latest_data(self) -> None:
         """Update the entity from the latest data."""
         raise NotImplementedError

homeassistant/components/iqvia/config_flow.py

@@ -1,9 +1,14 @@
 """Config flow to configure the IQVIA component."""
+from __future__ import annotations
+
+from typing import Any
+
 from pyiqvia import Client
 from pyiqvia.errors import InvalidZipError
 import voluptuous as vol
 
 from homeassistant import config_entries
+from homeassistant.data_entry_flow import FlowResult
 from homeassistant.helpers import aiohttp_client
 
 from .const import CONF_ZIP_CODE, DOMAIN
@@ -14,11 +19,13 @@ class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
 
     VERSION = 1
 
-    def __init__(self):
+    def __init__(self) -> None:
         """Initialize the config flow."""
         self.data_schema = vol.Schema({vol.Required(CONF_ZIP_CODE): str})
 
-    async def async_step_user(self, user_input=None):
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
         """Handle the start of the config flow."""
         if not user_input:
             return self.async_show_form(step_id="user", data_schema=self.data_schema)

homeassistant/components/iqvia/sensor.py

@@ -1,10 +1,14 @@
 """Support for IQVIA sensors."""
+from __future__ import annotations
+
 from statistics import mean
 
 import numpy as np
 
+from homeassistant.config_entries import ConfigEntry
 from homeassistant.const import ATTR_STATE
-from homeassistant.core import callback
+from homeassistant.core import HomeAssistant, callback
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
 
 from . import IQVIAEntity
 from .const import (
@@ -58,7 +62,9 @@ TREND_INCREASING = "Increasing"
 TREND_SUBSIDING = "Subsiding"
 
 
-async def async_setup_entry(hass, entry, async_add_entities):
+async def async_setup_entry(
+    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
+) -> None:
     """Set up IQVIA sensors based on a config entry."""
     sensor_class_mapping = {
         TYPE_ALLERGY_FORECAST: ForecastSensor,
@@ -76,17 +82,17 @@ async def async_setup_entry(hass, entry, async_add_entities):
         api_category = API_CATEGORY_MAPPING.get(sensor_type, sensor_type)
         coordinator = hass.data[DOMAIN][DATA_COORDINATOR][entry.entry_id][api_category]
         sensor_class = sensor_class_mapping[sensor_type]
-
         sensors.append(sensor_class(coordinator, entry, sensor_type, name, icon))
 
     async_add_entities(sensors)
 
 
-def calculate_trend(indices):
+@callback
+def calculate_trend(indices: list[float]) -> str:
     """Calculate the "moving average" of a set of indices."""
     index_range = np.arange(0, len(indices))
     index_array = np.array(indices)
-    linear_fit = np.polyfit(index_range, index_array, 1)
+    linear_fit = np.polyfit(index_range, index_array, 1)  # type: ignore
     slope = round(linear_fit[0], 2)
 
     if slope > 0:
@@ -102,7 +108,7 @@ class ForecastSensor(IQVIAEntity):
     """Define sensor related to forecast data."""
 
     @callback
-    def update_from_latest_data(self):
+    def update_from_latest_data(self) -> None:
         """Update the sensor."""
         if not self.available:
             return
@@ -151,7 +157,7 @@ class IndexSensor(IQVIAEntity):
     """Define sensor related to indices."""
 
     @callback
-    def update_from_latest_data(self):
+    def update_from_latest_data(self) -> None:
         """Update the sensor."""
         if not self.coordinator.last_update_success:
             return

mypy.ini

@@ -605,6 +605,17 @@ no_implicit_optional = true
 warn_return_any = true
 warn_unreachable = true
 
+[mypy-homeassistant.components.iqvia.*]
+check_untyped_defs = true
+disallow_incomplete_defs = true
+disallow_subclassing_any = true
+disallow_untyped_calls = true
+disallow_untyped_decorators = true
+disallow_untyped_defs = true
+no_implicit_optional = true
+warn_return_any = true
+warn_unreachable = true
+
 [mypy-homeassistant.components.knx.*]
 check_untyped_defs = true
 disallow_incomplete_defs = true