Use PEP 695 for class annotations (3) (#117777)

pull/117191/head^2
Marc Mueller 2024-05-20 10:46:01 +02:00 committed by GitHub
parent 8f0fb4db3e
commit 7b27101f8a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 38 additions and 59 deletions

View File

@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Any, TypeVar
from typing import Any
from switchbee.api.central_unit import SwitchBeeDeviceOfflineError, SwitchBeeError
from switchbee.device import (
@ -23,16 +23,6 @@ from .const import DOMAIN
from .coordinator import SwitchBeeCoordinator
from .entity import SwitchBeeDeviceEntity
_DeviceTypeT = TypeVar(
"_DeviceTypeT",
bound=(
SwitchBeeTimedSwitch
| SwitchBeeGroupSwitch
| SwitchBeeSwitch
| SwitchBeeTimerSwitch
),
)
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
@ -55,7 +45,12 @@ async def async_setup_entry(
)
class SwitchBeeSwitchEntity(SwitchBeeDeviceEntity[_DeviceTypeT], SwitchEntity):
class SwitchBeeSwitchEntity[
_DeviceTypeT: SwitchBeeTimedSwitch
| SwitchBeeGroupSwitch
| SwitchBeeSwitch
| SwitchBeeTimerSwitch
](SwitchBeeDeviceEntity[_DeviceTypeT], SwitchEntity):
"""Representation of a Switchbee switch."""
def __init__(

View File

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Awaitable, Callable, Coroutine
from datetime import timedelta
import logging
from typing import Any, Concatenate, TypeVar
from typing import Any, Concatenate
from synology_dsm.api.surveillance_station.camera import SynoCamera
from synology_dsm.exceptions import (
@ -28,7 +28,6 @@ from .const import (
)
_LOGGER = logging.getLogger(__name__)
_DataT = TypeVar("_DataT")
def async_re_login_on_expired[_T: SynologyDSMUpdateCoordinator[Any], **_P, _R](
@ -57,7 +56,7 @@ def async_re_login_on_expired[_T: SynologyDSMUpdateCoordinator[Any], **_P, _R](
return _async_wrap
class SynologyDSMUpdateCoordinator(DataUpdateCoordinator[_DataT]):
class SynologyDSMUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
"""DataUpdateCoordinator base class for synology_dsm."""
def __init__(

View File

@ -3,7 +3,7 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, TypeVar
from typing import Any
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import EntityDescription
@ -16,8 +16,6 @@ from .coordinator import (
SynologyDSMUpdateCoordinator,
)
_CoordinatorT = TypeVar("_CoordinatorT", bound=SynologyDSMUpdateCoordinator[Any])
@dataclass(frozen=True, kw_only=True)
class SynologyDSMEntityDescription(EntityDescription):
@ -26,7 +24,9 @@ class SynologyDSMEntityDescription(EntityDescription):
api_key: str
class SynologyDSMBaseEntity(CoordinatorEntity[_CoordinatorT]):
class SynologyDSMBaseEntity[_CoordinatorT: SynologyDSMUpdateCoordinator[Any]](
CoordinatorEntity[_CoordinatorT]
):
"""Representation of a Synology NAS entry."""
entity_description: SynologyDSMEntityDescription

View File

@ -3,7 +3,6 @@
import asyncio
from datetime import timedelta
import logging
from typing import Generic, TypeVar
from tplink_omada_client import OmadaSiteClient
from tplink_omada_client.exceptions import OmadaClientException
@ -13,10 +12,8 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
_LOGGER = logging.getLogger(__name__)
T = TypeVar("T")
class OmadaCoordinator(DataUpdateCoordinator[dict[str, T]], Generic[T]):
class OmadaCoordinator[_T](DataUpdateCoordinator[dict[str, _T]]):
"""Coordinator for synchronizing bulk Omada data."""
def __init__(
@ -35,7 +32,7 @@ class OmadaCoordinator(DataUpdateCoordinator[dict[str, T]], Generic[T]):
)
self.omada_client = omada_client
async def _async_update_data(self) -> dict[str, T]:
async def _async_update_data(self) -> dict[str, _T]:
"""Fetch data from API endpoint."""
try:
async with asyncio.timeout(10):
@ -43,6 +40,6 @@ class OmadaCoordinator(DataUpdateCoordinator[dict[str, T]], Generic[T]):
except OmadaClientException as err:
raise UpdateFailed(f"Error communicating with API: {err}") from err
async def poll_update(self) -> dict[str, T]:
async def poll_update(self) -> dict[str, _T]:
"""Poll the current data from the controller."""
raise NotImplementedError("Update method not implemented")

View File

@ -1,6 +1,6 @@
"""Base entity definitions."""
from typing import Any, Generic, TypeVar
from typing import Any
from tplink_omada_client.devices import OmadaDevice
@ -11,13 +11,11 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import OmadaCoordinator
T = TypeVar("T", bound="OmadaCoordinator[Any]")
class OmadaDeviceEntity(CoordinatorEntity[T], Generic[T]):
class OmadaDeviceEntity[_T: OmadaCoordinator[Any]](CoordinatorEntity[_T]):
"""Common base class for all entities associated with Omada SDN Devices."""
def __init__(self, coordinator: T, device: OmadaDevice) -> None:
def __init__(self, coordinator: _T, device: OmadaDevice) -> None:
"""Initialize the device."""
super().__init__(coordinator)
self.device = device

View File

@ -6,7 +6,7 @@ import asyncio
from collections import defaultdict
from collections.abc import Awaitable
import logging
from typing import Any, Generic, TypeVar
from typing import Any
import pyvera as veraApi
from requests.exceptions import RequestException
@ -207,10 +207,7 @@ def map_vera_device(
)
_DeviceTypeT = TypeVar("_DeviceTypeT", bound=veraApi.VeraDevice)
class VeraDevice(Generic[_DeviceTypeT], Entity):
class VeraDevice[_DeviceTypeT: veraApi.VeraDevice](Entity):
"""Representation of a Vera device entity."""
def __init__(

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from abc import abstractmethod
from datetime import date, datetime, timedelta
from typing import TYPE_CHECKING, TypeVar
from typing import TYPE_CHECKING
from aiowithings import (
Activity,
@ -30,12 +30,10 @@ from .const import LOGGER
if TYPE_CHECKING:
from . import WithingsConfigEntry
_T = TypeVar("_T")
UPDATE_INTERVAL = timedelta(minutes=10)
class WithingsDataUpdateCoordinator(DataUpdateCoordinator[_T]):
class WithingsDataUpdateCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
"""Base coordinator."""
config_entry: WithingsConfigEntry
@ -75,14 +73,14 @@ class WithingsDataUpdateCoordinator(DataUpdateCoordinator[_T]):
)
await self.async_request_refresh()
async def _async_update_data(self) -> _T:
async def _async_update_data(self) -> _DataT:
try:
return await self._internal_update_data()
except (WithingsUnauthorizedError, WithingsAuthenticationFailedError) as exc:
raise ConfigEntryAuthFailed from exc
@abstractmethod
async def _internal_update_data(self) -> _T:
async def _internal_update_data(self) -> _DataT:
"""Update coordinator data."""

View File

@ -2,7 +2,7 @@
from __future__ import annotations
from typing import TypeVar
from typing import Any
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
@ -10,10 +10,8 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import WithingsDataUpdateCoordinator
_T = TypeVar("_T", bound=WithingsDataUpdateCoordinator)
class WithingsEntity(CoordinatorEntity[_T]):
class WithingsEntity[_T: WithingsDataUpdateCoordinator[Any]](CoordinatorEntity[_T]):
"""Base class for withings entities."""
_attr_has_entity_name = True

View File

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from typing import Generic, TypeVar
from typing import Any
from aiowithings import (
Activity,
@ -767,11 +767,10 @@ async def async_setup_entry(
async_add_entities(entities)
_T = TypeVar("_T", bound=WithingsDataUpdateCoordinator)
_ED = TypeVar("_ED", bound=SensorEntityDescription)
class WithingsSensor(WithingsEntity[_T], SensorEntity, Generic[_T, _ED]):
class WithingsSensor[
_T: WithingsDataUpdateCoordinator[Any],
_ED: SensorEntityDescription,
](WithingsEntity[_T], SensorEntity):
"""Implementation of a Withings sensor."""
entity_description: _ED

View File

@ -2,7 +2,7 @@
from collections.abc import Callable, Coroutine
from logging import Logger
from typing import Any, TypeVar
from typing import Any
from xiaomi_ble import SensorUpdate, XiaomiBluetoothDeviceData
@ -22,8 +22,6 @@ from homeassistant.helpers.debounce import Debouncer
from .const import CONF_SLEEPY_DEVICE
_T = TypeVar("_T")
class XiaomiActiveBluetoothProcessorCoordinator(
ActiveBluetoothProcessorCoordinator[SensorUpdate]
@ -72,7 +70,7 @@ class XiaomiActiveBluetoothProcessorCoordinator(
return self.entry.data.get(CONF_SLEEPY_DEVICE, self.device_data.sleepy_device)
class XiaomiPassiveBluetoothDataProcessor(
class XiaomiPassiveBluetoothDataProcessor[_T](
PassiveBluetoothDataProcessor[_T, SensorUpdate]
):
"""Define a Xiaomi Bluetooth Passive Update Data Processor."""

View File

@ -4,7 +4,7 @@ import datetime
from enum import Enum
from functools import partial
import logging
from typing import Any, TypeVar
from typing import Any
from construct.core import ChecksumError
from miio import Device, DeviceException
@ -22,8 +22,6 @@ from .const import DOMAIN, AuthException, SetupException
_LOGGER = logging.getLogger(__name__)
_T = TypeVar("_T", bound=DataUpdateCoordinator[Any])
class ConnectXiaomiDevice:
"""Class to async connect to a Xiaomi Device."""
@ -109,7 +107,9 @@ class XiaomiMiioEntity(Entity):
return device_info
class XiaomiCoordinatedMiioEntity(CoordinatorEntity[_T]):
class XiaomiCoordinatedMiioEntity[_T: DataUpdateCoordinator[Any]](
CoordinatorEntity[_T]
):
"""Representation of a base a coordinated Xiaomi Miio Entity."""
_attr_has_entity_name = True