diff --git a/homeassistant/components/bluetooth/passive_update_coordinator.py b/homeassistant/components/bluetooth/passive_update_coordinator.py index ccff85e5027546..be232f87b24a3f 100644 --- a/homeassistant/components/bluetooth/passive_update_coordinator.py +++ b/homeassistant/components/bluetooth/passive_update_coordinator.py @@ -4,6 +4,8 @@ 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 ( BaseCoordinatorEntity, @@ -18,6 +20,12 @@ from . import BluetoothChange, BluetoothScanningMode, BluetoothServiceInfoBleak +_PassiveBluetoothDataUpdateCoordinatorT = TypeVar( + "_PassiveBluetoothDataUpdateCoordinatorT", + bound="PassiveBluetoothDataUpdateCoordinator", + default="PassiveBluetoothDataUpdateCoordinator", +) + class PassiveBluetoothDataUpdateCoordinator( BasePassiveBluetoothCoordinator, BaseDataUpdateCoordinatorProtocol @@ -90,9 +98,7 @@ def _async_handle_bluetooth_event( self.async_update_listeners() -class PassiveBluetoothCoordinatorEntity[ - _PassiveBluetoothDataUpdateCoordinatorT: PassiveBluetoothDataUpdateCoordinator = PassiveBluetoothDataUpdateCoordinator -]( # pylint: disable=hass-enforce-class-module +class PassiveBluetoothCoordinatorEntity( # pylint: disable=hass-enforce-class-module BaseCoordinatorEntity[_PassiveBluetoothDataUpdateCoordinatorT] ): """A class for entities using DataUpdateCoordinator.""" diff --git a/homeassistant/components/broadlink/device.py b/homeassistant/components/broadlink/device.py index 082af07ebbd045..75b6236a4731ef 100644 --- a/homeassistant/components/broadlink/device.py +++ b/homeassistant/components/broadlink/device.py @@ -3,6 +3,7 @@ from contextlib import suppress from functools import partial import logging +from typing import Generic import broadlink as blk from broadlink.exceptions import ( @@ -12,6 +13,7 @@ ConnectionClosedError, NetworkTimeoutError, ) +from typing_extensions import TypeVar from homeassistant.config_entries import ConfigEntry from homeassistant.const import ( @@ -29,6 +31,8 @@ from .const import DEFAULT_PORT, DOMAIN, DOMAINS_AND_TYPES from .updater import BroadlinkUpdateManager, get_update_manager +_ApiT = TypeVar("_ApiT", bound=blk.Device, default=blk.Device) + _LOGGER = logging.getLogger(__name__) @@ -37,7 +41,7 @@ def get_domains(device_type: str) -> set[Platform]: return {d for d, t in DOMAINS_AND_TYPES.items() if device_type in t} -class BroadlinkDevice[_ApiT: blk.Device = blk.Device]: +class BroadlinkDevice(Generic[_ApiT]): """Manages a Broadlink device.""" api: _ApiT diff --git a/homeassistant/components/weather/__init__.py b/homeassistant/components/weather/__init__.py index 1333045dec8bfe..557765795ee88d 100644 --- a/homeassistant/components/weather/__init__.py +++ b/homeassistant/components/weather/__init__.py @@ -8,9 +8,10 @@ from datetime import timedelta from functools import partial import logging -from typing import Any, Final, Literal, Required, TypedDict, cast, final +from typing import Any, Final, Generic, Literal, Required, TypedDict, cast, final from propcache import cached_property +from typing_extensions import TypeVar import voluptuous as vol from homeassistant.config_entries import ConfigEntry @@ -123,6 +124,28 @@ SERVICE_GET_FORECASTS: Final = "get_forecasts" +_ObservationUpdateCoordinatorT = TypeVar( + "_ObservationUpdateCoordinatorT", + bound=DataUpdateCoordinator[Any], + default=DataUpdateCoordinator[dict[str, Any]], +) + +_DailyForecastUpdateCoordinatorT = TypeVar( + "_DailyForecastUpdateCoordinatorT", + bound=TimestampDataUpdateCoordinator[Any], + default=TimestampDataUpdateCoordinator[None], +) +_HourlyForecastUpdateCoordinatorT = TypeVar( + "_HourlyForecastUpdateCoordinatorT", + bound=TimestampDataUpdateCoordinator[Any], + default=_DailyForecastUpdateCoordinatorT, +) +_TwiceDailyForecastUpdateCoordinatorT = TypeVar( + "_TwiceDailyForecastUpdateCoordinatorT", + bound=TimestampDataUpdateCoordinator[Any], + default=_DailyForecastUpdateCoordinatorT, +) + # mypy: disallow-any-generics @@ -998,22 +1021,15 @@ async def async_get_forecasts_service( } -class CoordinatorWeatherEntity[ - _ObservationUpdateCoordinatorT: DataUpdateCoordinator[Any] = DataUpdateCoordinator[ - dict[str, Any] - ], - _DailyForecastUpdateCoordinatorT: DataUpdateCoordinator[ - Any - ] = TimestampDataUpdateCoordinator[None], - _HourlyForecastUpdateCoordinatorT: DataUpdateCoordinator[ - Any - ] = _DailyForecastUpdateCoordinatorT, - _TwiceDailyForecastUpdateCoordinatorT: DataUpdateCoordinator[ - Any - ] = _DailyForecastUpdateCoordinatorT, -]( +class CoordinatorWeatherEntity( CoordinatorEntity[_ObservationUpdateCoordinatorT], WeatherEntity, + Generic[ + _ObservationUpdateCoordinatorT, + _DailyForecastUpdateCoordinatorT, + _HourlyForecastUpdateCoordinatorT, + _TwiceDailyForecastUpdateCoordinatorT, + ], ): """A class for weather entities using DataUpdateCoordinators.""" @@ -1176,11 +1192,7 @@ async def async_forecast_twice_daily(self) -> list[Forecast] | None: return await self._async_forecast("twice_daily") -class SingleCoordinatorWeatherEntity[ - _ObservationUpdateCoordinatorT: DataUpdateCoordinator[Any] = DataUpdateCoordinator[ - dict[str, Any] - ] -]( +class SingleCoordinatorWeatherEntity( CoordinatorWeatherEntity[ _ObservationUpdateCoordinatorT, TimestampDataUpdateCoordinator[None] ], diff --git a/homeassistant/helpers/collection.py b/homeassistant/helpers/collection.py index 08b58aedde4fd1..86d3450c3a012f 100644 --- a/homeassistant/helpers/collection.py +++ b/homeassistant/helpers/collection.py @@ -11,8 +11,9 @@ from itertools import groupby import logging from operator import attrgetter -from typing import Any, TypedDict +from typing import Any, Generic, TypedDict +from typing_extensions import TypeVar import voluptuous as vol from voluptuous.humanize import humanize_error @@ -36,6 +37,8 @@ CHANGE_UPDATED = "updated" CHANGE_REMOVED = "removed" +_EntityT = TypeVar("_EntityT", bound=Entity, default=Entity) + @dataclass(slots=True) class CollectionChange: @@ -445,7 +448,7 @@ async def async_load(self, data: list[dict]) -> None: @dataclass(slots=True, frozen=True) -class _CollectionLifeCycle[_EntityT: Entity = Entity]: +class _CollectionLifeCycle(Generic[_EntityT]): """Life cycle for a collection of entities.""" domain: str @@ -520,7 +523,7 @@ async def _collection_changed(self, change_set: Iterable[CollectionChange]) -> N @callback -def sync_entity_lifecycle[_EntityT: Entity = Entity]( +def sync_entity_lifecycle( hass: HomeAssistant, domain: str, platform: str, diff --git a/homeassistant/helpers/entity_component.py b/homeassistant/helpers/entity_component.py index 02508e9ee9e708..1be7289401ccfa 100644 --- a/homeassistant/helpers/entity_component.py +++ b/homeassistant/helpers/entity_component.py @@ -7,7 +7,9 @@ from datetime import timedelta import logging from types import ModuleType -from typing import Any +from typing import Any, Generic + +from typing_extensions import TypeVar from homeassistant import config as conf_util from homeassistant.config_entries import ConfigEntry @@ -37,6 +39,8 @@ DEFAULT_SCAN_INTERVAL = timedelta(seconds=15) DATA_INSTANCES = "entity_components" +_EntityT = TypeVar("_EntityT", bound=entity.Entity, default=entity.Entity) + @bind_hass async def async_update_entity(hass: HomeAssistant, entity_id: str) -> None: @@ -60,7 +64,7 @@ async def async_update_entity(hass: HomeAssistant, entity_id: str) -> None: await entity_obj.async_update_ha_state(True) -class EntityComponent[_EntityT: entity.Entity = entity.Entity]: +class EntityComponent(Generic[_EntityT]): """The EntityComponent manages platforms that manage entities. An example of an entity component is 'light', which manages platforms such diff --git a/homeassistant/helpers/update_coordinator.py b/homeassistant/helpers/update_coordinator.py index d86e995ad3bf80..6cc4584935e5f4 100644 --- a/homeassistant/helpers/update_coordinator.py +++ b/homeassistant/helpers/update_coordinator.py @@ -37,6 +37,11 @@ REQUEST_REFRESH_DEFAULT_IMMEDIATE = True _DataT = TypeVar("_DataT", default=dict[str, Any]) +_DataUpdateCoordinatorT = TypeVar( + "_DataUpdateCoordinatorT", + bound="DataUpdateCoordinator[Any]", + default="DataUpdateCoordinator[dict[str, Any]]", +) class UpdateFailed(HomeAssistantError): @@ -560,11 +565,7 @@ async def async_update(self) -> None: """ -class CoordinatorEntity[ - _DataUpdateCoordinatorT: DataUpdateCoordinator[Any] = DataUpdateCoordinator[ - dict[str, Any] - ] -](BaseCoordinatorEntity[_DataUpdateCoordinatorT]): +class CoordinatorEntity(BaseCoordinatorEntity[_DataUpdateCoordinatorT]): """A class for entities using DataUpdateCoordinator.""" def __init__( diff --git a/homeassistant/util/event_type.py b/homeassistant/util/event_type.py index e008305727275b..509a35d33ae2a9 100644 --- a/homeassistant/util/event_type.py +++ b/homeassistant/util/event_type.py @@ -6,10 +6,14 @@ from __future__ import annotations from collections.abc import Mapping -from typing import Any +from typing import Any, Generic +from typing_extensions import TypeVar -class EventType[_DataT: Mapping[str, Any] = Mapping[str, Any]](str): +_DataT = TypeVar("_DataT", bound=Mapping[str, Any], default=Mapping[str, Any]) + + +class EventType(str, Generic[_DataT]): """Custom type for Event.event_type. At runtime this is a generic subclass of str. diff --git a/homeassistant/util/event_type.pyi b/homeassistant/util/event_type.pyi index 7536babbcb9a6e..4285e54e8c98b1 100644 --- a/homeassistant/util/event_type.pyi +++ b/homeassistant/util/event_type.pyi @@ -2,13 +2,17 @@ # ruff: noqa: PYI021 # Allow docstrings from collections.abc import Mapping -from typing import Any +from typing import Any, Generic + +from typing_extensions import TypeVar __all__ = [ "EventType", ] -class EventType[_DataT: Mapping[str, Any] = Mapping[str, Any]]: +_DataT = TypeVar("_DataT", bound=Mapping[str, Any], default=Mapping[str, Any]) + +class EventType(Generic[_DataT]): """Custom type for Event.event_type. At runtime delegated to str. For type checkers pretend to be its own separate class. diff --git a/tests/common.py b/tests/common.py index a1492df5dac0a6..ac6f10b8c44ef8 100644 --- a/tests/common.py +++ b/tests/common.py @@ -31,6 +31,7 @@ from aiohttp.test_utils import unused_port as get_test_instance_port # noqa: F401 import pytest from syrupy import SnapshotAssertion +from typing_extensions import TypeVar import voluptuous as vol from homeassistant import auth, bootstrap, config_entries, loader @@ -113,6 +114,8 @@ import_deprecated_constant, ) +_DataT = TypeVar("_DataT", bound=Mapping[str, Any], default=dict[str, Any]) + _LOGGER = logging.getLogger(__name__) INSTANCES = [] CLIENT_ID = "https://example.com/app" @@ -1534,7 +1537,7 @@ def mock_platform( module_cache[platform_path] = module or Mock() -def async_capture_events[_DataT: Mapping[str, Any] = dict[str, Any]]( +def async_capture_events( hass: HomeAssistant, event_name: EventType[_DataT] | str ) -> list[Event[_DataT]]: """Create a helper that captures events."""