Add support for fetching channels (#51)

Author: joostlek

src/youtubeaio/youtube.py

@@ -12,7 +12,7 @@
     build_url,
     first,
 )
-from youtubeaio.models import YouTubeVideo
+from youtubeaio.models import YouTubeChannel, YouTubeVideo
 from youtubeaio.types import (
     AuthScope,
     MissingScopeError,
@@ -191,14 +191,47 @@ async def get_videos(
             "videos",
             param,
             YouTubeVideo,
-            split_lists=True,
         ):
             yield item  # type: ignore[misc]
 
     async def get_video(self, video_id: str) -> YouTubeVideo | None:
         """Get a single video."""
         return await first(self.get_videos([video_id]))
 
+    async def _get_channels(
+        self,
+        param: dict[str, Any],
+    ) -> AsyncGenerator[YouTubeChannel, None]:
+        """Get channels."""
+        async for item in self._build_generator(
+            "GET",
+            "channels",
+            param,
+            YouTubeChannel,
+        ):
+            yield item  # type: ignore[misc]
+
+    async def get_user_channels(self) -> AsyncGenerator[YouTubeChannel, None]:
+        """Return channels owned by the authenticated user."""
+        param = {
+            "part": "snippet",
+            "mine": "true",
+        }
+        async for item in self._get_channels(param):
+            yield item
+
+    async def get_channels(
+        self,
+        channel_ids: list[str],
+    ) -> AsyncGenerator[YouTubeChannel, None]:
+        """Return list of channels."""
+        param = {
+            "part": "snippet",
+            "id": channel_ids,
+        }
+        async for item in self._get_channels(param):
+            yield item
+
     async def close(self) -> None:
         """Close open client session."""
         if self.session and self._close_session:

tests/const.py

@@ -0,0 +1,3 @@
+"""Constants for YouTube tests."""
+
+YOUTUBE_URL = "youtube.googleapis.com"

tests/fixtures/channel_response_snippet.json

@@ -0,0 +1,43 @@
+{
+  "kind": "youtube#channelListResponse",
+  "etag": "sM5ffk5W79swi2V870w6CmMlPgs",
+  "pageInfo": {
+    "totalResults": 1,
+    "resultsPerPage": 5
+  },
+  "items": [
+    {
+      "kind": "youtube#channel",
+      "etag": "ieb4dig-VKDKW3FEut81IwUEiog",
+      "id": "UC_x5XG1OV2P6uZZ5FSM9Ttw",
+      "snippet": {
+        "title": "Google for Developers",
+        "description": "Subscribe to join a community of creative developers and learn the latest in Google technology — from AI and cloud, to mobile and web.\n\nExplore more at developers.google.com\n\n",
+        "customUrl": "@googledevelopers",
+        "publishedAt": "2007-08-23T00:34:43Z",
+        "thumbnails": {
+          "default": {
+            "url": "https://yt3.ggpht.com/fca_HuJ99xUxflWdex0XViC3NfctBFreIl8y4i9z411asnGTWY-Ql3MeH_ybA4kNaOjY7kyA=s88-c-k-c0x00ffffff-no-rj",
+            "width": 88,
+            "height": 88
+          },
+          "medium": {
+            "url": "https://yt3.ggpht.com/fca_HuJ99xUxflWdex0XViC3NfctBFreIl8y4i9z411asnGTWY-Ql3MeH_ybA4kNaOjY7kyA=s240-c-k-c0x00ffffff-no-rj",
+            "width": 240,
+            "height": 240
+          },
+          "high": {
+            "url": "https://yt3.ggpht.com/fca_HuJ99xUxflWdex0XViC3NfctBFreIl8y4i9z411asnGTWY-Ql3MeH_ybA4kNaOjY7kyA=s800-c-k-c0x00ffffff-no-rj",
+            "width": 800,
+            "height": 800
+          }
+        },
+        "localized": {
+          "title": "Google for Developers",
+          "description": "Subscribe to join a community of creative developers and learn the latest in Google technology — from AI and cloud, to mobile and web.\n\nExplore more at developers.google.com\n\n"
+        },
+        "country": "US"
+      }
+    }
+  ]
+}

tests/test_channel.py

@@ -0,0 +1,82 @@
+"""Tests for the YouTube client."""
+from datetime import datetime, timezone
+
+import aiohttp
+import pytest
+from aresponses import ResponsesMockServer
+
+from youtubeaio.youtube import YouTube
+
+from . import load_fixture
+from .const import YOUTUBE_URL
+
+
+async def test_fetch_channel(
+    aresponses: ResponsesMockServer,
+) -> None:
+    """Test retrieving a channel."""
+    aresponses.add(
+        YOUTUBE_URL,
+        "/youtube/v3/channels",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text=load_fixture("channel_response_snippet.json"),
+        ),
+    )
+    async with aiohttp.ClientSession() as session:
+        youtube = YouTube(session=session)
+        channel_generator = youtube.get_channels(
+            channel_ids=["UC_x5XG1OV2P6uZZ5FSM9Ttw"],
+        )
+        channel = await channel_generator.__anext__()
+        assert channel
+        assert channel.snippet
+        assert channel.snippet.published_at == datetime(
+            2007,
+            8,
+            23,
+            0,
+            34,
+            43,
+            tzinfo=timezone.utc,
+        )
+        with pytest.raises(StopAsyncIteration):
+            await channel_generator.__anext__()
+        await youtube.close()
+
+
+async def test_fetch_own_channel(
+    aresponses: ResponsesMockServer,
+) -> None:
+    """Test retrieving own channel."""
+    aresponses.add(
+        YOUTUBE_URL,
+        "/youtube/v3/channels?part=snippet&mine=true",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text=load_fixture("channel_response_snippet.json"),
+        ),
+        match_querystring=True,
+    )
+    async with aiohttp.ClientSession() as session:
+        youtube = YouTube(session=session)
+        channel_generator = youtube.get_user_channels()
+        channel = await channel_generator.__anext__()
+        assert channel
+        assert channel.snippet
+        assert channel.snippet.published_at == datetime(
+            2007,
+            8,
+            23,
+            0,
+            34,
+            43,
+            tzinfo=timezone.utc,
+        )
+        with pytest.raises(StopAsyncIteration):
+            await channel_generator.__anext__()
+        await youtube.close()

tests/test_video.py

@@ -10,8 +10,7 @@
 from youtubeaio.youtube import YouTube
 
 from . import load_fixture
-
-YOUTUBE_URL = "youtube.googleapis.com"
+from .const import YOUTUBE_URL
 
 
 async def test_fetch_video(

tests/test_youtube.py

@@ -15,8 +15,7 @@
 from youtubeaio.youtube import YouTube
 
 from . import load_fixture
-
-YOUTUBE_URL = "youtube.googleapis.com"
+from .const import YOUTUBE_URL
 
 
 async def test_new_session(