starred/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2026-09-26 17:31:15 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix youtube stream playlist items and stop early to avoid paginating entire playlists (#181321)

Browse Source
This commit is contained in:
DevHugo
2026-09-14 21:37:04 +02:00
committed by GitHub
parent 14ab7767fb
commit 1aab95f9b5
3 changed files with 168 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
homeassistant/components/youtube/coordinator.py
+60 -46
View File
@@ -2,7 +2,7 @@
import asyncio
from datetime import timedelta
from typing import Any, override
from typing import Any, Final, override
from youtubeaio.types import UnauthorizedError, YouTubeBackendError
@@ -33,6 +33,11 @@ from .const import (
type YouTubeConfigEntry = ConfigEntry[YouTubeDataUpdateCoordinator]
# Twice youtubeaio's 10s per-call timeout: single slow calls are handled by
# the library's own timeout (backend error / non-Short fallback); this budget
# only aborts the serial accumulation of several slow checks per channel.
_SHORTS_DETECTION_TIMEOUT: Final = 20
def _build_video_dict(video: Any, is_short: bool) -> dict[str, Any]:
"""Build the video attribute dict shared by all video sensors."""
@@ -76,31 +81,42 @@ class YouTubeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
channel_ids = self.config_entry.options[CONF_CHANNELS]
try:
async for channel in youtube.get_channels(channel_ids):
# Fetch up to 10 recent videos to find a Short and a non-Short.
videos = [
v
async for v in youtube.get_playlist_items(
channel.upload_playlist_id, 10
)
]
LOGGER.debug(
"Fetched %d videos for channel %s", len(videos), channel.channel_id
)
is_short_flags = await self._get_is_short_flags(youtube, videos)
checked = 0
latest_video: dict[str, Any] | None = None
latest_short: dict[str, Any] | None = None
latest_video_non_short: dict[str, Any] | None = None
for video, is_short in zip(videos, is_short_flags, strict=False):
entry = _build_video_dict(video, is_short)
if latest_video is None:
latest_video = entry
if is_short and latest_short is None:
latest_short = entry
if not is_short and latest_video_non_short is None:
latest_video_non_short = entry
if latest_short is not None and latest_video_non_short is not None:
break
try:
async with asyncio.timeout(_SHORTS_DETECTION_TIMEOUT):
# Only examine the first page (10 items): paginating
# further burns API quota for no benefit.
async for video in youtube.get_playlist_items(
channel.upload_playlist_id, 10
):
checked += 1
is_short = await self._resolve_is_short(youtube, video)
entry = _build_video_dict(video, is_short)
if latest_video is None:
latest_video = entry
if is_short and latest_short is None:
latest_short = entry
if not is_short and latest_video_non_short is None:
latest_video_non_short = entry
if (
latest_short is not None
and latest_video_non_short is not None
):
break
if checked >= 10:
break
except TimeoutError:
LOGGER.warning(
"Timed out processing recent uploads for channel %s; "
"continuing with partial results",
channel.channel_id,
)
LOGGER.debug(
"Examined %d videos for channel %s", checked, channel.channel_id
)
res[channel.channel_id] = {
ATTR_ID: channel.channel_id,
@@ -119,27 +135,25 @@ class YouTubeDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
raise UpdateFailed("Couldn't connect to YouTube") from err
return res
async def _get_is_short_flags(self, youtube: Any, videos: list[Any]) -> list[bool]:
"""Return is_short flags for each video, using cache when available."""
uncached = [
v for v in videos if v.content_details.video_id not in self._is_short_cache
]
if uncached:
results = await asyncio.gather(
*[youtube.is_short(v.content_details.video_id) for v in uncached],
return_exceptions=True,
async def _resolve_is_short(self, youtube: Any, video: Any) -> bool:
"""Return whether a single video is a Short.
Uses the cache when available. Videos can stop being checked as soon
as both a Short and a non-Short have been found. On error the result
is treated as non-Short without caching so the next refresh can retry.
"""
video_id = video.content_details.video_id
if video_id in self._is_short_cache:
return self._is_short_cache[video_id]
try:
result = await youtube.is_short(video_id)
except Exception as exc: # noqa: BLE001
LOGGER.warning(
"Error determining if video %s is a Short; treating as non-Short: %s",
video_id,
exc,
)
for video, result in zip(uncached, results, strict=False):
if isinstance(result, Exception):
LOGGER.warning(
"Error determining if video %s is a Short; "
"treating as non-Short: %s",
video.content_details.video_id,
result,
)
# Don't cache on error — let the next refresh retry.
else:
self._is_short_cache[video.content_details.video_id] = bool(result)
return [
self._is_short_cache.get(v.content_details.video_id, False) for v in videos
]
return False
is_short = bool(result)
self._is_short_cache[video_id] = is_short
return is_short
tests/components/youtube/__init__.py
+23 -3
View File
@@ -1,5 +1,6 @@
"""Tests for the YouTube integration."""
import asyncio
from collections.abc import AsyncGenerator
from youtubeaio.models import YouTubeChannel, YouTubePlaylistItem, YouTubeSubscription
@@ -23,6 +24,7 @@ class MockYouTube:
playlist_items_fixture: str = "get_playlist_items.json",
subscriptions_fixture: str = "get_subscriptions.json",
short_video_ids: set[str] | None = None,
short_check_delay: float = 0.0,
) -> None:
"""Initialize mock service."""
self.hass = hass
@@ -30,6 +32,9 @@ class MockYouTube:
self._playlist_items_fixture = playlist_items_fixture
self._subscriptions_fixture = subscriptions_fixture
self._short_video_ids: set[str] = short_video_ids or set()
self._short_check_delay = short_check_delay
self.playlist_item_requests = 0
self.playlist_items_yielded = 0
async def set_user_authentication(
self, token: str, scopes: list[AuthScope]
@@ -59,12 +64,26 @@ class MockYouTube:
async def get_playlist_items(
self, playlist_id: str, amount: int
) -> AsyncGenerator[YouTubePlaylistItem]:
"""Get channels."""
"""Get playlist items, paginating like the real API.
Cycles the fixture items forever in pages of `amount` (like a large
upload playlist), so tests can assert how far the coordinator
iterates before stopping.
"""
channels = await async_load_json_object_fixture(
self.hass, self._playlist_items_fixture, DOMAIN
)
for item in channels["items"]:
yield YouTubePlaylistItem(**item)
items = channels["items"]
if not items:
self.playlist_item_requests += 1
return
index = 0
while True:
self.playlist_item_requests += 1
for _ in range(amount):
self.playlist_items_yielded += 1
yield YouTubePlaylistItem(**items[index % len(items)])
index += 1
async def get_user_subscriptions(self) -> AsyncGenerator[YouTubeSubscription]:
"""Get channels for authenticated user."""
@@ -80,4 +99,5 @@ class MockYouTube:
async def is_short(self, video_id: str) -> bool:
"""Return whether the video is a Short."""
await asyncio.sleep(self._short_check_delay)
return video_id in self._short_video_ids
tests/components/youtube/test_sensor.py
+85 -9
View File
@@ -4,6 +4,7 @@ import asyncio
from datetime import timedelta
from unittest.mock import patch
import pytest
from syrupy.assertion import SnapshotAssertion
from youtubeaio.types import UnauthorizedError, YouTubeBackendError
@@ -77,19 +78,11 @@ async def test_sensor_with_short(
hass: HomeAssistant, snapshot: SnapshotAssertion, setup_integration: ComponentSetup
) -> None:
"""Test sensors when the channel has a Short upload."""
await setup_integration()
with patch(
"homeassistant.components.youtube.api.AsyncConfigEntryAuth.get_resource",
return_value=MockYouTube(hass, short_video_ids={"wysukDrMdqU"}),
):
# Clear the coordinator's is_short cache so the Short is re-detected.
entry = hass.config_entries.async_entries(DOMAIN)[0]
entry.runtime_data._is_short_cache.clear()
future = dt_util.utcnow() + timedelta(minutes=15)
async_fire_time_changed(hass, future)
await hass.async_block_till_done()
await asyncio.sleep(0.1)
await setup_integration()
state = hass.states.get("sensor.google_for_developers_latest_short")
assert state == snapshot
@@ -202,3 +195,86 @@ async def test_sensor_unavailable(
state = hass.states.get("sensor.google_for_developers_views")
assert state.state == "unavailable"
@pytest.mark.parametrize(
("short_video_ids", "expected_yielded", "expected_latest_video_state"),
[
pytest.param(
{"wysukDrMdqU"},
2,
"Google I/O 2023 Developer Keynote in 5 minutes",
id="stops_after_short_and_non_short_found",
),
pytest.param(
{"wysukDrMdqU", "hleLlcHwQLM", "lMKjtSFujcw", "c0mqBuXPrpA", "_n9xwuTORas"},
10,
"unavailable",
id="caps_at_page_size_when_only_shorts",
),
],
)
async def test_sensor_playlist_iteration(
hass: HomeAssistant,
setup_integration: ComponentSetup,
short_video_ids: set[str],
expected_yielded: int,
expected_latest_video_state: str,
) -> None:
"""Test playlist iteration stops as early as possible and never paginates."""
mock = MockYouTube(hass, short_video_ids=short_video_ids)
with patch(
"homeassistant.components.youtube.api.AsyncConfigEntryAuth.get_resource",
return_value=mock,
):
await setup_integration()
assert mock.playlist_item_requests == 1
assert mock.playlist_items_yielded == expected_yielded
state = hass.states.get("sensor.google_for_developers_latest_short")
assert state.state == "What's new in Google Home in less than 1 minute"
state = hass.states.get("sensor.google_for_developers_latest_video")
assert state.state == expected_latest_video_state
async def test_sensor_shorts_detection_timeout(
hass: HomeAssistant,
caplog: pytest.LogCaptureFixture,
setup_integration: ComponentSetup,
) -> None:
"""Test the per-channel pass is bounded when is_short hangs.
Channel-level sensors stay available while the video sensors report
unavailable from the partial results.
"""
mock = MockYouTube(hass, short_check_delay=1)
with (
patch(
"homeassistant.components.youtube.api.AsyncConfigEntryAuth.get_resource",
return_value=mock,
),
patch(
"homeassistant.components.youtube.coordinator._SHORTS_DETECTION_TIMEOUT",
0.1,
),
):
await setup_integration()
assert "Timed out processing recent uploads" in caplog.text
state = hass.states.get("sensor.google_for_developers_latest_upload")
assert state.state == "unavailable"
state = hass.states.get("sensor.google_for_developers_latest_short")
assert state.state == "unavailable"
state = hass.states.get("sensor.google_for_developers_latest_video")
assert state.state == "unavailable"
state = hass.states.get("sensor.google_for_developers_subscribers")
assert state.state == "2290000"
state = hass.states.get("sensor.google_for_developers_views")
assert state.state == "214141263"