Refactorisation : utilise le Network Requests Manager de QGIS à la place de requests

qtribu/logic/json_feed.py

@@ -1,13 +1,27 @@
+#! python3  # noqa: E265
+
+"""
+    JSON Feed wrapper.
+"""
+
+# ############################################################################
+# ########## Imports ###############
+# ##################################
+
+
+import json
 from datetime import datetime
 from typing import Any, List, Optional
 
-import requests
-from requests import Response
+# 3rd party
+from qgis.PyQt.QtCore import QByteArray
 
+# plugin
 from qtribu.__about__ import __title__, __version__
 from qtribu.logic import RssItem
-from qtribu.toolbelt import PlgLogger, PlgOptionsManager
+from qtribu.toolbelt import NetworkRequestsManager, PlgLogger, PlgOptionsManager
 
+# -- GLOBALS --
 HEADERS: dict = {
     b"Accept": b"application/json",
     b"User-Agent": bytes(f"{__title__}/{__version__}", "utf8"),
@@ -16,9 +30,12 @@
 FETCH_UPDATE_INTERVAL_SECONDS = 7200
 
 
+## -- CLASSES --
+
+
 class JsonFeedClient:
     """
-    Class representing a Geotribu's JSON feed client
+    Class representing a client for JSON feed built with Mkdocs website with RSS plugin.
     """
 
     items: Optional[List[RssItem]] = None
@@ -27,35 +44,48 @@ class JsonFeedClient:
     def __init__(
         self, url: str = PlgOptionsManager.get_plg_settings().json_feed_source
     ):
-        """Class initialization."""
+        """Class initialization.
+
+        :param url: JSON Feed URL, defaults to PlgOptionsManager.get_plg_settings().json_feed_source
+        :type url: str, optional
+        """
         self.log = PlgLogger().log
         self.url = url
+        self.qntwk = NetworkRequestsManager()
 
     def fetch(self, query: str = "") -> list[RssItem]:
-        """
-        Fetch RSS feed items using JSON Feed
+        """Fetch RSS feed items using JSON Feed
 
-        :param query: filter to look for items matching this query
-        :type query: str
+        :param query: filter to look for items matching this query, defaults to ""
+        :type query: str, optional
 
         :return: list of RssItem objects matching the query filter
+        :rtype: list[RssItem]
         """
         if not self.items or (
             self.last_fetch_date
             and (datetime.now() - self.last_fetch_date).total_seconds()
             > FETCH_UPDATE_INTERVAL_SECONDS
         ):
-            r: Response = requests.get(self.url, headers=HEADERS)
-            r.raise_for_status()
-            self.items = [self._map_item(i) for i in r.json()["items"]]
+
+            response: QByteArray = self.qntwk.get_from_source(
+                headers=HEADERS,
+                url=self.url,
+                response_expected_content_type="application/json; charset=utf-8",
+            )
+
+            self.items = [
+                self._map_item(i) for i in json.loads(str(response, "UTF8"))["items"]
+            ]
             self.last_fetch_date = datetime.now()
+
         return [i for i in self.items if self._matches(query, i)]
 
     def authors(self) -> list[str]:
-        """
-        Get a list of authors available in the RSS feed
+        """Get a list of authors available in the RSS feed
 
         :return: list of authors
+        :rtype: list[str]
         """
         authors = []
         for content in self.fetch():
@@ -64,10 +94,10 @@ def authors(self) -> list[str]:
         return sorted(set(authors))
 
     def categories(self) -> list[str]:
-        """
-        Get a list of all categories available in the RSS feed
+        """Get a list of all categories available in the RSS feed.
 
         :return: list of categories available in the RSS feed
+        :rtype: list[str]
         """
         tags = []
         for content in self.fetch():
@@ -76,8 +106,7 @@ def categories(self) -> list[str]:
 
     @staticmethod
     def _map_item(item: dict[str, Any]) -> RssItem:
-        """
-        Map raw JSON object coming from JSON feed to an RssItem object
+        """Map raw JSON object coming from JSON feed to an RssItem object.
 
         :param item: raw JSON object
         :type item: dict[str, Any]
@@ -99,8 +128,7 @@ def _map_item(item: dict[str, Any]) -> RssItem:
 
     @staticmethod
     def _matches(query: str, item: RssItem) -> bool:
-        """
-        Check if item matches given query
+        """Check if item matches given query.
 
         :param query: filter to look for items matching this query
         :type query: str

qtribu/plugin_main.py

@@ -214,7 +214,12 @@ def post_ui_init(self):
         """
         try:
             qntwk = NetworkRequestsManager()
-            self.rss_rdr.read_feed(qntwk.get_from_source(headers=self.rss_rdr.HEADERS))
+            rss_feed_content = qntwk.get_from_source(
+                headers=self.rss_rdr.HEADERS,
+                response_expected_content_type="application/xml",
+            )
+
+            self.rss_rdr.read_feed(rss_feed_content)
             if not self.rss_rdr.latest_item:
                 raise Exception("No item found")
 

qtribu/toolbelt/log_handler.py

@@ -128,10 +128,8 @@ def log(
             # QGIS or custom dialog
             if parent_location and isinstance(parent_location, QWidget):
                 msg_bar = parent_location.findChild(QgsMessageBar)
-                print(msg_bar)
 
             if not msg_bar:
-                print("use QGIS message bar as fallback")
                 msg_bar = iface.messageBar()
 
             # calc duration

qtribu/toolbelt/network_manager.py

@@ -109,7 +109,12 @@ def build_request(self, url: Optional[QUrl] = None) -> QNetworkRequest:
 
         return qreq
 
-    def get_from_source(self, headers: dict = None) -> QByteArray:
+    def get_from_source(
+        self,
+        url: Optional[str] = None,
+        headers: Optional[dict] = None,
+        response_expected_content_type: str = "application/xml",
+    ) -> Optional[QByteArray]:
         """Method to retrieve a RSS feed from a referenced source in preferences. \
         Use cache.
 
@@ -126,7 +131,10 @@ def get_from_source(self, headers: dict = None) -> QByteArray:
             import json
             response_as_dict = json.loads(str(response, "UTF8"))
         """
-        url = self.build_url(PlgOptionsManager.get_plg_settings().rss_source)
+        if not url:
+            url = self.build_url(PlgOptionsManager.get_plg_settings().rss_source)
+        else:
+            url = self.build_url(url)
 
         try:
             # prepare request
@@ -155,23 +163,25 @@ def get_from_source(self, headers: dict = None) -> QByteArray:
             self.log(
                 message=f"Request to {url} succeeded.",
                 log_level=3,
-                push=0,
+                push=False,
             )
 
             req_reply = self.ntwk_requester.reply()
-            if not req_reply.rawHeader(b"Content-Type") == "application/xml":
+            if (
+                not req_reply.rawHeader(b"Content-Type")
+                == response_expected_content_type
+            ):
                 raise TypeError(
-                    "Response mime-type is '{}' not 'application/xml' as required.".format(
-                        req_reply.rawHeader(b"Content-type")
-                    )
+                    f"Response mime-type is '{req_reply.rawHeader(b'Content-type')}' "
+                    f"not '{response_expected_content_type}' as required.".format()
                 )
 
             return req_reply.content()
 
         except Exception as err:
             err_msg = f"Houston, we've got a problem: {err}"
             logger.error(err_msg)
-            self.log(message=err_msg, log_level=2, push=1)
+            self.log(message=err_msg, log_level=2, push=True)
 
     def download_file(self, remote_url: str, local_path: str) -> str:
         """Download a file from a remote web server accessible through HTTP.