Refacto: use requests to download files (#367)

qgis_deployment_toolbelt/commands/upgrade.py

@@ -12,17 +12,16 @@
 
 # Standard library
 import argparse
-import json
 import logging
 import sys
 from collections.abc import Iterable
 from os import getenv
 from pathlib import Path
 from sys import platform as opersys
 from urllib.parse import urlsplit, urlunsplit
-from urllib.request import Request
 
 # 3rd party library
+import requests
 from packaging.version import Version
 
 # submodules
@@ -39,7 +38,7 @@
     exit_cli_success,
 )
 from qgis_deployment_toolbelt.utils.file_downloader import download_remote_file_to_local
-from qgis_deployment_toolbelt.utils.proxies import get_proxy_handler
+from qgis_deployment_toolbelt.utils.proxies import get_proxy_settings
 from qgis_deployment_toolbelt.utils.str2bool import str2bool
 
 # #############################################################################
@@ -108,13 +107,13 @@ def get_latest_release(api_repo_url: str) -> dict | None:
         )
         headers["Authorization"] = f"Bearer {getenv('GITHUB_TOKEN')}"
 
-    request = Request(url=request_url, headers=headers)
-
     try:
         release_info = None
-        with get_proxy_handler().open(request) as response:
-            if response.status == 200:
-                release_info = json.loads(response.read())
+        req = requests.get(
+            url=request_url, headers=headers, proxies=get_proxy_settings()
+        )
+        req.raise_for_status()
+        release_info = req.json()
         return release_info
     except Exception as err:
         logger.error(err)

qgis_deployment_toolbelt/utils/file_downloader.py

@@ -8,12 +8,14 @@
 # standard library
 import logging
 from pathlib import Path
-from urllib.error import HTTPError, URLError
-from urllib.request import Request
+
+# 3rd party
+from requests import Session
+from requests.exceptions import ConnectionError, HTTPError
 
 # package
 from qgis_deployment_toolbelt.__about__ import __title_clean__, __version__
-from qgis_deployment_toolbelt.utils.proxies import get_proxy_handler
+from qgis_deployment_toolbelt.utils.proxies import get_proxy_settings
 
 # ############################################################################
 # ########## GLOBALS #############
@@ -34,6 +36,7 @@ def download_remote_file_to_local(
     user_agent: str = f"{__title_clean__}/{__version__}",
     content_type: str | None = None,
     chunk_size: int = 8192,
+    timeout=(800, 800),
 ) -> Path:
     """Check if the local index file exists. If not, download the search index from \
         remote URL. If it does exist, check if it has been modified.
@@ -51,7 +54,7 @@ def download_remote_file_to_local(
     """
     # check if file exists
     if local_file_path.exists():
-        logger.warning(f"{local_file_path} already exists. It's about to be replaced.")
+        logger.info(f"{local_file_path} already exists. It's about to be replaced.")
         local_file_path.unlink(missing_ok=True)
 
     # make sure parents folder exist
@@ -62,37 +65,33 @@ def download_remote_file_to_local(
     if content_type:
         headers["Accept"] = content_type
 
-    # download the remote file into local file
-    custom_request = Request(url=remote_url_to_download, headers=headers)
-
     try:
-        with get_proxy_handler().open(custom_request) as response, local_file_path.open(
-            mode="wb"
-        ) as buffile:
-            while True:
-                chunk = response.read(chunk_size)
-                if not chunk:
-                    break
-                buffile.write(chunk)
-        logger.info(
-            f"Downloading {remote_url_to_download} to {local_file_path} succeeded."
-        )
+        with Session() as dl_session:
+            dl_session.proxies.update(get_proxy_settings())
+            dl_session.headers.update(headers)
+
+            with dl_session.get(
+                url=remote_url_to_download, stream=True, timeout=timeout
+            ) as req:
+                req.raise_for_status()
+
+                with local_file_path.open(mode="wb") as buffile:
+                    for chunk in req.iter_content(chunk_size=chunk_size):
+                        if chunk:
+                            buffile.write(chunk)
+            logger.info(
+                f"Downloading {remote_url_to_download} to {local_file_path} succeeded."
+            )
     except HTTPError as error:
         logger.error(
             f"Downloading {remote_url_to_download} to {local_file_path} failed. "
             f"Cause: HTTPError. Trace: {error}"
         )
         raise error
-    except URLError as error:
-        logger.error(
-            f"Downloading {remote_url_to_download} to {local_file_path} failed. "
-            f"Cause: URLError. Trace: {error}"
-        )
-        raise error
-    except TimeoutError as error:
+    except ConnectionError as error:
         logger.error(
             f"Downloading {remote_url_to_download} to {local_file_path} failed. "
-            f"Cause: TimeoutError. Trace: {error}"
+            f"Cause: ConnectionError. Trace: {error}"
         )
         raise error
     except Exception as error:

qgis_deployment_toolbelt/utils/proxies.py

@@ -49,13 +49,14 @@ def get_proxy_handler() -> OpenerDirector:
 
 
 @lru_cache
-def get_proxy_settings() -> dict | None:
+def get_proxy_settings() -> dict:
     """Retrieves network proxy settings from operating system configuration or
     environment variables.
 
     Returns:
-        dict | None: system proxy settings or None if no proxy is set
+        dict: proxy settings with protocl as key and URL as value
     """
+    proxy_settings = {}
     if environ.get("QDT_PROXY_HTTP"):
         proxy_settings = {
             "http": environ.get("QDT_PROXY_HTTP"),
@@ -96,7 +97,6 @@ def get_proxy_settings() -> dict | None:
         logger.debug(f"Proxies settings found in the OS: {proxy_settings}")
     else:
         logger.debug("No proxy settings found in environment vars nor OS settings.")
-        proxy_settings = None
 
     # check scheme and URL validity
     if isinstance(proxy_settings, dict):

tests/test_utils_file_downloader.py

@@ -13,7 +13,9 @@
 # standard library
 import unittest
 from pathlib import Path
-from urllib.error import HTTPError, URLError
+
+# 3rd party
+from requests.exceptions import ConnectionError, HTTPError
 
 # project
 from qgis_deployment_toolbelt.utils.file_downloader import download_remote_file_to_local
@@ -49,7 +51,7 @@ def test_download_file_raise_http_error(self):
     def test_download_file_raise_url_error(self):
         """Test download with a bad URL."""
 
-        with self.assertRaises(URLError):
+        with self.assertRaises(ConnectionError):
             download_remote_file_to_local(
                 remote_url_to_download="https://fake_url/youpi.dmg",
                 local_file_path=Path("README.md"),

tests/test_utils_proxies.py

@@ -28,7 +28,7 @@ class TestUtilsNetworkProxies(unittest.TestCase):
     def test_proxy_settings(self):
         """Test proxy settings retriever."""
         # by default, no proxy
-        self.assertIsNone(get_proxy_settings())
+        self.assertDictEqual(get_proxy_settings(), {})
 
         # using generic - only http
         get_proxy_settings.cache_clear()