Replace http.client with HTTPX

setup.cfg

@@ -32,6 +32,7 @@ keywords =
 [options]
 packages = find:
 install_requires =
+    httpx
     importlib-metadata;python_version < '3.8'
 python_requires = >=3.7
 package_dir = =src

src/pylast/__init__.py

@@ -30,10 +30,11 @@ import ssl
 import tempfile
 import time
 import xml.dom
-from http.client import HTTPSConnection
 from urllib.parse import quote_plus
 from xml.dom import Node, minidom
 
+import httpx
+
 try:
     # Python 3.8+
     import importlib.metadata as importlib_metadata
@@ -125,6 +126,12 @@ DELAY_TIME = 0.2
 # Python >3.4 has sane defaults
 SSL_CONTEXT = ssl.create_default_context()
 
+HEADERS = {
+    "Content-type": "application/x-www-form-urlencoded",
+    "Accept-Charset": "utf-8",
+    "User-Agent": f"pylast/{__version__}",
+}
+
 logger = logging.getLogger(__name__)
 logging.getLogger(__name__).addHandler(logging.NullHandler())
 
@@ -390,7 +397,7 @@ class _Network:
     def enable_proxy(self, host, port):
         """Enable a default web proxy"""
 
-        self.proxy = [host, _number(port)]
+        self.proxy = f"{host}:{_number(port)}"
         self.proxy_enabled = True
 
     def disable_proxy(self):
@@ -906,68 +913,41 @@ class _Request:
             self.network._delay_call()
 
         username = self.params.pop("username", None)
-        username = f"?username={username}" if username is not None else ""
-
-        data = []
-        for name in self.params.keys():
-            data.append("=".join((name, quote_plus(_string(self.params[name])))))
-        data = "&".join(data)
-
-        headers = {
-            "Content-type": "application/x-www-form-urlencoded",
-            "Accept-Charset": "utf-8",
-            "User-Agent": "pylast/" + __version__,
-        }
+        username = "" if username is None else f"?username={username}"
 
         (host_name, host_subdir) = self.network.ws_server
 
         if self.network.is_proxy_enabled():
-            conn = HTTPSConnection(
-                context=SSL_CONTEXT,
-                host=self.network._get_proxy()[0],
-                port=self.network._get_proxy()[1],
+            client = httpx.Client(
+                verify=SSL_CONTEXT,
+                base_url=f"https://{host_name}",
+                headers=HEADERS,
+                proxies=f"http://{self.network._get_proxy()}",
+            )
+        else:
+            client = httpx.Client(
+                verify=SSL_CONTEXT,
+                base_url=f"https://{host_name}",
+                headers=HEADERS,
             )
 
-            try:
-                conn.request(
-                    method="POST",
-                    url=f"https://{host_name}{host_subdir}{username}",
-                    body=data,
-                    headers=headers,
-                )
-            except Exception as e:
-                raise NetworkError(self.network, e) from e
-
-        else:
-            conn = HTTPSConnection(context=SSL_CONTEXT, host=host_name)
-
-            try:
-                conn.request(
-                    method="POST",
-                    url=f"{host_subdir}{username}",
-                    body=data,
-                    headers=headers,
-                )
-            except Exception as e:
-                raise NetworkError(self.network, e) from e
-
         try:
-            response = conn.getresponse()
-            if response.status in [500, 502, 503, 504]:
-                raise WSError(
-                    self.network,
-                    response.status,
-                    "Connection to the API failed with HTTP code "
-                    + str(response.status),
-                )
-            response_text = _unicode(response.read())
+            response = client.post(f"{host_subdir}{username}", data=self.params)
         except Exception as e:
-            raise MalformedResponseError(self.network, e) from e
+            raise NetworkError(self.network, e) from e
+
+        if response.status_code in (500, 502, 503, 504):
+            raise WSError(
+                self.network,
+                response.status_code,
+                f"Connection to the API failed with HTTP code {response.status_code}",
+            )
+        response_text = _unicode(response.read())
 
         try:
             self._check_response_for_errors(response_text)
         finally:
-            conn.close()
+            client.close()
         return response_text
 
     def execute(self, cacheable: bool = False) -> xml.dom.minidom.Document:

tests/test_network.py

@@ -1,4 +1,3 @@
-#!/usr/bin/env python
 """
 Integration (not unit) tests for pylast.py
 """
@@ -297,13 +296,13 @@ class TestPyLastNetwork(TestPyLastWithLastFm):
 
     def test_proxy(self):
         # Arrange
-        host = "https://example.com"
+        host = "example.com"
         port = 1234
 
         # Act / Assert
         self.network.enable_proxy(host, port)
         assert self.network.is_proxy_enabled()
-        assert self.network._get_proxy() == ["https://example.com", 1234]
+        assert self.network._get_proxy() == "example.com:1234"
 
         self.network.disable_proxy()
         assert not self.network.is_proxy_enabled()