Merge pull request jdepoix#468 from jdepoix/feature/ipblocked-exception-on-429

Added IpBlocked exception when status 429 is returned

Author: jdepoix

youtube_transcript_api/_api.py

@@ -2,6 +2,8 @@
 from typing import Optional, Iterable
 
 from requests import Session
+from requests.adapters import HTTPAdapter
+from urllib3 import Retry
 
 from .proxies import ProxyConfig, GenericProxyConfig
 
@@ -38,6 +40,13 @@ def __init__(
             http_client.proxies = proxy_config.to_requests_dict()
             if proxy_config.prevent_keeping_connections_alive:
                 http_client.headers.update({"Connection": "close"})
+            if proxy_config.retries_when_blocked > 0:
+                retry_config = Retry(
+                    total=proxy_config.retries_when_blocked,
+                    status_forcelist=[429],
+                )
+                http_client.mount("http://", HTTPAdapter(max_retries=retry_config))
+                http_client.mount("https://", HTTPAdapter(max_retries=retry_config))
         self._fetcher = TranscriptListFetcher(http_client, proxy_config=proxy_config)
 
     def fetch(

youtube_transcript_api/_errors.py

@@ -152,8 +152,9 @@ class RequestBlocked(CouldNotRetrieveTranscript):
     )
     WITH_GENERIC_PROXY_CAUSE_MESSAGE = (
         "YouTube is blocking your requests, despite you using proxies. Keep in mind "
-        "a proxy is just a way to hide your real IP behind the IP of that proxy, but "
-        "there is no guarantee that the IP of that proxy won't be blocked as well.\n\n"
+        "that a proxy is just a way to hide your real IP behind the IP of that proxy, "
+        "but there is no guarantee that the IP of that proxy won't be blocked as "
+        "well.\n\n"
         "The only truly reliable way to prevent IP blocks is rotating through a large "
         "pool of residential IPs, by using a provider like Webshare "
         "(https://www.webshare.io/?referral_code=w0xno53eb50g), which provides you "

youtube_transcript_api/_transcripts.py

@@ -92,6 +92,8 @@ class _PlayabilityFailedReason(str, Enum):
 
 def _raise_http_errors(response: Response, video_id: str) -> Response:
     try:
+        if response.status_code == 429:
+            raise IpBlocked(video_id)
         response.raise_for_status()
         return response
     except HTTPError as error:

youtube_transcript_api/test/test_api.py

@@ -272,6 +272,18 @@ def test_get_transcript__exception_if_youtube_request_limit_reached(
         with self.assertRaises(IpBlocked):
             YouTubeTranscriptApi().fetch("abc")
 
+    def test_get_transcript__exception_if_timedtext_request_limit_reached(
+        self,
+    ):
+        httpretty.register_uri(
+            httpretty.GET,
+            "https://www.youtube.com/api/timedtext",
+            status=429,
+        )
+
+        with self.assertRaises(IpBlocked):
+            YouTubeTranscriptApi().fetch("abc")
+
     def test_fetch__exception_if_age_restricted(self):
         httpretty.register_uri(
             httpretty.POST,