Add support for fetching subscriptions (#52)

Author: joostlek

src/youtubeaio/models.py

@@ -107,3 +107,24 @@ class YouTubeChannel(BaseModel):
         None,
         alias="contentDetails",
     )
+
+
+class YouTubeSubscriptionSnippet(BaseModel):
+    """Model representing a YouTube subscription snippet."""
+
+    title: str = Field(...)
+    description: str = Field(...)
+    subscribed_at: datetime = Field(..., alias="publishedAt")
+    channel_info: dict[str, str] = Field(..., alias="resourceId")
+
+    @property
+    def channel_id(self) -> str:
+        """Return channel id."""
+        return self.channel_info["channelId"]
+
+
+class YouTubeSubscription(BaseModel):
+    """Model representing a YouTube subscription."""
+
+    subscription_id: str = Field(..., alias="id")
+    snippet: YouTubeSubscriptionSnippet | None = None

src/youtubeaio/youtube.py

@@ -12,7 +12,7 @@
     build_url,
     first,
 )
-from youtubeaio.models import YouTubeChannel, YouTubeVideo
+from youtubeaio.models import YouTubeChannel, YouTubeSubscription, YouTubeVideo
 from youtubeaio.types import (
     AuthScope,
     MissingScopeError,
@@ -232,6 +232,22 @@ async def get_channels(
         async for item in self._get_channels(param):
             yield item
 
+    async def get_user_subscriptions(
+        self,
+    ) -> AsyncGenerator[YouTubeSubscription, None]:
+        """Get subscriptions for authenticated user."""
+        param = {
+            "part": "snippet",
+            "mine": "true",
+        }
+        async for item in self._build_generator(
+            "GET",
+            "subscriptions",
+            param,
+            YouTubeSubscription,
+        ):
+            yield item  # type: ignore[misc]
+
     async def close(self) -> None:
         """Close open client session."""
         if self.session and self._close_session:

tests/fixtures/subscription_response_snippet.json

@@ -0,0 +1,62 @@
+{
+  "kind": "youtube#SubscriptionListResponse",
+  "etag": "ToySCAR2vCoWXFlWF3I2pOR9odY",
+  "pageInfo": {
+    "totalResults": 547,
+    "resultsPerPage": 2
+  },
+  "items": [
+    {
+      "kind": "youtube#subscription",
+      "etag": "qQ8Upr97jeB0ksUo2lFQXxXtcis",
+      "id": "l6YW-siEBx2rtBlTJ_ip132nwdjtfEj_s_XrLrZyj3Y",
+      "snippet": {
+        "publishedAt": "2020-05-18T01:04:54.7417Z",
+        "title": "DougDoug",
+        "description": "I'm a pepper who teach how to videogame real good.\n\nI stream live on Twitch! https://www.twitch.tv/DougDoug\n\nFollow me on twitter! https://twitter.com/DougDougFood",
+        "resourceId": {
+          "kind": "youtube#channel",
+          "channelId": "UClyGlKOhDUooPJFy4v_mqPg"
+        },
+        "channelId": "UCJA_MqaDMFv6Dh-E8bpszIw",
+        "thumbnails": {
+          "default": {
+            "url": "https://yt3.ggpht.com/ytc/AOPolaQNqmde36_-rZt3loE_aZJ1iJKX936_5ebnJA2dVQ=s88-c-k-c0x00ffffff-no-rj"
+          },
+          "medium": {
+            "url": "https://yt3.ggpht.com/ytc/AOPolaQNqmde36_-rZt3loE_aZJ1iJKX936_5ebnJA2dVQ=s240-c-k-c0x00ffffff-no-rj"
+          },
+          "high": {
+            "url": "https://yt3.ggpht.com/ytc/AOPolaQNqmde36_-rZt3loE_aZJ1iJKX936_5ebnJA2dVQ=s800-c-k-c0x00ffffff-no-rj"
+          }
+        }
+      }
+    },
+    {
+      "kind": "youtube#subscription",
+      "etag": "3Ma9gfFVk8R7tC6js2OPirKjleg",
+      "id": "l6YW-siEBx0PYqtFJQsPPR3JU1r9dzJpv-pjRAZJ0iU",
+      "snippet": {
+        "publishedAt": "2023-01-04T20:35:54.095866Z",
+        "title": "Home Assistant",
+        "description": "Home Assistant is an open source home automation that puts local control and privacy first. Powered by a worldwide community of tinkerers and DIY enthusiasts. Perfect to run on a Raspberry Pi or a local server.",
+        "resourceId": {
+          "kind": "youtube#channel",
+          "channelId": "UCbX3YkedQunLt7EQAdVxh7w"
+        },
+        "channelId": "UCJA_MqaDMFv6Dh-E8bpszIw",
+        "thumbnails": {
+          "default": {
+            "url": "https://yt3.ggpht.com/ytc/AOPolaTRAmW34QPGjkXJ9QKJaSXXzbP3K_BlPY5IodnE0Q=s88-c-k-c0x00ffffff-no-rj"
+          },
+          "medium": {
+            "url": "https://yt3.ggpht.com/ytc/AOPolaTRAmW34QPGjkXJ9QKJaSXXzbP3K_BlPY5IodnE0Q=s240-c-k-c0x00ffffff-no-rj"
+          },
+          "high": {
+            "url": "https://yt3.ggpht.com/ytc/AOPolaTRAmW34QPGjkXJ9QKJaSXXzbP3K_BlPY5IodnE0Q=s800-c-k-c0x00ffffff-no-rj"
+          }
+        }
+      }
+    }
+  ]
+}

tests/test_subscription.py

@@ -0,0 +1,36 @@
+"""Tests for the YouTube client."""
+
+import aiohttp
+from aresponses import ResponsesMockServer
+
+from youtubeaio.youtube import YouTube
+
+from . import load_fixture
+from .const import YOUTUBE_URL
+
+
+async def test_fetch_user_subscriptions(
+    aresponses: ResponsesMockServer,
+) -> None:
+    """Test retrieving a video."""
+    aresponses.add(
+        YOUTUBE_URL,
+        "/youtube/v3/subscriptions?part=snippet&mine=true",
+        "GET",
+        aresponses.Response(
+            status=200,
+            headers={"Content-Type": "application/json"},
+            text=load_fixture("subscription_response_snippet.json"),
+        ),
+        match_querystring=True,
+    )
+    async with aiohttp.ClientSession() as session:
+        youtube = YouTube(session=session)
+        count = 0
+        async for subscription in youtube.get_user_subscriptions():
+            count += 1
+            assert subscription
+            assert subscription.snippet
+            assert subscription.snippet.channel_id
+        assert count == 2
+        await youtube.close()