Add TypeVar defaults for DataUpdateCoordinator and EntityComponent (#95026)

homeassistant/components/bluetooth/passive_update_coordinator.py

@@ -1,7 +1,9 @@
 """Passive update coordinator for the Bluetooth integration."""
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any, TypeVar
+from typing import TYPE_CHECKING, Any
+
+from typing_extensions import TypeVar
 
 from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
 from homeassistant.helpers.update_coordinator import (
@@ -20,6 +22,7 @@ if TYPE_CHECKING:
 _PassiveBluetoothDataUpdateCoordinatorT = TypeVar(
     "_PassiveBluetoothDataUpdateCoordinatorT",
     bound="PassiveBluetoothDataUpdateCoordinator",
+    default="PassiveBluetoothDataUpdateCoordinator",
 )
 
 

homeassistant/helpers/collection.py

@@ -8,8 +8,9 @@ from dataclasses import dataclass
 from itertools import groupby
 import logging
 from operator import attrgetter
-from typing import Any, Generic, TypedDict, TypeVar
+from typing import Any, Generic, TypedDict
 
+from typing_extensions import TypeVar
 import voluptuous as vol
 from voluptuous.humanize import humanize_error
 
@@ -35,6 +36,7 @@ CHANGE_REMOVED = "removed"
 _ItemT = TypeVar("_ItemT")
 _StoreT = TypeVar("_StoreT", bound="SerializedStorageCollection")
 _StorageCollectionT = TypeVar("_StorageCollectionT", bound="StorageCollection")
+_EntityT = TypeVar("_EntityT", bound=Entity, default=Entity)
 
 
 @dataclass(slots=True)
@@ -421,7 +423,7 @@ def sync_entity_lifecycle(
     hass: HomeAssistant,
     domain: str,
     platform: str,
-    entity_component: EntityComponent,
+    entity_component: EntityComponent[_EntityT],
     collection: StorageCollection | YamlCollection,
     entity_class: type[CollectionEntity],
 ) -> None:

homeassistant/helpers/entity_component.py

@@ -7,8 +7,9 @@ from datetime import timedelta
 from itertools import chain
 import logging
 from types import ModuleType
-from typing import Any, Generic, TypeVar
+from typing import Any, Generic
 
+from typing_extensions import TypeVar
 import voluptuous as vol
 
 from homeassistant import config as conf_util
@@ -30,7 +31,7 @@ from .typing import ConfigType, DiscoveryInfoType
 DEFAULT_SCAN_INTERVAL = timedelta(seconds=15)
 DATA_INSTANCES = "entity_components"
 
-_EntityT = TypeVar("_EntityT", bound=entity.Entity)
+_EntityT = TypeVar("_EntityT", bound=entity.Entity, default=entity.Entity)
 
 
 @bind_hass

homeassistant/helpers/update_coordinator.py

@@ -8,11 +8,12 @@ from datetime import datetime, timedelta
 import logging
 from random import randint
 from time import monotonic
-from typing import Any, Generic, Protocol, TypeVar
+from typing import Any, Generic, Protocol
 import urllib.error
 
 import aiohttp
 import requests
+from typing_extensions import TypeVar
 
 from homeassistant import config_entries
 from homeassistant.const import EVENT_HOMEASSISTANT_STOP
@@ -30,12 +31,14 @@ from .debounce import Debouncer
 REQUEST_REFRESH_DEFAULT_COOLDOWN = 10
 REQUEST_REFRESH_DEFAULT_IMMEDIATE = True
 
-_T = TypeVar("_T")
+_DataT = TypeVar("_DataT", default=dict[str, Any])
 _BaseDataUpdateCoordinatorT = TypeVar(
     "_BaseDataUpdateCoordinatorT", bound="BaseDataUpdateCoordinatorProtocol"
 )
 _DataUpdateCoordinatorT = TypeVar(
-    "_DataUpdateCoordinatorT", bound="DataUpdateCoordinator[Any]"
+    "_DataUpdateCoordinatorT",
+    bound="DataUpdateCoordinator[Any]",
+    default="DataUpdateCoordinator[dict[str, Any]]",
 )
 
 
@@ -53,7 +56,7 @@ class BaseDataUpdateCoordinatorProtocol(Protocol):
         """Listen for data updates."""
 
 
-class DataUpdateCoordinator(BaseDataUpdateCoordinatorProtocol, Generic[_T]):
+class DataUpdateCoordinator(BaseDataUpdateCoordinatorProtocol, Generic[_DataT]):
     """Class to manage fetching data from single endpoint."""
 
     def __init__(
@@ -63,7 +66,7 @@ class DataUpdateCoordinator(BaseDataUpdateCoordinatorProtocol, Generic[_T]):
         *,
         name: str,
         update_interval: timedelta | None = None,
-        update_method: Callable[[], Awaitable[_T]] | None = None,
+        update_method: Callable[[], Awaitable[_DataT]] | None = None,
         request_refresh_debouncer: Debouncer[Coroutine[Any, Any, None]] | None = None,
     ) -> None:
         """Initialize global data updater."""
@@ -80,7 +83,7 @@ class DataUpdateCoordinator(BaseDataUpdateCoordinatorProtocol, Generic[_T]):
         # to make sure the first update was successful.
         # Set type to just T to remove annoying checks that data is not None
         # when it was already checked during setup.
-        self.data: _T = None  # type: ignore[assignment]
+        self.data: _DataT = None  # type: ignore[assignment]
 
         # Pick a random microsecond to stagger the refreshes
         # and avoid a thundering herd.
@@ -235,7 +238,7 @@ class DataUpdateCoordinator(BaseDataUpdateCoordinatorProtocol, Generic[_T]):
         """
         await self._debounced_refresh.async_call()
 
-    async def _async_update_data(self) -> _T:
+    async def _async_update_data(self) -> _DataT:
         """Fetch the latest data from the source."""
         if self.update_method is None:
             raise NotImplementedError("Update method not implemented")
@@ -383,7 +386,7 @@ class DataUpdateCoordinator(BaseDataUpdateCoordinatorProtocol, Generic[_T]):
             self.async_update_listeners()
 
     @callback
-    def async_set_updated_data(self, data: _T) -> None:
+    def async_set_updated_data(self, data: _DataT) -> None:
         """Manually update data, notify listeners and reset refresh interval."""
         self._async_unsub_refresh()
         self._debounced_refresh.async_cancel()