Use runtime_data in youtube (#168696)

This commit is contained in:
epenet
2026-04-21 13:04:49 +02:00
committed by GitHub
parent 12af4bd0f4
commit b1fcc0ebde
6 changed files with 27 additions and 41 deletions
+9 -14
View File
@@ -4,7 +4,6 @@ from __future__ import annotations
from aiohttp.client_exceptions import ClientError, ClientResponseError
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
@@ -16,13 +15,13 @@ from homeassistant.helpers.config_entry_oauth2_flow import (
)
from .api import AsyncConfigEntryAuth
from .const import AUTH, COORDINATOR, DOMAIN
from .coordinator import YouTubeDataUpdateCoordinator
from .const import DOMAIN
from .coordinator import YouTubeConfigEntry, YouTubeDataUpdateCoordinator
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: YouTubeConfigEntry) -> bool:
"""Set up YouTube from a config entry."""
try:
implementation = await async_get_config_entry_implementation(hass, entry)
@@ -49,25 +48,21 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
await delete_devices(hass, entry, coordinator)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
COORDINATOR: coordinator,
AUTH: auth,
}
entry.runtime_data = coordinator
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: YouTubeConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
async def delete_devices(
hass: HomeAssistant, entry: ConfigEntry, coordinator: YouTubeDataUpdateCoordinator
hass: HomeAssistant,
entry: YouTubeConfigEntry,
coordinator: YouTubeDataUpdateCoordinator,
) -> None:
"""Delete all devices created by integration."""
channel_ids = list(coordinator.data)
@@ -10,12 +10,7 @@ import voluptuous as vol
from youtubeaio.types import AuthScope, ForbiddenError
from youtubeaio.youtube import YouTube
from homeassistant.config_entries import (
SOURCE_REAUTH,
ConfigEntry,
ConfigFlowResult,
OptionsFlow,
)
from homeassistant.config_entries import SOURCE_REAUTH, ConfigFlowResult, OptionsFlow
from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN
from homeassistant.core import callback
from homeassistant.helpers import config_entry_oauth2_flow
@@ -33,6 +28,7 @@ from .const import (
DOMAIN,
LOGGER,
)
from .coordinator import YouTubeConfigEntry
class OAuth2FlowHandler(
@@ -50,7 +46,7 @@ class OAuth2FlowHandler(
@staticmethod
@callback
def async_get_options_flow(
config_entry: ConfigEntry,
config_entry: YouTubeConfigEntry,
) -> YouTubeOptionsFlowHandler:
"""Get the options flow for this handler."""
return YouTubeOptionsFlowHandler()
@@ -9,8 +9,6 @@ CHANNEL_CREATION_HELP_URL = "https://support.google.com/youtube/answer/1646861"
CONF_CHANNELS = "channels"
CONF_UPLOAD_PLAYLIST = "upload_playlist_id"
COORDINATOR = "coordinator"
AUTH = "auth"
LOGGER = logging.getLogger(__package__)
@@ -14,7 +14,7 @@ from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from . import AsyncConfigEntryAuth
from .api import AsyncConfigEntryAuth
from .const import (
ATTR_DESCRIPTION,
ATTR_LATEST_VIDEO,
@@ -29,14 +29,19 @@ from .const import (
LOGGER,
)
type YouTubeConfigEntry = ConfigEntry[YouTubeDataUpdateCoordinator]
class YouTubeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""A YouTube Data Update Coordinator."""
config_entry: ConfigEntry
config_entry: YouTubeConfigEntry
def __init__(
self, hass: HomeAssistant, config_entry: ConfigEntry, auth: AsyncConfigEntryAuth
self,
hass: HomeAssistant,
config_entry: YouTubeConfigEntry,
auth: AsyncConfigEntryAuth,
) -> None:
"""Initialize the YouTube data coordinator."""
self._auth = auth
@@ -4,20 +4,17 @@ from __future__ import annotations
from typing import Any
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import ATTR_DESCRIPTION, ATTR_LATEST_VIDEO, COORDINATOR, DOMAIN
from .coordinator import YouTubeDataUpdateCoordinator
from .const import ATTR_DESCRIPTION, ATTR_LATEST_VIDEO
from .coordinator import YouTubeConfigEntry
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, entry: ConfigEntry
hass: HomeAssistant, entry: YouTubeConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
coordinator: YouTubeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
COORDINATOR
]
coordinator = entry.runtime_data
sensor_data: dict[str, Any] = {}
for channel_id, channel_data in coordinator.data.items():
channel_data.get(ATTR_LATEST_VIDEO, {}).pop(ATTR_DESCRIPTION)
+3 -8
View File
@@ -11,13 +11,11 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ATTR_ICON
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import StateType
from . import YouTubeDataUpdateCoordinator
from .const import (
ATTR_LATEST_VIDEO,
ATTR_PUBLISHED_AT,
@@ -26,9 +24,8 @@ from .const import (
ATTR_TITLE,
ATTR_TOTAL_VIEWS,
ATTR_VIDEO_ID,
COORDINATOR,
DOMAIN,
)
from .coordinator import YouTubeConfigEntry
from .entity import YouTubeChannelEntity
@@ -79,13 +76,11 @@ SENSOR_TYPES = [
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: YouTubeConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up the YouTube sensor."""
coordinator: YouTubeDataUpdateCoordinator = hass.data[DOMAIN][entry.entry_id][
COORDINATOR
]
coordinator = entry.runtime_data
async_add_entities(
YouTubeSensor(coordinator, sensor_type, channel_id)
for channel_id in coordinator.data