replace Requests http client by Niquests

.github/workflows/lint.yml

@@ -28,4 +28,5 @@ jobs:
         with:
           python-version: "3.9"
       - run: pip install mypy==1.10.0
-      - run:  mypy --install-types --non-interactive
+      - run: pip install .
+      - run: mypy --install-types --non-interactive

pyproject.toml

@@ -9,7 +9,7 @@ classifiers = [
     "License :: OSI Approved :: MIT License",
 ]
 dependencies = [
-    "requests >= 2.22",
+    "niquests >= 3, <4",
 ]
 dynamic = ["version", "readme"]
 
@@ -78,6 +78,5 @@ dev = [
     "mypy>=1.8.0",
     "pytest>=7.4.4",
     "pytest-cov>=4.1.0",
-    "types-requests>=2.31.0.20240218",
     "pytest-retry>=1.6.3",
 ]

tests/auth/test_oauth.py

@@ -6,7 +6,7 @@
 from unittest import mock
 
 import pytest
-from requests import Response
+from niquests import Response
 
 from ytmusicapi.auth.oauth import OAuthToken
 from ytmusicapi.auth.types import AuthType
@@ -37,8 +37,8 @@ def fixture_yt_alt_oauth(browser_filepath: str, alt_oauth_credentials: OAuthCred
 
 
 class TestOAuth:
-    @mock.patch("requests.Response.json")
-    @mock.patch("requests.Session.post")
+    @mock.patch("niquests.Response.json")
+    @mock.patch("niquests.Session.post")
     def test_setup_oauth(self, session_mock, json_mock, blank_code, config):
         session_mock.return_value = Response()
         token_code = json.loads(config["auth"]["oauth_token"])

tests/mixins/test_uploads.py

@@ -86,7 +86,7 @@ def test_upload_song_and_verify(self, config, yt_auth: YTMusic):
             upload_response = yt_auth.upload_song(get_resource(config["uploads"]["file"]))
 
         assert (
-            upload_response == ResponseStatus.SUCCEEDED or upload_response.status_code == 200
+            upload_response == ResponseStatus.SUCCEEDED or upload_response.status_code == 200  # type: ignore[union-attr]
         ), f"Song failed to upload {upload_response}"
 
         # Wait for upload to finish processing and verify it can be retrieved

ytmusicapi/auth/browser.py

@@ -1,7 +1,7 @@
 import platform
 from typing import Optional
 
-from requests.structures import CaseInsensitiveDict
+from niquests.structures import CaseInsensitiveDict
 
 from ytmusicapi.exceptions import YTMusicError, YTMusicUserError
 from ytmusicapi.helpers import *

ytmusicapi/auth/oauth/credentials.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass
 from typing import Optional
 
-import requests
+import niquests as requests
 
 from ytmusicapi.constants import (
     OAUTH_CLIENT_ID,

ytmusicapi/auth/oauth/token.py

@@ -1,12 +1,11 @@
 import json
 import time
+import typing
 import webbrowser
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Optional
 
-from requests.structures import CaseInsensitiveDict
-
 from ytmusicapi.auth.oauth.credentials import Credentials
 from ytmusicapi.auth.oauth.models import BaseTokenDict, Bearer, DefaultScope, RefreshableTokenDict
 
@@ -51,7 +50,7 @@ class OAuthToken(Token):
     """Wrapper for an OAuth token implementing expiration methods."""
 
     @staticmethod
-    def is_oauth(headers: CaseInsensitiveDict) -> bool:
+    def is_oauth(headers: dict[str, typing.Any]) -> bool:
         return all(key in headers for key in Token.members())
 
     def update(self, fresh_access: BaseTokenDict):

ytmusicapi/mixins/_protocol.py

@@ -2,7 +2,7 @@
 
 from typing import Optional, Protocol
 
-from requests import Response
+from niquests import Response
 
 from ytmusicapi.auth.types import AuthType
 from ytmusicapi.parsers.i18n import Parser

ytmusicapi/mixins/browsing.py

@@ -477,7 +477,15 @@ def get_album_browse_id(self, audioPlaylistId: str) -> Optional[str]:
         with warnings.catch_warnings():
             # merge this with statement with catch_warnings on Python>=3.11
             warnings.simplefilter(action="ignore", category=DeprecationWarning)
-            decoded = response.text.encode("utf8").decode("unicode_escape")
+
+            decoded_body = response.text
+
+            # the server sent nothing or content cannot be decoded
+            if decoded_body is None:
+                # this is unlikely. only defensive.
+                raise YTMusicError("Unable to read response body (album_browse_id)")
+
+            decoded = decoded_body.encode("utf8").decode("unicode_escape")
 
         matches = re.search(r"\"MPRE.+?\"", decoded)
         browse_id = None
@@ -890,6 +898,10 @@ def get_signatureTimestamp(self, url: Optional[str] = None) -> int:
         if url is None:
             url = self.get_basejs_url()
         response = self._send_get_request(url=url)
+
+        if response.text is None:
+            raise YTMusicError("Unable to decode response body as text for signatureTimestamp")
+
         match = re.search(r"signatureTimestamp[:=](\d+)", response.text)
         if match is None:
             raise YTMusicError("Unable to identify the signatureTimestamp.")

ytmusicapi/mixins/uploads.py

@@ -1,7 +1,7 @@
 from pathlib import Path
 from typing import Optional, Union
 
-import requests
+import niquests as requests
 
 from ytmusicapi.continuations import get_continuations
 from ytmusicapi.helpers import *
@@ -238,7 +238,7 @@ def upload_song(self, filepath: str) -> Union[ResponseStatus, requests.Response]
         response = requests.post(upload_url, data=body, headers=headers, proxies=self.proxies)
         headers["X-Goog-Upload-Command"] = "upload, finalize"
         headers["X-Goog-Upload-Offset"] = "0"
-        upload_url = response.headers["X-Goog-Upload-URL"]
+        upload_url = response.headers["X-Goog-Upload-URL"]  # type: ignore[assignment]
         with open(fp, "rb") as file:
             response = requests.post(upload_url, data=file, headers=headers, proxies=self.proxies)
 

ytmusicapi/setup.py

@@ -3,7 +3,7 @@
 from pathlib import Path
 from typing import Optional
 
-import requests
+import niquests as requests
 
 from ytmusicapi.auth.browser import setup_browser
 from ytmusicapi.auth.oauth import OAuthCredentials, RefreshingToken

ytmusicapi/ytmusic.py

@@ -2,14 +2,14 @@
 import json
 import locale
 import time
+import typing
 from contextlib import suppress
 from functools import partial
 from pathlib import Path
 from typing import Optional, Union
 
-import requests
-from requests import Response
-from requests.structures import CaseInsensitiveDict
+import niquests as requests
+from niquests import Response
 
 from ytmusicapi.helpers import (
     SUPPORTED_LANGUAGES,
@@ -93,9 +93,7 @@ def __init__(
         self._headers = None  #: cache formed headers including auth
 
         self.auth = auth  #: raw auth
-        self._input_dict: CaseInsensitiveDict = (
-            CaseInsensitiveDict()
-        )  #: parsed auth arg value in dictionary format
+        self._input_dict: dict[str, typing.Any] = {}  #: parsed auth arg value in dictionary format
 
         self.auth_type: AuthType = AuthType.UNAUTHORIZED
 
@@ -129,10 +127,9 @@ def __init__(
                         input_json = json.load(json_file)
                 else:
                     input_json = json.loads(auth_str)
-                self._input_dict = CaseInsensitiveDict(input_json)
-
+                self._input_dict = {k.lower(): v for k, v in input_json.items()}
             else:
-                self._input_dict = CaseInsensitiveDict(self.auth)
+                self._input_dict = {k.lower(): v for k, v in self.auth.items()}
 
             if OAuthToken.is_oauth(self._input_dict):
                 self._token = RefreshingToken(
@@ -234,9 +231,15 @@ def _send_request(self, endpoint: str, body: dict, additionalParams: str = "") -
             proxies=self.proxies,
             cookies=self.cookies,
         )
-        response_text = json.loads(response.text)
-        if response.status_code >= 400:
-            message = "Server returned HTTP " + str(response.status_code) + ": " + response.reason + ".\n"
+        response_text = response.json()
+        if response.status_code is None or response.status_code >= 400:
+            message = (
+                "Server returned HTTP "
+                + str(response.status_code)
+                + ": "
+                + (response.reason or "Unspecified")
+                + ".\n"
+            )
             error = response_text.get("error", {}).get("message")
             raise YTMusicServerError(message + error)
         return response_text