Switch from to

SECURITY.md

@@ -4,8 +4,8 @@
 
 | Version  | Supported          |
 | -------- | ------------------ |
-| 1.0a10   | :white_check_mark: |
-| < 1.0a10 | :x:                |
+| 1.4      | :white_check_mark: |
+| < 1.4    | :x:                |
 
 ## Reporting a Vulnerability
 

pyproject.toml

@@ -30,7 +30,7 @@ classifiers = [
 keywords = ["Python", "pCloud", "REST"]
 
 dependencies = [
-    "httpx",
+    "httpx>=0.28.1",
     "setuptools>=75.8.0",
     "wheel>=0.45.1",
 ]

src/pcloud/api.py

@@ -1,5 +1,5 @@
 import os
-import requests
+import httpx
 import zipfile
 
 from hashlib import sha1
@@ -113,7 +113,7 @@ def oauth2_authorize(
         params = {"client_id": client_id, "client_secret": client_secret, "code": code}
         endpoint = ep.get(hostname)
         endpoint_url = PyCloud.endpoints.get(endpoint).endpoint
-        resp = requests.get(endpoint_url + "oauth2_token", params=params).json()
+        resp = httpx.get(endpoint_url + "oauth2_token", params=params).json()
         access_token = resp.get("access_token")
         return cls("", access_token, endpoint, token_expire, oauth2=True)
 

src/pcloud/dummyprotocol.py

@@ -1,7 +1,3 @@
-import requests
-
-from pcloud.utils import log
-from requests_toolbelt.multipart.encoder import MultipartEncoder
 from pcloud.jsonprotocol import PCloudJSONConnection
 
 

src/pcloud/jsonprotocol.py

@@ -1,7 +1,6 @@
-import requests
+import httpx
 
 from pcloud.utils import log
-from requests_toolbelt.multipart.encoder import MultipartEncoder
 
 
 class PCloudJSONConnection(object):
@@ -11,7 +10,7 @@ class PCloudJSONConnection(object):
 
     def __init__(self, api):
         """Connect to pcloud API based on their JSON protocol."""
-        self.session = requests.Session()
+        self.session = httpx.Client()
         self.api = api
 
     def connect(self):
@@ -32,8 +31,8 @@ def do_get_request(self, method, authenticate=True, json=True, endpoint=None, **
         if "use_session" in kw:
             get_method = self.session.get
         else:
-            get_method = requests.get
-        resp = get_method(endpoint + method, params=params, allow_redirects=False)
+            get_method = httpx.get
+        resp = get_method(endpoint + method, params=params)
         resp.raise_for_status()
         if json:
             result = resp.json()
@@ -47,10 +46,7 @@ def upload(self, method, files, **kwargs):
             kwargs["auth"] = self.api.auth_token
         elif self.api.access_token:  # OAuth2 authentication
             kwargs["access_token"] = self.api.access_token
-        fields = list(kwargs.items())
-        fields.extend(files)
-        m = MultipartEncoder(fields=fields)
-        resp = requests.post(
-            self.api.endpoint + method, data=m, headers={"Content-Type": m.content_type}
-        )
+        log.debug(f"Upload files: {files}")
+        log.debug(f"Upload fields: {kwargs}")
+        resp = httpx.post(self.api.endpoint + method, data=kwargs, files=files)
         return resp.json()

src/pcloud/tests/test_api.py

@@ -6,7 +6,7 @@
 import json
 import os.path
 import pytest
-import requests
+import httpx
 
 
 class NoOpSession(object):
@@ -123,6 +123,6 @@ def test_getpublinkdownload(self):
 
     def test_server_security(self):
         papi = DummyPyCloud("", "")
-        resp = requests.get(papi.endpoint + "../../bogus.sh", params={})
+        resp = httpx.get(papi.endpoint + "../../bogus.sh", params={})
         assert resp.content == b'{"Error": "Path not found or not accessible!"}'
         assert resp.status_code == 404