Merge pull request #58 from sentinel-hub/fix/pylint_errors

Fix pylint errors

Author: beltran-sinergise

SentinelHub/__init__.py

@@ -30,7 +30,7 @@ def classFactory(iface):
     # pylint: disable=unused-import
 
     # The following initializes UI
-    from . import resources
+    from . import resources  # pylint: disable=import-self,no-name-in-module
     from .utils.meta import ensure_import
 
     ensure_import("defusedxml")

SentinelHub/constants.py

@@ -1,6 +1,7 @@
 """
 Module containing constants
 """
+
 from enum import Enum
 
 from qgis.core import Qgis

SentinelHub/exceptions.py

@@ -1,15 +1,67 @@
 """
 Utilities for handling exceptions and error messaging
 """
+
 import time
 from abc import ABC, abstractmethod
+from defusedxml import ElementTree
+import requests
 
 from qgis.utils import iface
 
+from sentinelhub.client import get_proxy_from_qsettings
 from .constants import ExtentType, MessageType
 from .utils.meta import PLUGIN_NAME
 
 
+def get_error_message(exception):
+    """Creates an error message from the given exception
+
+    :param exception: Exception obtained during download
+    :type exception: requests.RequestException
+    :return: error message
+    :rtype: str
+    """
+    message = f"{exception.__class__.__name__}: "
+
+    if isinstance(exception, (requests.ConnectionError, requests.Timeout)):
+        if isinstance(exception, requests.ConnectionError):
+            message += "Cannot access service, check your internet connection."
+        else:
+            message += "Connection timed out, service is too slow"
+
+        enabled, host, port, _, _ = get_proxy_from_qsettings()
+        if enabled:
+            message += f" QGIS is configured to use proxy: {host}"
+            if port:
+                message += f":{port}"
+
+        return message
+
+    if isinstance(exception, requests.HTTPError):
+        try:
+            server_message = ""
+            for elem in ElementTree.fromstring(exception.response.content):
+                if "ServiceException" in elem.tag:
+                    server_message += elem.text.strip("\n\t ")
+        except ElementTree.ParseError:
+            server_message = exception.response.text.strip("\n\t ")
+
+        server_message = server_message.encode("ascii", errors="ignore").decode("utf-8")
+
+        # Include HTTP status code
+        status_code = exception.response.status_code
+        message += f"HTTP {status_code} - "
+
+        # Provide meaningful message when server response is empty
+        if not server_message:
+            server_message = "No additional details provided by server"
+
+        return message + f"{server_message}"
+
+    return message + str(exception)
+
+
 def show_message(message, message_type):
     """Show message for user
 
@@ -192,7 +244,10 @@ def check(self, plugin):
             return True
 
         return (
-            plugin.settings.lat_min and plugin.settings.lat_max and plugin.settings.lng_min and plugin.settings.lng_max
+            plugin.settings.lat_min
+            and plugin.settings.lat_max
+            and plugin.settings.lng_min
+            and plugin.settings.lng_max
         )
 
 

SentinelHub/metadata.txt

@@ -6,7 +6,7 @@ name=SentinelHub
 qgisMinimumVersion=3.22
 qgisMaximumVersion=4.99
 description=SentinelHub plugin enables users to harness the power of Sentinel Hub services directly from QGIS.
-version=2.0.5
+version=2.0.6
 author=Sinergise
 email=info@sentinel-hub.com
 
@@ -26,6 +26,8 @@ repository=https://github.com/sentinel-hub/sentinelhub-qgis-plugin
 
 # Recommended items:
 changelog=
+    2.0.6
+        - Maintenance: linting fixes
     2.0.5
         - Use defusedxml for parsing XML responses to address security scanner findings
     2.0.4

SentinelHub/sentinelhub/client.py

@@ -2,14 +2,12 @@
 Download client for Sentinel Hub service
 """
 
-from defusedxml import ElementTree
-
 import requests
 import requests.auth
 from qgis.PyQt.QtCore import QSettings
 
 from ..constants import DEFAULT_REQUEST_TIMEOUT
-from ..exceptions import DownloadError
+from ..exceptions import DownloadError, get_error_message
 from ..utils.meta import get_plugin_version
 from .session import Session
 
@@ -38,7 +36,9 @@ def download(self, url, timeout=DEFAULT_REQUEST_TIMEOUT, session_settings=None):
         proxy_dict, auth = get_proxy_config()
         headers = self._prepare_headers(session_settings)
         try:
-            response = requests.get(url, headers=headers, timeout=timeout, proxies=proxy_dict, auth=auth)
+            response = requests.get(
+                url, headers=headers, timeout=timeout, proxies=proxy_dict, auth=auth
+            )
             response.raise_for_status()
         except requests.RequestException as exception:
             raise DownloadError(get_error_message(exception)) from exception
@@ -63,61 +63,15 @@ def _get_session(settings):
             return Client._CACHED_SESSIONS[cache_key]
 
         session = Session(
-            base_url=settings.base_url, client_id=settings.client_id, client_secret=settings.client_secret
+            base_url=settings.base_url,
+            client_id=settings.client_id,
+            client_secret=settings.client_secret,
         )
 
         Client._CACHED_SESSIONS[cache_key] = session
         return session
 
 
-def get_error_message(exception):
-    """Creates an error message from the given exception
-
-    :param exception: Exception obtained during download
-    :type exception: requests.RequestException
-    :return: error message
-    :rtype: str
-    """
-    message = f"{exception.__class__.__name__}: "
-
-    if isinstance(exception, (requests.ConnectionError, requests.Timeout)):
-        if isinstance(exception, requests.ConnectionError):
-            message += "Cannot access service, check your internet connection."
-        else:
-            message += "Connection timed out, service is too slow"
-
-        enabled, host, port, _, _ = get_proxy_from_qsettings()
-        if enabled:
-            message += f" QGIS is configured to use proxy: {host}"
-            if port:
-                message += f":{port}"
-
-        return message
-
-    if isinstance(exception, requests.HTTPError):
-        try:
-            server_message = ""
-            for elem in ElementTree.fromstring(exception.response.content):
-                if "ServiceException" in elem.tag:
-                    server_message += elem.text.strip("\n\t ")
-        except ElementTree.ParseError:
-            server_message = exception.response.text.strip("\n\t ")
-
-        server_message = server_message.encode("ascii", errors="ignore").decode("utf-8")
-
-        # Include HTTP status code
-        status_code = exception.response.status_code
-        message += f"HTTP {status_code} - "
-
-        # Provide meaningful message when server response is empty
-        if not server_message:
-            server_message = "No additional details provided by server"
-
-        return message + f"{server_message}"
-
-    return message + str(exception)
-
-
 def get_proxy_config():
     """Get proxy config from QSettings and builds proxy parameters
 
@@ -132,7 +86,11 @@ def get_proxy_config():
         for protocol in ["http", "https", "ftp"]:
             proxy_dict[protocol] = f"{protocol}://{host}{port_str}"
 
-    auth = requests.auth.HTTPProxyAuth(user, password) if enabled and user and password else None
+    auth = (
+        requests.auth.HTTPProxyAuth(user, password)
+        if enabled and user and password
+        else None
+    )
 
     return proxy_dict, auth
 

SentinelHub/sentinelhub/common.py

@@ -36,7 +36,9 @@ def load(cls, payload):
         elif "datasetSource" in payload:
             data_source_id = payload["datasetSource"]["@id"].rsplit("/", 1)[-1]
         else:
-            raise ValueError("Layer payload missing both 'datasetSourceId' and 'datasetSource' fields")
+            raise ValueError(
+                "Layer payload missing both 'datasetSourceId' and 'datasetSource' fields"
+            )
 
         return cls(
             layer_id=payload["id"],
@@ -52,7 +54,14 @@ def load(cls, payload):
 class DataSource:
     """Stores info about a Sentinel Hub data source"""
 
-    def __init__(self, data_source_type, data_source_id, collection_id=None, name=None, service_url=None):
+    def __init__(
+        self,
+        data_source_type,
+        data_source_id,
+        collection_id=None,
+        name=None,
+        service_url=None,
+    ):  # pylint: disable=too-many-positional-arguments
         self.type = data_source_type
         self.id = int(data_source_id)
         self.collection_id = collection_id

SentinelHub/sentinelhub/ogc.py

@@ -1,6 +1,7 @@
 """
 Module with Sentinel Hub OGC utilities
 """
+
 import datetime as dt
 from urllib.parse import quote_plus, urlencode
 
@@ -92,7 +93,7 @@ def get_wfs_url(settings, layer, bbox_str, time_range, maxcc=None):
     return _build_url(base_url, params)
 
 
-def get_wcs_url(settings, layer, bbox, crs=None):
+def get_wcs_url(settings, bbox, crs=None):
     """Generate a URL for WCS request from parameters"""
     base_url = _get_service_endpoint(settings, ServiceType.WCS)
     params = {
@@ -147,7 +148,9 @@ def _build_uri(base_url, url_params, uri_params, use_builder=False):
 
 def _build_time(settings):
     """Builds a time string to be sent to Sentinel Hub service"""
-    if (settings.is_exact_date and not settings.start_time) or (not settings.start_time and not settings.end_time):
+    if (settings.is_exact_date and not settings.start_time) or (
+        not settings.start_time and not settings.end_time
+    ):
         return ""
 
     start_time = settings.start_time

SentinelHub/sentinelhub/session.py

@@ -10,7 +10,7 @@
 from qgis.core import QgsMessageLog
 from requests_oauthlib import OAuth2Session
 
-from ..exceptions import SessionError
+from ..exceptions import SessionError, get_error_message
 
 
 class Session:
@@ -42,6 +42,7 @@ def __init__(self, base_url, client_id, client_secret):
 
     @staticmethod
     def select_oauth_url(base_url):
+        """Selects the appropriate OAuth URL based on the base URL."""
         if base_url == "https://sh.dataspace.copernicus.eu":
             return "https://identity.dataspace.copernicus.eu/auth/realms/CDSE/protocol/openid-connect/token"
         return f"{base_url}/oauth/token"
@@ -53,7 +54,10 @@ def token(self):
         :return: A token in a form of dictionary of parameters
         :rtype: dict
         """
-        if self._token and self._token["expires_at"] > time.time() + self.SECONDS_BEFORE_EXPIRY:
+        if (
+            self._token
+            and self._token["expires_at"] > time.time() + self.SECONDS_BEFORE_EXPIRY
+        ):
             return self._token
 
         self._token = self._fetch_token()
@@ -72,20 +76,24 @@ def _fetch_token(self):
         """Collects a new token from Sentinel Hub service"""
         oauth_client = BackendApplicationClient(client_id=self.client_id)
 
-        QgsMessageLog.logMessage("Creating a new authentication session with Sentinel Hub service")
+        QgsMessageLog.logMessage(
+            "Creating a new authentication session with Sentinel Hub service"
+        )
 
         try:
             with OAuth2Session(client=oauth_client) as oauth_session:
                 return oauth_session.fetch_token(
-                    token_url=self.oauth_url, client_id=self.client_id, client_secret=self.client_secret
+                    token_url=self.oauth_url,
+                    client_id=self.client_id,
+                    client_secret=self.client_secret,
                 )
         except requests.HTTPError as exception:
-            from ..sentinelhub.client import get_error_message
-
             error_msg = get_error_message(exception)
             raise SessionError(error_msg) from exception
         except OAuth2Error as exception:
             error_details = str(exception)
             if error_details:
-                raise SessionError(f"Authentication failed: {error_details}") from exception
+                raise SessionError(
+                    f"Authentication failed: {error_details}"
+                ) from exception
             raise SessionError() from exception

SentinelHub/sentinelhub/wcs.py

@@ -1,6 +1,7 @@
 """
 Utilities for interacting with Sentinel Hub WCS service
 """
+
 import os
 
 from ..constants import CrsType, ExtentType
@@ -11,9 +12,13 @@
 
 def download_wcs_image(settings, layer, bbox, client):
     """Downloads and saves an image from Sentinel Hub WCS service"""
-    crs = settings.crs if settings.download_extent_type is ExtentType.CURRENT else CrsType.WGS84
+    crs = (
+        settings.crs
+        if settings.download_extent_type is ExtentType.CURRENT
+        else CrsType.WGS84
+    )
     bbox_str = bbox_to_string(bbox, crs)
-    url = get_wcs_url(settings, layer, bbox_str, crs)
+    url = get_wcs_url(settings, bbox_str, crs)
 
     filename = get_filename(settings, layer, bbox_str)
     path = os.path.join(settings.download_folder, filename)

SentinelHub/settings.py

@@ -1,11 +1,19 @@
 """
 Module containing parameters and settings for Sentinel Hub services
 """
+
 import copy
 
 from qgis.PyQt.QtCore import QSettings
 
-from .constants import CrsType, ExtentType, ImageFormat, ImagePriority, ServiceType, TimeType
+from .constants import (
+    CrsType,
+    ExtentType,
+    ImageFormat,
+    ImagePriority,
+    ServiceType,
+    TimeType,
+)
 
 
 class Settings:
@@ -107,4 +115,5 @@ def copy(self):
         return copy.copy(self)
 
     def clear(self):
+        """Clears all settings from the local store and from the instance"""
         self.qsettings.clear()