Add strict typing for lidarr (#79241)

This commit is contained in:
Marc Mueller 2022-10-07 20:54:29 +02:00 committed by GitHub
parent a809f645a7
commit 14d2bbfcd6
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
5 changed files with 53 additions and 37 deletions

View file

@ -162,6 +162,7 @@ homeassistant.components.lacrosse_view.*
homeassistant.components.lametric.* homeassistant.components.lametric.*
homeassistant.components.laundrify.* homeassistant.components.laundrify.*
homeassistant.components.lcn.* homeassistant.components.lcn.*
homeassistant.components.lidarr.*
homeassistant.components.lifx.* homeassistant.components.lifx.*
homeassistant.components.light.* homeassistant.components.light.*
homeassistant.components.litterrobot.* homeassistant.components.litterrobot.*

View file

@ -1,6 +1,8 @@
"""The Lidarr component.""" """The Lidarr component."""
from __future__ import annotations from __future__ import annotations
from typing import Any
from aiopyarr.lidarr_client import LidarrClient from aiopyarr.lidarr_client import LidarrClient
from aiopyarr.models.host_configuration import PyArrHostConfiguration from aiopyarr.models.host_configuration import PyArrHostConfiguration
@ -18,6 +20,7 @@ from .coordinator import (
LidarrDataUpdateCoordinator, LidarrDataUpdateCoordinator,
QueueDataUpdateCoordinator, QueueDataUpdateCoordinator,
StatusDataUpdateCoordinator, StatusDataUpdateCoordinator,
T,
WantedDataUpdateCoordinator, WantedDataUpdateCoordinator,
) )
@ -36,7 +39,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
session=async_get_clientsession(hass, host_configuration.verify_ssl), session=async_get_clientsession(hass, host_configuration.verify_ssl),
request_timeout=60, request_timeout=60,
) )
coordinators: dict[str, LidarrDataUpdateCoordinator] = { coordinators: dict[str, LidarrDataUpdateCoordinator[Any]] = {
"disk_space": DiskSpaceDataUpdateCoordinator(hass, host_configuration, lidarr), "disk_space": DiskSpaceDataUpdateCoordinator(hass, host_configuration, lidarr),
"queue": QueueDataUpdateCoordinator(hass, host_configuration, lidarr), "queue": QueueDataUpdateCoordinator(hass, host_configuration, lidarr),
"status": StatusDataUpdateCoordinator(hass, host_configuration, lidarr), "status": StatusDataUpdateCoordinator(hass, host_configuration, lidarr),
@ -63,13 +66,15 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return unload_ok return unload_ok
class LidarrEntity(CoordinatorEntity[LidarrDataUpdateCoordinator]): class LidarrEntity(CoordinatorEntity[LidarrDataUpdateCoordinator[T]]):
"""Defines a base Lidarr entity.""" """Defines a base Lidarr entity."""
_attr_has_entity_name = True _attr_has_entity_name = True
def __init__( def __init__(
self, coordinator: LidarrDataUpdateCoordinator, description: EntityDescription self,
coordinator: LidarrDataUpdateCoordinator[T],
description: EntityDescription,
) -> None: ) -> None:
"""Initialize the Lidarr entity.""" """Initialize the Lidarr entity."""
super().__init__(coordinator) super().__init__(coordinator)

View file

@ -3,7 +3,7 @@ from __future__ import annotations
from abc import abstractmethod from abc import abstractmethod
from datetime import timedelta from datetime import timedelta
from typing import Generic, TypeVar, cast from typing import Generic, TypeVar, Union, cast
from aiopyarr import LidarrAlbum, LidarrQueue, LidarrRootFolder, exceptions from aiopyarr import LidarrAlbum, LidarrQueue, LidarrRootFolder, exceptions
from aiopyarr.lidarr_client import LidarrClient from aiopyarr.lidarr_client import LidarrClient
@ -16,10 +16,10 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, Upda
from .const import DEFAULT_MAX_RECORDS, DOMAIN, LOGGER from .const import DEFAULT_MAX_RECORDS, DOMAIN, LOGGER
T = TypeVar("T", list[LidarrRootFolder], LidarrQueue, str, LidarrAlbum) T = TypeVar("T", bound=Union[list[LidarrRootFolder], LidarrQueue, str, LidarrAlbum])
class LidarrDataUpdateCoordinator(DataUpdateCoordinator, Generic[T]): class LidarrDataUpdateCoordinator(DataUpdateCoordinator[T], Generic[T]):
"""Data update coordinator for the Lidarr integration.""" """Data update coordinator for the Lidarr integration."""
config_entry: ConfigEntry config_entry: ConfigEntry
@ -59,15 +59,19 @@ class LidarrDataUpdateCoordinator(DataUpdateCoordinator, Generic[T]):
raise NotImplementedError raise NotImplementedError
class DiskSpaceDataUpdateCoordinator(LidarrDataUpdateCoordinator): class DiskSpaceDataUpdateCoordinator(
LidarrDataUpdateCoordinator[list[LidarrRootFolder]]
):
"""Disk space update coordinator for Lidarr.""" """Disk space update coordinator for Lidarr."""
async def _fetch_data(self) -> list[LidarrRootFolder]: async def _fetch_data(self) -> list[LidarrRootFolder]:
"""Fetch the data.""" """Fetch the data."""
return cast(list, await self.api_client.async_get_root_folders()) return cast(
list[LidarrRootFolder], await self.api_client.async_get_root_folders()
)
class QueueDataUpdateCoordinator(LidarrDataUpdateCoordinator): class QueueDataUpdateCoordinator(LidarrDataUpdateCoordinator[LidarrQueue]):
"""Queue update coordinator.""" """Queue update coordinator."""
async def _fetch_data(self) -> LidarrQueue: async def _fetch_data(self) -> LidarrQueue:
@ -75,7 +79,7 @@ class QueueDataUpdateCoordinator(LidarrDataUpdateCoordinator):
return await self.api_client.async_get_queue(page_size=DEFAULT_MAX_RECORDS) return await self.api_client.async_get_queue(page_size=DEFAULT_MAX_RECORDS)
class StatusDataUpdateCoordinator(LidarrDataUpdateCoordinator): class StatusDataUpdateCoordinator(LidarrDataUpdateCoordinator[str]):
"""Status update coordinator for Lidarr.""" """Status update coordinator for Lidarr."""
async def _fetch_data(self) -> str: async def _fetch_data(self) -> str:
@ -83,7 +87,7 @@ class StatusDataUpdateCoordinator(LidarrDataUpdateCoordinator):
return (await self.api_client.async_get_system_status()).version return (await self.api_client.async_get_system_status()).version
class WantedDataUpdateCoordinator(LidarrDataUpdateCoordinator): class WantedDataUpdateCoordinator(LidarrDataUpdateCoordinator[LidarrAlbum]):
"""Wanted update coordinator.""" """Wanted update coordinator."""
async def _fetch_data(self) -> LidarrAlbum: async def _fetch_data(self) -> LidarrAlbum:

View file

@ -4,10 +4,9 @@ from __future__ import annotations
from collections.abc import Callable from collections.abc import Callable
from copy import deepcopy from copy import deepcopy
from dataclasses import dataclass from dataclasses import dataclass
from datetime import datetime from typing import Any, Generic
from typing import Generic
from aiopyarr import LidarrQueueItem, LidarrRootFolder from aiopyarr import LidarrQueue, LidarrQueueItem, LidarrRootFolder
from homeassistant.components.sensor import ( from homeassistant.components.sensor import (
SensorEntity, SensorEntity,
@ -18,7 +17,6 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.const import DATA_GIGABYTES from homeassistant.const import DATA_GIGABYTES
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import StateType
from . import LidarrEntity from . import LidarrEntity
from .const import BYTE_SIZES, DOMAIN from .const import BYTE_SIZES, DOMAIN
@ -27,7 +25,7 @@ from .coordinator import LidarrDataUpdateCoordinator, T
def get_space(data: list[LidarrRootFolder], name: str) -> str: def get_space(data: list[LidarrRootFolder], name: str) -> str:
"""Get space.""" """Get space."""
space = [] space: list[float] = []
for mount in data: for mount in data:
if name in mount.path: if name in mount.path:
mount.freeSpace = mount.freeSpace if mount.accessible else 0 mount.freeSpace = mount.freeSpace if mount.accessible else 0
@ -36,8 +34,8 @@ def get_space(data: list[LidarrRootFolder], name: str) -> str:
def get_modified_description( def get_modified_description(
description: LidarrSensorEntityDescription, mount: LidarrRootFolder description: LidarrSensorEntityDescription[T], mount: LidarrRootFolder
) -> tuple[LidarrSensorEntityDescription, str]: ) -> tuple[LidarrSensorEntityDescription[T], str]:
"""Return modified description and folder name.""" """Return modified description and folder name."""
desc = deepcopy(description) desc = deepcopy(description)
name = mount.path.rsplit("/")[-1].rsplit("\\")[-1] name = mount.path.rsplit("/")[-1].rsplit("\\")[-1]
@ -50,25 +48,23 @@ def get_modified_description(
class LidarrSensorEntityDescriptionMixIn(Generic[T]): class LidarrSensorEntityDescriptionMixIn(Generic[T]):
"""Mixin for required keys.""" """Mixin for required keys."""
value_fn: Callable[[T, str], str] value_fn: Callable[[T, str], str | int]
@dataclass @dataclass
class LidarrSensorEntityDescription( class LidarrSensorEntityDescription(
SensorEntityDescription, LidarrSensorEntityDescriptionMixIn, Generic[T] SensorEntityDescription, LidarrSensorEntityDescriptionMixIn[T], Generic[T]
): ):
"""Class to describe a Lidarr sensor.""" """Class to describe a Lidarr sensor."""
attributes_fn: Callable[ attributes_fn: Callable[[T], dict[str, str] | None] = lambda _: None
[T], dict[str, StateType | datetime] | None
] = lambda _: None
description_fn: Callable[ description_fn: Callable[
[LidarrSensorEntityDescription, LidarrRootFolder], [LidarrSensorEntityDescription[T], LidarrRootFolder],
tuple[LidarrSensorEntityDescription, str] | None, tuple[LidarrSensorEntityDescription[T], str] | None,
] = lambda _, __: None ] | None = None
SENSOR_TYPES: dict[str, LidarrSensorEntityDescription] = { SENSOR_TYPES: dict[str, LidarrSensorEntityDescription[Any]] = {
"disk_space": LidarrSensorEntityDescription( "disk_space": LidarrSensorEntityDescription(
key="disk_space", key="disk_space",
name="Disk space", name="Disk space",
@ -78,7 +74,7 @@ SENSOR_TYPES: dict[str, LidarrSensorEntityDescription] = {
state_class=SensorStateClass.TOTAL, state_class=SensorStateClass.TOTAL,
description_fn=get_modified_description, description_fn=get_modified_description,
), ),
"queue": LidarrSensorEntityDescription( "queue": LidarrSensorEntityDescription[LidarrQueue](
key="queue", key="queue",
name="Queue", name="Queue",
native_unit_of_measurement="Albums", native_unit_of_measurement="Albums",
@ -87,7 +83,7 @@ SENSOR_TYPES: dict[str, LidarrSensorEntityDescription] = {
state_class=SensorStateClass.TOTAL, state_class=SensorStateClass.TOTAL,
attributes_fn=lambda data: {i.title: queue_str(i) for i in data.records}, attributes_fn=lambda data: {i.title: queue_str(i) for i in data.records},
), ),
"wanted": LidarrSensorEntityDescription( "wanted": LidarrSensorEntityDescription[LidarrQueue](
key="wanted", key="wanted",
name="Wanted", name="Wanted",
native_unit_of_measurement="Albums", native_unit_of_measurement="Albums",
@ -108,10 +104,10 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback, async_add_entities: AddEntitiesCallback,
) -> None: ) -> None:
"""Set up Lidarr sensors based on a config entry.""" """Set up Lidarr sensors based on a config entry."""
coordinators: dict[str, LidarrDataUpdateCoordinator] = hass.data[DOMAIN][ coordinators: dict[str, LidarrDataUpdateCoordinator[Any]] = hass.data[DOMAIN][
entry.entry_id entry.entry_id
] ]
entities = [] entities: list[LidarrSensor[Any]] = []
for coordinator_type, description in SENSOR_TYPES.items(): for coordinator_type, description in SENSOR_TYPES.items():
coordinator = coordinators[coordinator_type] coordinator = coordinators[coordinator_type]
if coordinator_type != "disk_space": if coordinator_type != "disk_space":
@ -125,15 +121,15 @@ async def async_setup_entry(
async_add_entities(entities) async_add_entities(entities)
class LidarrSensor(LidarrEntity, SensorEntity): class LidarrSensor(LidarrEntity[T], SensorEntity):
"""Implementation of the Lidarr sensor.""" """Implementation of the Lidarr sensor."""
entity_description: LidarrSensorEntityDescription entity_description: LidarrSensorEntityDescription[T]
def __init__( def __init__(
self, self,
coordinator: LidarrDataUpdateCoordinator, coordinator: LidarrDataUpdateCoordinator[T],
description: LidarrSensorEntityDescription, description: LidarrSensorEntityDescription[T],
folder_name: str = "", folder_name: str = "",
) -> None: ) -> None:
"""Create Lidarr entity.""" """Create Lidarr entity."""
@ -141,12 +137,12 @@ class LidarrSensor(LidarrEntity, SensorEntity):
self.folder_name = folder_name self.folder_name = folder_name
@property @property
def extra_state_attributes(self) -> dict[str, StateType | datetime] | None: def extra_state_attributes(self) -> dict[str, str] | None:
"""Return the state attributes of the sensor.""" """Return the state attributes of the sensor."""
return self.entity_description.attributes_fn(self.coordinator.data) return self.entity_description.attributes_fn(self.coordinator.data)
@property @property
def native_value(self) -> StateType: def native_value(self) -> str | int:
"""Return the state of the sensor.""" """Return the state of the sensor."""
return self.entity_description.value_fn(self.coordinator.data, self.folder_name) return self.entity_description.value_fn(self.coordinator.data, self.folder_name)

View file

@ -1372,6 +1372,16 @@ disallow_untyped_defs = true
warn_return_any = true warn_return_any = true
warn_unreachable = true warn_unreachable = true
[mypy-homeassistant.components.lidarr.*]
check_untyped_defs = true
disallow_incomplete_defs = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_decorators = true
disallow_untyped_defs = true
warn_return_any = true
warn_unreachable = true
[mypy-homeassistant.components.lifx.*] [mypy-homeassistant.components.lifx.*]
check_untyped_defs = true check_untyped_defs = true
disallow_incomplete_defs = true disallow_incomplete_defs = true