feat(notification): ✨ add notification method

.github/workflows/docs.yaml

@@ -9,7 +9,7 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout code
-        uses: actions/checkout@v2
+        uses: actions/checkout@v4
 
       - name: Deploy docs
         uses: mhausenblas/mkdocs-deploy-gh-pages@master

.github/workflows/release.yaml

@@ -7,7 +7,7 @@ jobs:
   build:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Build and publish to pypi
       uses: JRubics/poetry-publish@v1.17
       with:

.github/workflows/test.yaml

@@ -8,12 +8,12 @@ on:
 
 jobs:
   test:
-    runs-on: ubuntu-20.04
+    runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ['3.6', '3.7', '3.8', '3.9', '3.10', '3.11', '3.12']
+        python-version: ['3.10', '3.11', '3.12', '3.13', '3.14']
         include:
-          - python-version: '3.8'
+          - python-version: '3.12'
             update-coverage: true
 
     steps:
@@ -23,7 +23,7 @@ jobs:
         with:
           python-version: ${{ matrix.python-version }}
       - name: Cache pip
-        uses: actions/cache@v2
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pip
           key: ${{ matrix.python-version }}-poetry-${{ hashFiles('pyproject.toml') }}
@@ -36,7 +36,7 @@ jobs:
           poetry run pytest
       - name: Upload coverage to Codecov
         if: ${{ matrix.update-coverage }}
-        uses: codecov/codecov-action@v4
+        uses: codecov/codecov-action@v5
         with:
           file: ./coverage.xml
           fail_ci_if_error: true
@@ -47,12 +47,12 @@ jobs:
     runs-on: ubuntu-latest
 
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v6
       - uses: actions/setup-python@v2
         with:
-          python-version: 3.8
+          python-version: 3.12
       - name: Cache pip
-        uses: actions/cache@v2
+        uses: actions/cache@v4
         with:
           path: ~/.cache/pip
           key: lintenv-v2

examples/clients/oauth_refreshing.py

@@ -12,28 +12,28 @@
 CLIENT_SECRET = "xxx"  # Your app secret
 CLIENT_SECRET_PATH = None  # or your path/to/client_secret_web.json
 
-TOKEN_PERSISTENT_PATH = None # path/to/persistent_token_storage_location
+TOKEN_PERSISTENT_PATH = None  # path/to/persistent_token_storage_location
 
 SCOPE = [
     "https://www.googleapis.com/auth/youtube",
     "https://www.googleapis.com/auth/youtube.force-ssl",
     "https://www.googleapis.com/auth/userinfo.profile",
 ]
 
+
 def do_refresh():
     token_location = Path(TOKEN_PERSISTENT_PATH)
 
     # Read the persistent token data if it exists
     token_data = {}
     if token_location.exists():
         token_data = loads(token_location.read_text())
-
 
     cli = Client(
         client_id=CLIENT_ID,
         client_secret=CLIENT_SECRET,
         access_token=token_data.get("access_token"),
-        refresh_token=token_data.get("refresh_token")
+        refresh_token=token_data.get("refresh_token"),
     )
     # or if you want to use a web type client_secret.json
     # cli = Client(
@@ -49,7 +49,9 @@ def do_refresh():
 
         response_uri = input("Input youtube redirect uri:\n")
 
-        token = cli.generate_access_token(authorization_response=response_uri, scope=SCOPE)
+        token = cli.generate_access_token(
+            authorization_response=response_uri, scope=SCOPE
+        )
         print(f"Your token: {token}")
 
     # Otherwise, refresh the access token if it has expired
@@ -65,16 +67,14 @@ def do_refresh():
     token_location.mkdir(parents=True, exist_ok=True)
     token_location.write_text(
         dumps(
-            {
-                "access_token": token.access_token,
-                "refresh_token": token.refresh_token
-            }
+            {"access_token": token.access_token, "refresh_token": token.refresh_token}
         )
     )
 
     # Now you can do things with the client
     resp = cli.channels.list(mine=True)
     print(f"Your channel id: {resp.items[0].id}")
 
+
 if __name__ == "__main__":
     do_refresh()

pyyoutube/client.py

@@ -33,6 +33,7 @@ class Client:
     AUTHORIZATION_URL = "https://accounts.google.com/o/oauth2/v2/auth"
     EXCHANGE_ACCESS_TOKEN_URL = "https://oauth2.googleapis.com/token"
     REVOKE_TOKEN_URL = "https://oauth2.googleapis.com/revoke"
+    HUB_URL = "https://pubsubhubbub.appspot.com/subscribe"
 
     DEFAULT_REDIRECT_URI = "https://localhost/"
     DEFAULT_SCOPE = [
@@ -520,3 +521,71 @@ def revoke_access_token(
         if response.ok:
             return True
         self.parse_response(response)
+
+    def subscribe_push_notification(
+        self,
+        channel_id: str,
+        callback_url: str,
+        mode: str = "subscribe",
+        lease_seconds: Optional[int] = None,
+        secret: Optional[str] = None,
+        verify: str = "async",
+    ) -> bool:
+        """Subscribe or unsubscribe to a YouTube channel's push notifications via PubSubHubbub.
+
+        When a subscribed channel publishes a new video or updates an existing one,
+        Google will send a notification to the callback_url.
+
+        Args:
+            channel_id:
+                The YouTube channel ID to subscribe to.
+            callback_url:
+                The URL that will receive push notifications from the hub.
+                Must be publicly accessible.
+            mode:
+                Either "subscribe" or "unsubscribe".
+            lease_seconds:
+                How long (in seconds) the subscription should remain active.
+                If omitted, the hub uses its own default (typically ~432000, i.e. 5 days).
+            secret:
+                A secret string used to compute an HMAC-SHA1 signature on each notification,
+                allowing you to verify the payload came from the hub.
+            verify:
+                Verification mode. Either "async" (default) or "sync".
+
+        Returns:
+            True if the hub accepted the request (HTTP 202 Accepted).
+
+        Raises:
+            PyYouTubeException: If the hub returns an error response.
+
+        References:
+            https://developers.google.com/youtube/v3/guides/push_notifications
+            https://pubsubhubbub.github.io/PubSubHubbub/pubsubhubbub-core-0.4.html
+        """
+        topic_url = (
+            f"https://www.youtube.com/xml/feeds/videos.xml?channel_id={channel_id}"
+        )
+
+        data = {
+            "hub.callback": callback_url,
+            "hub.mode": mode,
+            "hub.topic": topic_url,
+            "hub.verify": verify,
+        }
+        if lease_seconds is not None:
+            data["hub.lease_seconds"] = str(lease_seconds)
+        if secret is not None:
+            data["hub.secret"] = secret
+
+        response = self.request(
+            method="POST",
+            path=self.HUB_URL,
+            data=data,
+            enforce_auth=False,
+        )
+
+        # Hub returns 202 Accepted on success (async) or 204 No Content (sync)
+        if response.status_code in (202, 204):
+            return True
+        self.parse_response(response)

tests/clients/test_client.py

@@ -113,3 +113,68 @@ def test_oauth(self, helpers):
                     status=400,
                 )
                 cli.revoke_access_token(token="token")
+
+    def test_subscribe_push_notification(self):
+        HUB_URL = "https://pubsubhubbub.appspot.com/subscribe"
+        cli = Client(client_id="id", client_secret="secret")
+
+        # subscribe returns True on 202 Accepted
+        with responses.RequestsMock() as m:
+            m.add(method="POST", url=HUB_URL, status=202)
+            result = cli.subscribe_push_notification(
+                channel_id="UCxxxxxx",
+                callback_url="https://example.com/webhook",
+            )
+            assert result is True
+            # verify hub.mode and hub.topic were sent correctly
+            assert m.calls[0].request.body is not None
+            assert "hub.mode=subscribe" in m.calls[0].request.body
+            assert "UCxxxxxx" in m.calls[0].request.body
+
+        # unsubscribe returns True on 202 Accepted
+        with responses.RequestsMock() as m:
+            m.add(method="POST", url=HUB_URL, status=202)
+            result = cli.subscribe_push_notification(
+                channel_id="UCxxxxxx",
+                callback_url="https://example.com/webhook",
+                mode="unsubscribe",
+            )
+            assert result is True
+            assert "hub.mode=unsubscribe" in m.calls[0].request.body
+
+        # sync verify returns True on 204 No Content
+        with responses.RequestsMock() as m:
+            m.add(method="POST", url=HUB_URL, status=204)
+            result = cli.subscribe_push_notification(
+                channel_id="UCxxxxxx",
+                callback_url="https://example.com/webhook",
+                verify="sync",
+            )
+            assert result is True
+
+        # optional params: lease_seconds and secret are included in request body
+        with responses.RequestsMock() as m:
+            m.add(method="POST", url=HUB_URL, status=202)
+            cli.subscribe_push_notification(
+                channel_id="UCxxxxxx",
+                callback_url="https://example.com/webhook",
+                lease_seconds=432000,
+                secret="mysecret",
+            )
+            body = m.calls[0].request.body
+            assert "hub.lease_seconds=432000" in body
+            assert "hub.secret=mysecret" in body
+
+        # hub error raises PyYouTubeException
+        with pytest.raises(PyYouTubeException):
+            with responses.RequestsMock() as m:
+                m.add(
+                    method="POST",
+                    url=HUB_URL,
+                    json={"error": {"code": 400, "message": "bad request"}},
+                    status=400,
+                )
+                cli.subscribe_push_notification(
+                    channel_id="UCxxxxxx",
+                    callback_url="https://example.com/webhook",
+                )