Merge pull request #61 from EchterAlsFake/httpx

Switch requests lib to HTTPX

.github/workflows/tests.yml

@@ -23,7 +23,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip
-        pip install pytest requests ffmpeg-progress-yield
+        pip install pytest httpx[brotli,socks] ffmpeg-progress-yield
     - name: Test with pytest
       run: |
         pytest

pyproject.toml

@@ -13,7 +13,7 @@ classifiers = [
   "Development Status :: 4 - Beta",
   "Programming Language :: Python"
 ]
-dependencies = ["requests", "ffmpeg-progress-yield"]
+dependencies = ["httpx[brotli,socks]", "ffmpeg-progress-yield"]
 requires-python = ">=3.9"
 
 [project.optional-dependencies]

requirements.txt

@@ -1,3 +1,3 @@
-requests>=2.28.1
+httpx[brotli,socks]>=0.27.0
 click>=8.0.4
 ffmpeg-progress-yield>=0.7.8

src/phub/core.py

@@ -6,7 +6,7 @@
 import logging
 import random
 
-import requests
+import httpx
 from typing import Iterable, Union
 from functools import cached_property
 
@@ -94,12 +94,13 @@ def reset(self) -> None:
         '''
 
         # Initialise session
-        self.session = requests.Session()
-        self._clear_granted_token()
+        self.session = httpx.Client(
+            headers = consts.HEADERS,
+            cookies = consts.COOKIES,
+            follow_redirects = True
+        )
 
-        # Insert cookies & headers
-        self.session.headers = consts.HEADERS
-        self.session.cookies.update(consts.COOKIES)
+        self._clear_granted_token()
 
         if self.bypass_geo_blocking:
             ip = random.choice(consts.GEO_BYPASS_IPs)
@@ -120,7 +121,7 @@ def call(self,
              headers: dict = None,
              timeout: float = consts.CALL_TIMEOUT,
              throw: bool = True,
-             silent: bool = False) -> requests.Response:
+             silent: bool = False) -> httpx.Response:
         '''
         Used internally to send a request or an API call.
 
@@ -167,7 +168,7 @@ def call(self,
                     url = url,
                     headers = headers,
                     data = data,
-                    timeout = timeout,
+                    timeout = timeout
                 )
 
                 # Silent 429 errors

src/phub/modules/download.py

@@ -2,7 +2,6 @@
 
 import time
 import logging
-import requests.adapters
 from pathlib import Path
 from ffmpeg_progress_yield import FfmpegProgress
 from typing import TYPE_CHECKING, Callable, Union
@@ -50,7 +49,7 @@ def default(video: Video,
             try:
                 segment = video.client.call(url, throw = False, timeout = 4, silent = True)
 
-                if segment.ok:
+                if segment.is_success:
                     buffer += segment.content
                     callback(i + 1, length)
                     break
@@ -122,7 +121,7 @@ def _thread(client: Client, url: str, timeout: int) -> bytes:
     '''
     try:
         response = client.call(url, timeout=timeout, silent=True)
-        response.raise_for_status()  # Assuming client.call returns an object with a similar interface to requests.Response
+        response.raise_for_status()
         return (url, response.content, True)
 
     except Exception as e:
@@ -138,11 +137,7 @@ def _base_threaded(client: Client, segments: list[str], callback: CallbackType,
     logging.info('Threaded download initiated')
     buffer = {}
     length = len(segments)
-    logger.info('Mounting download adapter')
-    old_adapter = client.session.adapters.get('https://')
-    adapter = requests.adapters.HTTPAdapter(pool_maxsize = max_workers)
-    client.session.mount('https://', adapter)
-
+
     with Pool(max_workers=max_workers) as executor:
         future_to_url = {executor.submit(_thread, client, url, timeout): url for url in segments}
 
@@ -157,7 +152,6 @@ def _base_threaded(client: Client, segments: list[str], callback: CallbackType,
             except Exception as e:
                 logging.warning(f"Error processing segment {url}: {e}")
 
-    client.session.mount('https://', old_adapter)
     return buffer
 
 def threaded(max_workers: int = 20,

src/phub/objects/image.py

@@ -37,12 +37,12 @@ def __init__(self,
         self.url = url
         self.name = name
         self.client = client
-        self._servers = servers
+        self._servers = servers or []
 
         logger.debug('Generated new image object: %s', self)
 
         # Check server image sizes
-        sizes = [s.get('size') for s in servers]
+        sizes = [s.get('size') for s in self._servers]
 
         if len(set(sizes)) > 1:
             logger.warning('Detected different image sizes on alt servers: %s', sizes)

src/phub/objects/user.py

@@ -91,7 +91,7 @@ def dictify(self,
                                               'bio', 'info', 'avatar'], recursive)
 
     @classmethod
-    def from_video(cls, video: Video) -> Self:
+    def from_video(cls, video: Video) -> 'User':
         '''
         Find the author of a video.
         
@@ -113,7 +113,7 @@ def from_video(cls, video: Video) -> Self:
                    url = utils.concat(consts.HOST, guess[0]))
 
     @classmethod
-    def get(cls, client: Client, user: str) -> Self:
+    def get(cls, client: Client, user: str) -> 'User':
         '''
         Fetch a user knowing its name or URL.
         Note - Using only a username makes the fetch between

src/phub/tests/test_auth.py

@@ -0,0 +1,24 @@
+import os
+
+try:
+    from phub import Client
+
+except (ModuleNotFoundError, ImportError):
+    from ...phub import Client
+
+client = Client(
+    email = os.getenv('EMAIL'),
+    password = os.getenv('PASSWORD'),
+    login = False
+)
+
+def test_auth():
+
+    # client.login()
+
+    # assert client.account
+
+    pass
+
+
+# EOF

src/phub/tests/test_model.py

@@ -34,3 +34,5 @@ def test_uploads():
     if len(total_uploads) >= 1:
         for upload in model.uploads:
             assert isinstance(upload.title, str) and len(upload.title) > 3
+
+# EOF

src/phub/tests/test_playlist.py

@@ -26,4 +26,6 @@ def test_playlist_objects():
     assert isinstance(playlist.hidden_videos_amount, int)
     assert isinstance(playlist.author, User)
     assert isinstance(playlist.title, str) and len(playlist.title) >= 1
-    assert isinstance(playlist.tags, list)
+    assert isinstance(playlist.tags, list)
+
+# EOF

src/phub/tests/test_search.py

@@ -34,3 +34,5 @@ def test_user_search():
 
         if idx == 5:
             break
+
+# EOF

src/phub/tests/test_video.py

@@ -32,7 +32,7 @@ def test_video_information():
     assert isinstance(duration.seconds, int) and len(str(duration)) >= 3
     assert isinstance(embed, str) and len(embed) >= 3
     assert isinstance(image, str) and len(image) >= 3
-    assert isinstance(id, str)
+    assert isinstance(id, (str, int))
 
 
 def test_video_segments():
@@ -44,7 +44,4 @@ def test_video_segments():
     assert isinstance(segments_2, list) and len(segments_2) >= 5
     assert isinstance(segments_3, list) and len(segments_3) >= 5
 
-
-
-
-
+# EOF

src/phub/utils.py

@@ -3,6 +3,7 @@
 '''
 
 import math
+import httpx
 import logging
 from typing import Generator, Iterable, Iterator, Union
 
@@ -234,10 +235,10 @@ def head(client: object, url: str) -> Union[str, bool]:
         str | bool: The redirect URL if success, False otherwise.
     '''
 
-    res = client.call(url, 'HEAD', throw = False, silent = True)
+    res: httpx.Response = client.call(url, 'HEAD', throw = False, silent = True)
 
     # Make sure we were not redirected
-    if res.ok and res.url.endswith(url):
+    if res.is_success and not res.has_redirect_location:
         return res.url
     return False